24 Commits

Author SHA1 Message Date
Joseph Doherty 8efbcdf6c3 merge: T58 scene compression + thread emission on close 2026-04-26 20:21:01 -04:00
Joseph Doherty 8aeadfd0e4 merge: T57 significance-aware retrieval ranking 2026-04-26 20:21:01 -04:00
Joseph Doherty 88350d7d2e merge: T56 event-completion promotion service 2026-04-26 20:21:00 -04:00
Joseph Doherty 343f305587 feat: significance-driven quote retention + thread emission on close (T58) 2026-04-26 20:18:34 -04:00
Joseph Doherty 021587b3df feat: event-completion promotion service (T56) 2026-04-26 20:15:51 -04:00
Joseph Doherty 5e6b29e0c5 feat: significance-aware retrieval ranking (T57) 2026-04-26 20:15:19 -04:00
Joseph Doherty a34931375c merge: T55 thread-detection service 2026-04-26 20:12:12 -04:00
Joseph Doherty 959fe11410 merge: T54 synthesized-memories service 2026-04-26 20:12:12 -04:00
Joseph Doherty 2959e1ac2a merge: T53 skip narration service 2026-04-26 20:12:12 -04:00
Joseph Doherty afe940259a merge: T52 event-lifecycle detection service 2026-04-26 20:12:12 -04:00
Joseph Doherty c2144cd9df feat: skip narration service (T53) 2026-04-26 20:10:42 -04:00
Joseph Doherty 7857da4112 feat: thread-detection service (T55) 2026-04-26 20:10:36 -04:00
Joseph Doherty adbbd32873 feat: synthesized-memories service for jump skips (T54) 2026-04-26 20:10:05 -04:00
Joseph Doherty 98250644ad feat: event-lifecycle detection service (T52) 2026-04-26 20:09:13 -04:00
Joseph Doherty da1f67fb6a test: bump schema_version assertion to 10 (0009 events + 0010 threads) 2026-04-26 20:07:08 -04:00
Joseph Doherty 03ba34272b merge: T51 threads table + projector handlers 2026-04-26 20:06:45 -04:00
Joseph Doherty e26885b011 merge: T50 time_skip event handlers 2026-04-26 20:06:45 -04:00
Joseph Doherty 5b7a195cf5 merge: T49 events table + lifecycle handlers 2026-04-26 20:06:45 -04:00
Joseph Doherty 25bcbac055 feat: threads table + projector handlers (T51) 2026-04-26 20:05:09 -04:00
Joseph Doherty ab2b494c21 feat: time_skip event handlers (T50) 2026-04-26 20:04:46 -04:00
Joseph Doherty b6888ff36a feat: events table + lifecycle handlers (T49) 2026-04-26 20:04:36 -04:00
dohertj2 e4fd888b53 Merge pull request 'Phase 2.5 cleanup: 15-item backlog burndown' (#3) from phase-2.5 into main 2026-04-26 20:00:38 -04:00
dohertj2 079774dce5 Merge pull request 'Phase 2: multi-entity scene support (you + host + guest)' (#2) from phase-2 into main 2026-04-26 20:00:16 -04:00
dohertj2 3be7920f41 Merge pull request 'Phase 1: v1 single-bot roleplay engine' (#1) from phase-1 into main 2026-04-26 19:59:29 -04:00
23 changed files with 2479 additions and 10 deletions
+14
View File
@@ -0,0 +1,14 @@
CREATE TABLE events (
id INTEGER PRIMARY KEY,
event_id TEXT NOT NULL UNIQUE,
chat_id TEXT NOT NULL,
kind TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'planned',
props_json TEXT NOT NULL DEFAULT '{}',
planned_for TEXT,
started_at TEXT,
completed_at TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX events_chat_idx ON events(chat_id, status);
+14
View File
@@ -0,0 +1,14 @@
CREATE TABLE threads (
id INTEGER PRIMARY KEY,
thread_id TEXT NOT NULL UNIQUE,
chat_id TEXT NOT NULL,
title TEXT NOT NULL,
summary TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'open',
opened_at TEXT NOT NULL DEFAULT (datetime('now')),
closed_at TEXT,
last_referenced_scene_id INTEGER,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX threads_chat_status_idx ON threads(chat_id, status);
+72
View File
@@ -0,0 +1,72 @@
"""Event-lifecycle detection (T52).
After each turn, classify whether any active events transitioned
(started, completed, cancelled). Conservative bias — most turns
return empty. T61 turn flow appends one event_started/completed/
cancelled per transition via append_and_apply.
"""
from __future__ import annotations
from pydantic import BaseModel, Field
from chat.llm.classify import classify
from chat.llm.client import LLMClient
class EventTransition(BaseModel):
event_id: str
new_status: str # "active" | "completed" | "cancelled"
reason: str = ""
class EventLifecycleDecision(BaseModel):
transitions: list[EventTransition] = Field(default_factory=list)
_SYSTEM = (
"You decide whether any active events transitioned this turn. "
"STRONGLY default to empty transitions — most turns do NOT resolve "
"or start a known event. Output only transitions that the narrative "
"text clearly resolves or starts. Each transition MUST reference an "
"event_id from the active_events list. new_status is one of "
"'active' (planned -> active), 'completed', or 'cancelled'. "
"Output strict JSON matching the schema."
)
async def detect_event_transitions(
client: LLMClient,
*,
classifier_model: str,
narrative_text: str,
active_events: list[dict], # [{event_id, kind, status, props}, ...]
timeout_s: float = 30.0,
) -> EventLifecycleDecision:
"""Classify event transitions for the latest turn. Empty active_events
short-circuits without an LLM call."""
if not active_events:
return EventLifecycleDecision()
user_lines = ["Active events:"]
for ev in active_events:
user_lines.append(
f"- event_id={ev['event_id']} kind={ev['kind']} "
f"status={ev['status']} props={ev.get('props', {})}"
)
user_lines.append("")
user_lines.append("Latest narrative:")
user_lines.append(narrative_text.strip())
user = "\n".join(user_lines)
return await classify(
client,
model=classifier_model,
system=_SYSTEM,
user=user,
schema=EventLifecycleDecision,
default=EventLifecycleDecision(),
timeout_s=timeout_s,
)
__all__ = ["EventTransition", "EventLifecycleDecision", "detect_event_transitions"]
+149
View File
@@ -0,0 +1,149 @@
"""Event-completion promotion (T56).
When an event reaches ``status='completed'``, read its ``props_json``
and emit promotion events into the appropriate state stores.
Synchronous, no LLM. Skips when the event status is not ``completed``
(cancelled / expired terminate the event without promoting).
Props recognized:
- ``acquired_objects: list[str]`` — emits a ``manual_edit`` with
``target_kind="memory_pov_summary"`` per object on the host's memory
row, recording the acquisition. Phase 3 is a stub: it requires both
``host_bot_id`` and ``host_memory_id`` (an existing memories.id) to
be present in props; missing either skips that object cleanly.
Phase 4 will introduce a real inventory schema.
- ``knowledge_facts: list[{owner_id, target_id, fact}]`` — emits an
``edge_update`` event on the directed ``owner_id -> target_id`` edge
with the fact appended to ``knowledge_facts``. The ``edge_update``
projector accepts ``knowledge_facts`` as a list and extends the
edge's stored ``knowledge_json``.
- ``relationship_change: {summary, source_id, target_id}`` — emits a
``manual_edit`` with ``target_kind="edge_summary"`` overwriting the
edge's ``summary`` field on the directed pair.
Anything else stays in the closed event record (the projector kept
the row; no further promotion).
"""
from __future__ import annotations
from sqlite3 import Connection
from chat.eventlog.log import append_and_apply
from chat.state.events import get_event
def promote_completed_event(
conn: Connection,
*,
event_id: str,
chat_id: str,
chat_clock_at: str | None,
) -> dict:
"""Read the completed event's props and emit promotion events.
Returns a dict of counts keyed by promoted artifact:
``{"acquired_objects", "knowledge_facts", "relationship_change"}``.
Skips silently if the event row is missing or its status is not
``completed`` — cancelled / expired events terminate without any
promotion.
"""
counts = {
"acquired_objects": 0,
"knowledge_facts": 0,
"relationship_change": 0,
}
event = get_event(conn, event_id)
if event is None or event["status"] != "completed":
return counts
props = event.get("props") or {}
# acquired_objects: each becomes a memory_pov_summary edit (Phase 3
# stub). The manual_edit projector requires a valid memory rowid as
# ``target_id`` (it does ``int(target_id)``), so skip cleanly when
# neither a host_bot_id nor a host_memory_id is supplied.
host_bot_id = props.get("host_bot_id")
host_memory_id = props.get("host_memory_id")
for obj in props.get("acquired_objects", []) or []:
if host_bot_id is None or host_memory_id is None:
continue
append_and_apply(
conn,
kind="manual_edit",
payload={
"target_kind": "memory_pov_summary",
"target_id": host_memory_id,
"owner_id": host_bot_id,
"chat_id": chat_id,
"prior_value": "",
"new_value": f"Acquired: {obj}",
"source": "event_promotion",
"event_id": event_id,
"chat_clock_at": chat_clock_at,
},
)
counts["acquired_objects"] += 1
# knowledge_facts: each becomes an edge_update appending the fact.
for fact_entry in props.get("knowledge_facts", []) or []:
owner_id = fact_entry.get("owner_id")
target_id = fact_entry.get("target_id")
fact = fact_entry.get("fact", "")
if not owner_id or not target_id or not fact:
continue
append_and_apply(
conn,
kind="edge_update",
payload={
"source_id": owner_id,
"target_id": target_id,
"chat_id": chat_id,
"affinity_delta": 0,
"trust_delta": 0,
"knowledge_facts": [fact],
"last_interaction_at": chat_clock_at,
"last_interaction_chat_id": chat_id,
"source": "event_promotion",
"event_id": event_id,
},
)
counts["knowledge_facts"] += 1
# relationship_change: edge_summary manual_edit on the directed pair.
# The manual_edit projector for ``edge_summary`` keys on a
# ``target_id`` dict ``{source_id, target_id}`` (see
# chat/state/manual_edit.py); we shape the payload to match.
rc = props.get("relationship_change") or {}
if rc:
source_id = rc.get("source_id")
rc_target_id = rc.get("target_id")
summary = rc.get("summary", "")
if source_id and rc_target_id and summary:
append_and_apply(
conn,
kind="manual_edit",
payload={
"target_kind": "edge_summary",
"target_id": {
"source_id": source_id,
"target_id": rc_target_id,
},
"chat_id": chat_id,
"prior_value": "",
"new_value": summary,
"source": "event_promotion",
"event_id": event_id,
"chat_clock_at": chat_clock_at,
},
)
counts["relationship_change"] += 1
return counts
__all__ = ["promote_completed_event"]
+102 -1
View File
@@ -29,6 +29,8 @@ keeps moving.
from __future__ import annotations
import json
import uuid
from datetime import datetime, timezone
from sqlite3 import Connection
from pydantic import BaseModel, Field
@@ -167,6 +169,7 @@ async def _summarize_and_apply_for_witness(
you_name: str,
dialogue: list[dict],
timeout_s: float,
key_quotes_suffix: str = "",
) -> ScenePOVSummary:
"""Run :func:`summarize_scene` for one bot witness and apply the
three projected updates (memory pov_summary rewrite, edge summary
@@ -175,6 +178,10 @@ async def _summarize_and_apply_for_witness(
Tolerant of missing pieces in the same way Phase 1 was: no memory
row -> skip the rewrite; no edge row -> skip the edge_summary write
(the empty-default classifier output simply yields no rewrites).
``key_quotes_suffix`` is appended verbatim to the per-POV summary
text before the rewrite lands (T58.1) — empty string is the no-op
default for low-significance scenes.
"""
from chat.state.edges import get_edge
from chat.state.entities import get_bot
@@ -206,6 +213,7 @@ async def _summarize_and_apply_for_witness(
# Empty default -> skip the memory rewrite; the seeded
# per-turn pov_summary stays in place.
continue
new_value = pov.summary + key_quotes_suffix
append_and_apply(
conn,
kind="manual_edit",
@@ -213,7 +221,7 @@ async def _summarize_and_apply_for_witness(
"target_kind": "memory_pov_summary",
"target_id": int(memory_id),
"prior_value": prior_pov,
"new_value": pov.summary,
"new_value": new_value,
},
)
@@ -255,6 +263,40 @@ async def _summarize_and_apply_for_witness(
return pov
def _build_key_quotes_suffix(conn: Connection, scene_id: int) -> str:
"""If the scene's max-turn-significance is >= 2, build the
"Key quotes:" suffix from the top-3 highest-significance memory rows
(per requirements §11.1). Otherwise return the empty string so the
per-POV summaries collapse fully (low-significance scenes lose all
raw text in favor of the classifier rewrite).
Quote source is each memory's current ``pov_summary`` — the raw
per-turn narrative seeded by T21, since this helper is called BEFORE
the per-POV rewrite. Texts are truncated to 200 chars to bound
memory row growth across many witnesses.
"""
row = conn.execute(
"SELECT MAX(significance) FROM memories WHERE scene_id = ?",
(scene_id,),
).fetchone()
max_sig = (row[0] if row else None) or 0
if max_sig < 2:
return ""
cur = conn.execute(
"SELECT pov_summary FROM memories WHERE scene_id = ? "
"ORDER BY significance DESC, id ASC LIMIT 3",
(scene_id,),
)
quotes = [
(r[0] or "")[:200]
for r in cur.fetchall()
]
if not quotes:
return ""
lines = "\n".join(f'- "{q}"' for q in quotes)
return f"\n\nKey quotes:\n{lines}"
async def apply_scene_close_summary(
conn: Connection,
client: LLMClient,
@@ -296,8 +338,10 @@ async def apply_scene_close_summary(
"""
# Local imports to keep the module-level surface tight and avoid
# any chance of a circular dep through chat.state.*.
from chat.services.thread_detection import detect_threads
from chat.state.entities import get_bot, get_you
from chat.state.group_node import get_group_node
from chat.state.threads import list_open_threads
from chat.state.world import get_chat
you_entity = get_you(conn) or {"name": "you", "persona": ""}
@@ -308,6 +352,11 @@ async def apply_scene_close_summary(
dialogue = _read_recent_dialogue(conn, chat_id)
# T58.1: build the "Key quotes:" suffix BEFORE the per-POV rewrites
# land — quote source is the raw seeded pov_summary text on each
# memory row, which the rewrite about to fire would clobber.
key_quotes_suffix = _build_key_quotes_suffix(conn, scene_id)
host_pov = await _summarize_and_apply_for_witness(
conn,
client,
@@ -318,6 +367,7 @@ async def apply_scene_close_summary(
you_name=you_name,
dialogue=dialogue,
timeout_s=timeout_s,
key_quotes_suffix=key_quotes_suffix,
)
guest_pov: ScenePOVSummary | None = None
@@ -332,6 +382,7 @@ async def apply_scene_close_summary(
you_name=you_name,
dialogue=dialogue,
timeout_s=timeout_s,
key_quotes_suffix=key_quotes_suffix,
)
# Group node update: T70 runs a third classifier call to merge the
@@ -364,6 +415,56 @@ async def apply_scene_close_summary(
},
)
# T58.2: thread detection on close. Reuses the dialogue we already
# gathered for per-POV summarization — same {speaker, text} shape
# detect_threads expects. Failure-tolerant: classify() returns the
# empty default on retry-exhaustion, and the broad except below
# protects the close pipeline from any other classifier/mock flap.
try:
thread_result = await detect_threads(
client,
classifier_model=classifier_model,
scene_transcript=dialogue,
open_threads=list_open_threads(conn, chat_id),
timeout_s=timeout_s,
)
except Exception:
from chat.services.thread_detection import ThreadDetectionResult
thread_result = ThreadDetectionResult()
for cand in thread_result.candidates:
if cand.action == "open":
new_thread_id = f"thr_{uuid.uuid4().hex[:12]}"
append_and_apply(
conn,
kind="thread_opened",
payload={
"thread_id": new_thread_id,
"chat_id": chat_id,
"title": cand.title,
"summary": cand.summary,
},
)
elif cand.action == "update" and cand.existing_thread_id:
append_and_apply(
conn,
kind="thread_updated",
payload={
"thread_id": cand.existing_thread_id,
"summary": cand.summary,
"last_referenced_scene_id": scene_id,
},
)
elif cand.action == "close" and cand.existing_thread_id:
append_and_apply(
conn,
kind="thread_closed",
payload={
"thread_id": cand.existing_thread_id,
"closed_at": datetime.now(timezone.utc).isoformat(),
},
)
return host_pov
+131
View File
@@ -0,0 +1,131 @@
"""Skip narration service (T53).
Generates brief transition prose for elision and jump skips.
Skips come in two flavors that read very differently:
* **Elision** — collapses an in-progress activity into its expected
end-state in 1-2 sentences, narrated from the speaker bot's POV.
Example: "skip ahead to when we arrive" while the characters are
driving — output describes pulling into the lot.
* **Jump** — bridges a longer fiction-time delta ("next morning", "a
week later") in 2-3 sentences, setting the scene at the new time.
Output is free-form prose, not structured JSON, so this service calls
``client.generate`` directly rather than going through the classifier
path used by, e.g., :mod:`chat.services.scene_summarize`. A
deterministic template fallback fires on any LLM failure so the skip
flow keeps moving even when the model is down — important because
skips are a UI-blocking operation; we'd rather show a parenthetical
sentence than hang the chat indefinitely.
"""
from __future__ import annotations
from chat.llm.client import LLMClient, Message
_ELISION_SYSTEM = (
"You write a brief 1-2 sentence transition that elides the time "
"between an in-progress activity and its expected end-state, "
"narrated from the speaker's POV. Keep it grounded and concrete. "
"Do not invent new events or characters."
)
_JUMP_SYSTEM = (
"You write a brief 2-3 sentence transition narrating a jump in "
"fiction time (e.g., 'next morning', 'a week later'), narrated "
"from the speaker's POV. Set the scene at the new time. Keep it "
"grounded — no invented major events. If a landing-state hint is "
"provided, weave it in naturally."
)
async def narrate_skip(
client: LLMClient,
*,
narrative_model: str,
skip_kind: str,
speaker_bot: dict,
you_name: str,
current_time: str,
new_time: str,
current_activity: str,
landing_state_hint: str = "",
timeout_s: float = 60.0,
) -> str:
"""Generate brief transition prose for a time skip.
``skip_kind`` is ``"elision"`` or ``"jump"``; any other value short-
circuits to the deterministic fallback (defensive — callers
shouldn't be inventing new kinds without updating this service).
Returns plain text. Never raises: any LLM error, an empty/blank
result, or an unknown ``skip_kind`` falls back to a parenthetical
template like ``"(next morning: having coffee in the kitchen.)"``
so the skip UI always has *something* to render.
"""
fallback = _build_fallback(
skip_kind=skip_kind,
new_time=new_time,
current_activity=current_activity,
landing_state_hint=landing_state_hint,
)
if skip_kind not in ("elision", "jump"):
return fallback
system = _ELISION_SYSTEM if skip_kind == "elision" else _JUMP_SYSTEM
user = (
f"Speaker: {speaker_bot.get('name', 'speaker')}\n"
f"Persona: {speaker_bot.get('persona', '')}\n"
f"Other party: {you_name}\n"
f"Current time: {current_time}\n"
f"New time: {new_time}\n"
f"Current activity: {current_activity}\n"
)
if landing_state_hint:
user += f"Landing state hint: {landing_state_hint}\n"
try:
result = await client.generate(
[
Message(role="system", content=system),
Message(role="user", content=user),
],
model=narrative_model,
max_tokens=200,
temperature=0.7,
)
text = (result or "").strip()
if not text:
return fallback
return text
except Exception:
# Any failure — network blip, timeout, mock raising in tests —
# collapses to the deterministic template so the skip pipeline
# is never blocked on the LLM being available.
return fallback
def _build_fallback(
*,
skip_kind: str,
new_time: str,
current_activity: str,
landing_state_hint: str,
) -> str:
"""Deterministic parenthetical narration used when the LLM fails.
Both flavors render the same shape today: ``(<new_time>:
<detail>.)``. They're separated as branches to make it easy to
diverge later (e.g. an elision-specific template) without churning
the call site or the public signature.
"""
detail = landing_state_hint or current_activity or "moments later"
if skip_kind == "elision":
return f"({new_time}: {detail}.)"
return f"({new_time}: {detail}.)"
__all__ = ["narrate_skip"]
+74
View File
@@ -0,0 +1,74 @@
"""Synthesized-memories service (T54).
When the user jump-skips with 'anything notable happen?' prose, parse
that prose into 1-N synthesized memories per present bot. Each memory
carries source="synthesized" and reliability=0.7 (lower than direct).
Caller (T62 skip flow) writes the memories via record_turn_memory_for_present.
"""
from __future__ import annotations
from pydantic import BaseModel, Field
from chat.llm.classify import classify
from chat.llm.client import LLMClient
class SynthesizedMemory(BaseModel):
text: str
significance: int = 1 # 0..3, default 1
affinity_delta: int = 0
trust_delta: int = 0
class SynthesizedDigest(BaseModel):
memories: list[SynthesizedMemory] = Field(default_factory=list)
_SYSTEM = (
"You parse a short user-supplied prose describing 'anything notable' "
"that happened during a time skip into 1-N synthesized memories from "
"a single bot's POV. Each memory has: text (one factual sentence "
"from that bot's perspective), significance (0-3, default 1; only "
"use 2 or 3 for genuinely scene-level or relationship-altering "
"events), affinity_delta and trust_delta (-10..+10, default 0; "
"use small adjustments only when prose explicitly describes a shift). "
"Empty/whitespace prose returns an empty memories list. Output "
"strict JSON matching the schema."
)
async def synthesize_memories(
client: LLMClient,
*,
classifier_model: str,
prose: str,
bot_name: str,
bot_persona: str,
you_name: str,
timeout_s: float = 30.0,
) -> SynthesizedDigest:
"""Parse 'anything notable' prose into structured memories from a
single bot's POV. Empty/whitespace prose short-circuits to an
empty digest (no LLM call)."""
if not prose or not prose.strip():
return SynthesizedDigest()
user = (
f"Bot: {bot_name}\n"
f"Persona: {bot_persona}\n"
f"Other party: {you_name}\n\n"
f"Prose:\n{prose.strip()}"
)
return await classify(
client,
model=classifier_model,
system=_SYSTEM,
user=user,
schema=SynthesizedDigest,
default=SynthesizedDigest(),
timeout_s=timeout_s,
)
__all__ = ["SynthesizedMemory", "SynthesizedDigest", "synthesize_memories"]
+89
View File
@@ -0,0 +1,89 @@
"""Thread-detection service (T55).
On scene close, classify the transcript into thread open/update/close
candidates. Returns ThreadCandidate list; caller (T58 scene compression)
emits one thread_opened/thread_updated/thread_closed event per candidate.
"""
from __future__ import annotations
from pydantic import BaseModel, Field
from chat.llm.classify import classify
from chat.llm.client import LLMClient
class ThreadCandidate(BaseModel):
action: str # "open" | "update" | "close"
title: str = "" # required for "open"; ignored otherwise
summary: str = ""
existing_thread_id: str | None = None # required for "update" / "close"
class ThreadDetectionResult(BaseModel):
candidates: list[ThreadCandidate] = Field(default_factory=list)
_SYSTEM = (
"You analyze a closed scene's transcript to identify narrative "
"threads (unresolved arcs, dangling questions, promises made, "
"open obligations). Choose actions:\n"
"- 'open': a NEW thread the scene introduced. Provide title (short "
"noun phrase) + summary (one sentence).\n"
"- 'update': an EXISTING open thread that the scene developed. "
"Provide existing_thread_id + new summary.\n"
"- 'close': an EXISTING open thread that the scene resolved. "
"Provide existing_thread_id; summary may capture the resolution.\n"
"Conservative bias: most scenes do NOT open new threads. Only "
"produce candidates when the transcript clearly justifies them. "
"Output strict JSON matching the schema."
)
async def detect_threads(
client: LLMClient,
*,
classifier_model: str,
scene_transcript: list[dict], # [{speaker, text}, ...]
open_threads: list[dict], # [{thread_id, title, summary}, ...]
timeout_s: float = 30.0,
) -> ThreadDetectionResult:
"""Classify scene close into thread open/update/close candidates."""
if not scene_transcript:
return ThreadDetectionResult()
transcript_lines = [
f"{turn.get('speaker', 'unknown')}: {turn.get('text', '')}"
for turn in scene_transcript
]
threads_lines = []
if open_threads:
threads_lines.append("Currently open threads:")
for t in open_threads:
threads_lines.append(
f"- thread_id={t['thread_id']} "
f"title={t.get('title', '')} "
f"summary={t.get('summary', '')}"
)
else:
threads_lines.append("No currently open threads.")
user = (
"Scene transcript:\n"
+ "\n".join(transcript_lines)
+ "\n\n"
+ "\n".join(threads_lines)
)
return await classify(
client,
model=classifier_model,
system=_SYSTEM,
user=user,
schema=ThreadDetectionResult,
default=ThreadDetectionResult(),
timeout_s=timeout_s,
)
__all__ = ["ThreadCandidate", "ThreadDetectionResult", "detect_threads"]
+127
View File
@@ -0,0 +1,127 @@
from __future__ import annotations
import json
from sqlite3 import Connection
from chat.eventlog.projector import on
from chat.eventlog.log import Event
_TERMINAL_STATUSES = {"completed", "cancelled", "expired"}
@on("event_planned")
def _apply_event_planned(conn: Connection, e: Event) -> None:
p = e.payload
conn.execute(
"INSERT OR IGNORE INTO events "
"(event_id, chat_id, kind, status, props_json, planned_for) "
"VALUES (?, ?, ?, 'planned', ?, ?)",
(
p["event_id"],
p["chat_id"],
p["kind"],
json.dumps(p.get("props", {})),
p.get("planned_for"),
),
)
@on("event_started")
def _apply_event_started(conn: Connection, e: Event) -> None:
p = e.payload
# Idempotent: only transition from non-terminal status.
conn.execute(
"UPDATE events SET status = 'active', started_at = ?, updated_at = datetime('now') "
"WHERE event_id = ? AND status NOT IN ('completed','cancelled','expired')",
(p.get("started_at"), p["event_id"]),
)
@on("event_completed")
def _apply_event_completed(conn: Connection, e: Event) -> None:
p = e.payload
conn.execute(
"UPDATE events SET status = 'completed', completed_at = ?, updated_at = datetime('now') "
"WHERE event_id = ? AND status NOT IN ('completed','cancelled','expired')",
(p.get("completed_at"), p["event_id"]),
)
@on("event_cancelled")
def _apply_event_cancelled(conn: Connection, e: Event) -> None:
p = e.payload
conn.execute(
"UPDATE events SET status = 'cancelled', completed_at = ?, updated_at = datetime('now') "
"WHERE event_id = ? AND status NOT IN ('completed','cancelled','expired')",
(p.get("completed_at"), p["event_id"]),
)
@on("event_expired")
def _apply_event_expired(conn: Connection, e: Event) -> None:
p = e.payload
conn.execute(
"UPDATE events SET status = 'expired', completed_at = ?, updated_at = datetime('now') "
"WHERE event_id = ? AND status NOT IN ('completed','cancelled','expired')",
(p.get("completed_at"), p["event_id"]),
)
def get_event(conn: Connection, event_id: str) -> dict | None:
row = conn.execute(
"SELECT event_id, chat_id, kind, status, props_json, planned_for, "
"started_at, completed_at, created_at, updated_at "
"FROM events WHERE event_id = ?",
(event_id,),
).fetchone()
if not row:
return None
return {
"event_id": row[0],
"chat_id": row[1],
"kind": row[2],
"status": row[3],
"props": json.loads(row[4]),
"planned_for": row[5],
"started_at": row[6],
"completed_at": row[7],
"created_at": row[8],
"updated_at": row[9],
}
def list_active_events(conn: Connection, chat_id: str) -> list[dict]:
rows = conn.execute(
"SELECT event_id, chat_id, kind, status, props_json, planned_for, "
"started_at, completed_at, created_at, updated_at "
"FROM events WHERE chat_id = ? AND status IN ('planned','active') "
"ORDER BY id ASC",
(chat_id,),
).fetchall()
return [
{
"event_id": r[0], "chat_id": r[1], "kind": r[2], "status": r[3],
"props": json.loads(r[4]),
"planned_for": r[5], "started_at": r[6], "completed_at": r[7],
"created_at": r[8], "updated_at": r[9],
}
for r in rows
]
def list_events_in_status(conn: Connection, chat_id: str, status: str) -> list[dict]:
rows = conn.execute(
"SELECT event_id, chat_id, kind, status, props_json, planned_for, "
"started_at, completed_at, created_at, updated_at "
"FROM events WHERE chat_id = ? AND status = ? ORDER BY id ASC",
(chat_id, status),
).fetchall()
return [
{
"event_id": r[0], "chat_id": r[1], "kind": r[2], "status": r[3],
"props": json.loads(r[4]),
"planned_for": r[5], "started_at": r[6], "completed_at": r[7],
"created_at": r[8], "updated_at": r[9],
}
for r in rows
]
+15 -2
View File
@@ -94,6 +94,14 @@ def get_pinned(conn: Connection, owner_id: str) -> list[dict]:
_SIGNIFICANCE_WEIGHT = 0.3
_RECENCY_WEIGHT = 0.5
# T57 (Phase 3, §11.1): significance multiplier applied to the SQL ORDER BY in
# ``search_memories`` so that the FTS over-fetch already prefers
# higher-significance rows for tied / near-tied BM25 ranks. Module-level so it
# can be tuned without a code change. BM25 ``rank`` is lower-is-better, so the
# bias is *subtracted* from rank in the ASC ordering — equivalent to multiplying
# a higher-is-better score by a positive constant per the spec wording.
SIGNIFICANCE_RANK_BIAS = 0.5
def search_memories(
conn: Connection,
@@ -137,10 +145,15 @@ def search_memories(
"JOIN memories m ON m.id = memories_fts.rowid "
f"WHERE m.owner_id = ? AND m.{witness_col} = 1 "
"AND memories_fts MATCH ? "
"ORDER BY memories_fts.rank "
# T57: significance multiplier biases the FTS over-fetch order. BM25
# ``rank`` is lower-is-better, so subtracting ``significance * BIAS``
# surfaces higher-significance rows above lower-significance rows with
# equal/near-equal match strength. Equivalent to ``score × constant``
# per §11.1 once the rank is inverted to a higher-is-better score.
"ORDER BY (memories_fts.rank - m.significance * ?) ASC "
"LIMIT ?"
)
cur = conn.execute(sql, (owner_id, query, over_fetch))
cur = conn.execute(sql, (owner_id, query, SIGNIFICANCE_RANK_BIAS, over_fetch))
rows = cur.fetchall()
if not rows:
return []
+123
View File
@@ -0,0 +1,123 @@
from __future__ import annotations
from sqlite3 import Connection
from chat.eventlog.projector import on
from chat.eventlog.log import Event
@on("thread_opened")
def _apply_thread_opened(conn: Connection, e: Event) -> None:
p = e.payload
conn.execute(
"INSERT OR IGNORE INTO threads "
"(thread_id, chat_id, title, summary, status) "
"VALUES (?, ?, ?, ?, 'open')",
(
p["thread_id"],
p["chat_id"],
p["title"],
p.get("summary", ""),
),
)
@on("thread_updated")
def _apply_thread_updated(conn: Connection, e: Event) -> None:
p = e.payload
# Idempotent: closed threads ignore subsequent updates.
conn.execute(
"UPDATE threads SET summary = ?, last_referenced_scene_id = ?, "
"updated_at = datetime('now') "
"WHERE thread_id = ? AND status = 'open'",
(
p.get("summary", ""),
p.get("last_referenced_scene_id"),
p["thread_id"],
),
)
@on("thread_closed")
def _apply_thread_closed(conn: Connection, e: Event) -> None:
p = e.payload
conn.execute(
"UPDATE threads SET status = 'closed', closed_at = ?, "
"updated_at = datetime('now') "
"WHERE thread_id = ? AND status = 'open'",
(p.get("closed_at"), p["thread_id"]),
)
def get_thread(conn: Connection, thread_id: str) -> dict | None:
row = conn.execute(
"SELECT thread_id, chat_id, title, summary, status, "
"opened_at, closed_at, last_referenced_scene_id, "
"created_at, updated_at "
"FROM threads WHERE thread_id = ?",
(thread_id,),
).fetchone()
if not row:
return None
return {
"thread_id": row[0],
"chat_id": row[1],
"title": row[2],
"summary": row[3],
"status": row[4],
"opened_at": row[5],
"closed_at": row[6],
"last_referenced_scene_id": row[7],
"created_at": row[8],
"updated_at": row[9],
}
def list_open_threads(conn: Connection, chat_id: str) -> list[dict]:
rows = conn.execute(
"SELECT thread_id, chat_id, title, summary, status, "
"opened_at, closed_at, last_referenced_scene_id, "
"created_at, updated_at "
"FROM threads WHERE chat_id = ? AND status = 'open' "
"ORDER BY id ASC",
(chat_id,),
).fetchall()
return [
{
"thread_id": r[0], "chat_id": r[1], "title": r[2],
"summary": r[3], "status": r[4],
"opened_at": r[5], "closed_at": r[6],
"last_referenced_scene_id": r[7],
"created_at": r[8], "updated_at": r[9],
}
for r in rows
]
def list_threads(conn: Connection, chat_id: str, status: str | None = None) -> list[dict]:
if status is None:
rows = conn.execute(
"SELECT thread_id, chat_id, title, summary, status, "
"opened_at, closed_at, last_referenced_scene_id, "
"created_at, updated_at "
"FROM threads WHERE chat_id = ? ORDER BY id ASC",
(chat_id,),
).fetchall()
else:
rows = conn.execute(
"SELECT thread_id, chat_id, title, summary, status, "
"opened_at, closed_at, last_referenced_scene_id, "
"created_at, updated_at "
"FROM threads WHERE chat_id = ? AND status = ? "
"ORDER BY id ASC",
(chat_id, status),
).fetchall()
return [
{
"thread_id": r[0], "chat_id": r[1], "title": r[2],
"summary": r[3], "status": r[4],
"opened_at": r[5], "closed_at": r[6],
"last_referenced_scene_id": r[7],
"created_at": r[8], "updated_at": r[9],
}
for r in rows
]
+28
View File
@@ -29,6 +29,34 @@ def _apply_chat_created(conn: Connection, e: Event) -> None:
)
@on("time_skip_elision")
def _apply_time_skip_elision(conn: Connection, e: Event) -> None:
p = e.payload
conn.execute(
"UPDATE chat_state SET time = ? WHERE chat_id = ?",
(p["new_time"], p["chat_id"]),
)
@on("time_skip_jump")
def _apply_time_skip_jump(conn: Connection, e: Event) -> None:
p = e.payload
chat_id = p["chat_id"]
conn.execute(
"UPDATE chat_state SET time = ? WHERE chat_id = ?",
(p["new_time"], chat_id),
)
if p.get("reset_activity", False):
# Activity rows are keyed by entity_id with a container_id FK.
# Each chat owns its containers, so deleting activity rows whose
# container_id belongs to this chat clears every present entity.
conn.execute(
"DELETE FROM activity "
"WHERE container_id IN (SELECT id FROM containers WHERE chat_id = ?)",
(chat_id,),
)
@on("guest_added")
def _apply_guest_added(conn: Connection, e: Event) -> None:
p = e.payload
+103
View File
@@ -0,0 +1,103 @@
"""Tests for the event-lifecycle detection service (T52).
Per Phase 3, after each narrated turn we ask a classifier whether any
active events transitioned (started, completed, cancelled). The bias is
strongly toward an empty result — most turns do NOT resolve or start a
known event, and the turn-flow caller (T61) only appends an
event_started/completed/cancelled record when this service yields one.
These tests cover:
* The classifier returning a single transition is honored end-to-end.
* An empty ``active_events`` list short-circuits before any LLM call,
so callers that hold no live events pay zero classifier cost.
* Three rounds of malformed JSON exhaust ``classify``'s retries and we
fall back to the empty default — graceful degradation per §3.3.
"""
from __future__ import annotations
import json
import pytest
from chat.llm.mock import MockLLMClient
from chat.services.event_lifecycle import (
EventLifecycleDecision,
detect_event_transitions,
)
@pytest.mark.asyncio
async def test_detects_one_transition_happy_path():
canned = json.dumps(
{
"transitions": [
{
"event_id": "evt_1",
"new_status": "completed",
"reason": "they arrived at the park",
}
]
}
)
mock = MockLLMClient(canned=[canned])
result = await detect_event_transitions(
mock,
classifier_model="x",
narrative_text="They walked through the park gate, finally there.",
active_events=[
{
"event_id": "evt_1",
"kind": "date_at_park",
"status": "active",
"props": {},
}
],
)
assert isinstance(result, EventLifecycleDecision)
assert len(result.transitions) == 1
assert result.transitions[0].event_id == "evt_1"
assert result.transitions[0].new_status == "completed"
assert result.transitions[0].reason == "they arrived at the park"
@pytest.mark.asyncio
async def test_empty_active_events_short_circuits_without_classifier_call():
"""No active events -> no classifier call.
The mock has an empty canned list; any ``generate`` call would raise
``IndexError`` from ``list.pop(0)``. The test passing proves the
short-circuit holds.
"""
mock = MockLLMClient(canned=[])
result = await detect_event_transitions(
mock,
classifier_model="x",
narrative_text="Just a quiet moment between them.",
active_events=[],
)
assert isinstance(result, EventLifecycleDecision)
assert result.transitions == []
@pytest.mark.asyncio
async def test_classifier_failure_returns_empty_default():
"""``classify`` retries 3 times; after all fail it returns the empty
default so the turn flow keeps moving (§3.3 graceful degradation)."""
mock = MockLLMClient(canned=["bad", "bad", "bad"])
result = await detect_event_transitions(
mock,
classifier_model="x",
narrative_text="Some text the classifier will choke on.",
active_events=[
{
"event_id": "evt_1",
"kind": "date_at_park",
"status": "active",
"props": {},
}
],
)
assert isinstance(result, EventLifecycleDecision)
assert result.transitions == []
+256
View File
@@ -0,0 +1,256 @@
"""Tests for the event-completion promotion service (T56).
When an event reaches ``status='completed'``, the orchestrator promotes
structured artifacts the event carried (``acquired_objects``,
``knowledge_facts``, ``relationship_change``) into the appropriate
state stores via downstream events. Cancelled / expired events do NOT
promote — the closed event row is left in place but no follow-on
events fire.
"""
from __future__ import annotations
import json
from chat.db.connection import open_db
from chat.db.migrate import apply_migrations
from chat.eventlog.log import append_event
from chat.eventlog.projector import project
from chat.services.event_promotion import promote_completed_event
from chat.state.edges import get_edge
import chat.state.edges # noqa: F401 - register edge_update handler
import chat.state.entities # noqa: F401 - register handlers
import chat.state.events # noqa: F401 - register events handlers
import chat.state.manual_edit # noqa: F401 - register manual_edit handler
import chat.state.world # noqa: F401 - register handlers
def _bot_payload(bot_id: str, name: str) -> dict:
return {
"id": bot_id,
"name": name,
"persona": "thoughtful, observant",
"voice_samples": [],
"traits": [],
"backstory": "",
"initial_relationship_to_you": "coworker",
"kickoff_prose": "",
}
def _chat_payload(chat_id: str = "chat_bot_a") -> dict:
return {
"id": chat_id,
"host_bot_id": "bot_a",
"guest_bot_id": "bot_b",
"initial_time": "2026-04-26T20:00:00+00:00",
"narrative_anchor": "Day 1 evening",
"weather": "clear",
}
def _seed_chat(conn) -> None:
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB"))
append_event(conn, kind="chat_created", payload=_chat_payload())
def _seed_event(
conn,
*,
event_id: str,
props: dict,
terminal_kind: str = "event_completed",
) -> None:
"""Append event_planned, then a terminal transition (default completed)."""
append_event(
conn,
kind="event_planned",
payload={
"event_id": event_id,
"chat_id": "chat_bot_a",
"kind": "story_event",
"props": props,
"planned_for": "2026-04-30T18:00:00+00:00",
},
)
append_event(
conn,
kind=terminal_kind,
payload={
"event_id": event_id,
"completed_at": "2026-04-30T20:00:00+00:00",
},
)
project(conn)
def _max_event_id(conn) -> int:
return conn.execute("SELECT COALESCE(MAX(id), 0) FROM event_log").fetchone()[0]
def _events_after(conn, after_id: int, kind: str) -> list[dict]:
rows = conn.execute(
"SELECT id, kind, payload_json FROM event_log "
"WHERE id > ? AND kind = ? ORDER BY id ASC",
(after_id, kind),
).fetchall()
return [
{"id": r[0], "kind": r[1], "payload": json.loads(r[2])} for r in rows
]
def test_empty_props_no_op(tmp_path):
"""Completed event with empty props produces no promotion events."""
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_chat(conn)
_seed_event(conn, event_id="evt_empty", props={})
before = _max_event_id(conn)
counts = promote_completed_event(
conn,
event_id="evt_empty",
chat_id="chat_bot_a",
chat_clock_at="2026-04-30T20:00:00+00:00",
)
assert counts == {
"acquired_objects": 0,
"knowledge_facts": 0,
"relationship_change": 0,
}
# No new edge_update or manual_edit rows after the promote call.
assert _events_after(conn, before, "edge_update") == []
assert _events_after(conn, before, "manual_edit") == []
def test_knowledge_facts_emits_edge_update(tmp_path):
"""A knowledge_facts entry promotes to an edge_update on the directed edge."""
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_chat(conn)
_seed_event(
conn,
event_id="evt_kf",
props={
"knowledge_facts": [
{
"owner_id": "bot_a",
"target_id": "you",
"fact": "Maya prefers tea over coffee",
}
]
},
)
before = _max_event_id(conn)
counts = promote_completed_event(
conn,
event_id="evt_kf",
chat_id="chat_bot_a",
chat_clock_at="2026-04-30T20:00:00+00:00",
)
assert counts["knowledge_facts"] == 1
assert counts["acquired_objects"] == 0
assert counts["relationship_change"] == 0
# An edge_update event landed in the event_log AFTER the promote call.
new_edge_updates = _events_after(conn, before, "edge_update")
assert len(new_edge_updates) == 1
payload = new_edge_updates[0]["payload"]
assert payload["source_id"] == "bot_a"
assert payload["target_id"] == "you"
assert payload["knowledge_facts"] == ["Maya prefers tea over coffee"]
# And the projected edge has the fact applied.
edge = get_edge(conn, "bot_a", "you")
assert edge is not None
assert "Maya prefers tea over coffee" in edge["knowledge"]
def test_relationship_change_emits_manual_edit(tmp_path):
"""A relationship_change promotes to a manual_edit edge_summary."""
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_chat(conn)
_seed_event(
conn,
event_id="evt_rc",
props={
"relationship_change": {
"source_id": "bot_a",
"target_id": "you",
"summary": "they're now dating",
}
},
)
before = _max_event_id(conn)
counts = promote_completed_event(
conn,
event_id="evt_rc",
chat_id="chat_bot_a",
chat_clock_at="2026-04-30T20:00:00+00:00",
)
assert counts["relationship_change"] == 1
assert counts["knowledge_facts"] == 0
assert counts["acquired_objects"] == 0
new_manual_edits = _events_after(conn, before, "manual_edit")
# Filter to edge_summary only — Phase 3 stub may also emit
# memory_pov_summary entries for acquired_objects, but here there
# are none.
edge_summary_edits = [
m for m in new_manual_edits
if m["payload"].get("target_kind") == "edge_summary"
]
assert len(edge_summary_edits) == 1
payload = edge_summary_edits[0]["payload"]
assert payload["target_kind"] == "edge_summary"
assert payload["target_id"] == {"source_id": "bot_a", "target_id": "you"}
assert payload["new_value"] == "they're now dating"
def test_cancelled_event_does_not_promote(tmp_path):
"""Cancelled events have promotable props ignored — no follow-on events."""
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_chat(conn)
_seed_event(
conn,
event_id="evt_canx",
props={
"knowledge_facts": [
{"owner_id": "bot_a", "target_id": "you", "fact": "x"}
],
"relationship_change": {
"source_id": "bot_a",
"target_id": "you",
"summary": "ignored",
},
},
terminal_kind="event_cancelled",
)
before = _max_event_id(conn)
counts = promote_completed_event(
conn,
event_id="evt_canx",
chat_id="chat_bot_a",
chat_clock_at="2026-04-30T20:00:00+00:00",
)
assert counts == {
"acquired_objects": 0,
"knowledge_facts": 0,
"relationship_change": 0,
}
assert _events_after(conn, before, "edge_update") == []
assert _events_after(conn, before, "manual_edit") == []
+235
View File
@@ -0,0 +1,235 @@
from __future__ import annotations
from chat.db.connection import open_db
from chat.db.migrate import apply_migrations
from chat.eventlog.log import append_and_apply, append_event
from chat.eventlog.projector import project
import chat.state.entities # registers handlers
import chat.state.world # registers handlers
import chat.state.group_node # registers handlers
import chat.state.events # registers handlers
from chat.state.events import (
get_event,
list_active_events,
list_events_in_status,
)
def _bot_payload(bot_id: str, name: str) -> dict:
return {
"id": bot_id,
"name": name,
"persona": "thoughtful, observant",
"voice_samples": [],
"traits": [],
"backstory": "",
"initial_relationship_to_you": "coworker",
"kickoff_prose": "",
}
def _chat_payload(chat_id: str = "chat_bot_a") -> dict:
return {
"id": chat_id,
"host_bot_id": "bot_a",
"guest_bot_id": "bot_b",
"initial_time": "2026-04-26T20:00:00+00:00",
"narrative_anchor": "Day 1 evening",
"weather": "clear",
}
def _seed_chat(conn) -> None:
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB"))
append_event(conn, kind="chat_created", payload=_chat_payload())
def test_event_planned_creates_row(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_chat(conn)
append_event(
conn,
kind="event_planned",
payload={
"event_id": "evt_abc",
"chat_id": "chat_bot_a",
"kind": "date_at_park",
"props": {"location": "park"},
"planned_for": "2026-04-30T18:00:00+00:00",
},
)
project(conn)
ev = get_event(conn, "evt_abc")
assert ev is not None
assert ev["event_id"] == "evt_abc"
assert ev["chat_id"] == "chat_bot_a"
assert ev["kind"] == "date_at_park"
assert ev["status"] == "planned"
assert ev["props"]["location"] == "park"
assert ev["planned_for"] == "2026-04-30T18:00:00+00:00"
assert ev["started_at"] is None
assert ev["completed_at"] is None
def test_event_started_then_completed_updates_status(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_chat(conn)
append_event(
conn,
kind="event_planned",
payload={
"event_id": "evt_abc",
"chat_id": "chat_bot_a",
"kind": "date_at_park",
"props": {},
"planned_for": "2026-04-30T18:00:00+00:00",
},
)
append_event(
conn,
kind="event_started",
payload={
"event_id": "evt_abc",
"started_at": "2026-04-30T18:01:00+00:00",
},
)
append_event(
conn,
kind="event_completed",
payload={
"event_id": "evt_abc",
"completed_at": "2026-04-30T20:00:00+00:00",
},
)
project(conn)
ev = get_event(conn, "evt_abc")
assert ev is not None
assert ev["status"] == "completed"
assert ev["started_at"] == "2026-04-30T18:01:00+00:00"
assert ev["completed_at"] == "2026-04-30T20:00:00+00:00"
def test_event_cancelled_terminal_subsequent_transitions_ignored(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_chat(conn)
append_event(
conn,
kind="event_planned",
payload={
"event_id": "evt_abc",
"chat_id": "chat_bot_a",
"kind": "date_at_park",
"props": {},
"planned_for": "2026-04-30T18:00:00+00:00",
},
)
append_event(
conn,
kind="event_cancelled",
payload={
"event_id": "evt_abc",
"completed_at": "2026-04-30T17:00:00+00:00",
},
)
project(conn)
ev = get_event(conn, "evt_abc")
assert ev is not None
assert ev["status"] == "cancelled"
assert ev["completed_at"] == "2026-04-30T17:00:00+00:00"
# Subsequent event_started must be no-oped because status is terminal.
# Use append_and_apply so we apply ONLY this new event without
# replaying earlier non-idempotent handlers (e.g. chat_created).
append_and_apply(
conn,
kind="event_started",
payload={
"event_id": "evt_abc",
"started_at": "2026-04-30T18:01:00+00:00",
},
)
ev2 = get_event(conn, "evt_abc")
assert ev2 is not None
assert ev2["status"] == "cancelled"
assert ev2["started_at"] is None
# completed_at unchanged from the cancelled transition
assert ev2["completed_at"] == "2026-04-30T17:00:00+00:00"
def test_list_active_events_filters_to_planned_and_active(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_chat(conn)
# Four events: one planned, one active, one completed, one cancelled.
for ev_id, kind in [
("evt_planned", "date_at_park"),
("evt_active", "movie_night"),
("evt_done", "dinner"),
("evt_canx", "trip"),
]:
append_event(
conn,
kind="event_planned",
payload={
"event_id": ev_id,
"chat_id": "chat_bot_a",
"kind": kind,
"props": {},
"planned_for": "2026-04-30T18:00:00+00:00",
},
)
append_event(
conn,
kind="event_started",
payload={
"event_id": "evt_active",
"started_at": "2026-04-30T18:01:00+00:00",
},
)
append_event(
conn,
kind="event_started",
payload={
"event_id": "evt_done",
"started_at": "2026-04-30T18:01:00+00:00",
},
)
append_event(
conn,
kind="event_completed",
payload={
"event_id": "evt_done",
"completed_at": "2026-04-30T20:00:00+00:00",
},
)
append_event(
conn,
kind="event_cancelled",
payload={
"event_id": "evt_canx",
"completed_at": "2026-04-30T17:00:00+00:00",
},
)
project(conn)
active = list_active_events(conn, "chat_bot_a")
active_ids = {e["event_id"] for e in active}
assert active_ids == {"evt_planned", "evt_active"}
completed = list_events_in_status(conn, "chat_bot_a", "completed")
assert [e["event_id"] for e in completed] == ["evt_done"]
cancelled = list_events_in_status(conn, "chat_bot_a", "cancelled")
assert [e["event_id"] for e in cancelled] == ["evt_canx"]
+34
View File
@@ -125,3 +125,37 @@ def test_search_invalid_witness_role_raises(tmp_path):
with open_db(db) as conn:
with pytest.raises(ValueError):
search_memories(conn, "bot_a", "invalid_role", "anything", k=4)
def test_higher_significance_outranks_equal_rank(tmp_path):
"""T57: significance multiplier biases the SQL ORDER BY.
Two memories with IDENTICAL FTS-matching text yield (effectively) equal
BM25 ranks. The significance bias applied in the SQL ORDER BY must
surface the higher-significance row first.
"""
db = tmp_path / "t.db"
_seed(
db,
memory_specs=[
# Identical pov_summary text -> FTS BM25 rank is the same for both.
{"pov_summary": "she swore an oath", "significance": 0},
{"pov_summary": "she swore an oath", "significance": 3},
],
)
with open_db(db) as conn:
out = search_memories(conn, "bot_a", "host", "oath", k=5)
assert len(out) == 2
# Higher significance wins despite tied FTS rank.
assert out[0]["significance"] == 3
assert out[1]["significance"] == 0
def test_significance_bias_is_constant_module_level():
"""T57: pin ``SIGNIFICANCE_RANK_BIAS`` as a tunable module-level numeric."""
from chat.state.memory import SIGNIFICANCE_RANK_BIAS
assert isinstance(SIGNIFICANCE_RANK_BIAS, (int, float))
# Must be non-negative -- a negative bias would invert the desired
# "higher significance ranks higher" semantics.
assert SIGNIFICANCE_RANK_BIAS >= 0
+255 -5
View File
@@ -504,13 +504,15 @@ async def test_close_with_no_guest_matches_phase1(tmp_path):
"relationship_summary": "BotA leaned in supportively.",
}
)
no_threads = json.dumps({"candidates": []})
with open_db(db) as conn:
_seed_single_bot_scene(conn)
project(conn)
# canned has 2 entries to detect any over-call; the assertion below
# confirms only one was consumed.
client = MockLLMClient(canned=[canned, canned])
# 1 host-POV entry + 1 thread-detection entry (T58.2) + 1 spare
# to detect any over-call. Assertion below confirms exactly two
# were consumed.
client = MockLLMClient(canned=[canned, no_threads, canned])
await apply_scene_close_summary(
conn,
client,
@@ -520,8 +522,8 @@ async def test_close_with_no_guest_matches_phase1(tmp_path):
host_bot_id="bot_a",
)
# Exactly one classifier call -> exactly one canned entry consumed,
# leaving the second untouched.
# Host POV + thread detection -> exactly two canned entries
# consumed, leaving the spare untouched.
assert len(client._canned) == 1
# Host memory rewritten with the per-POV summary content.
@@ -845,3 +847,251 @@ async def test_group_summary_skipped_when_no_guest(tmp_path):
"SELECT 1 FROM event_log WHERE kind = 'group_node_updated'"
).fetchall()
assert rows == []
# ---------------------------------------------------------------------------
# T58: significance-driven quote retention + thread detection on close.
# ---------------------------------------------------------------------------
def _seed_single_bot_scene_no_memory(conn) -> None:
"""Like ``_seed_single_bot_scene`` but skips the memory_written event so
callers can seed memories with custom significance / text themselves."""
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
append_event(
conn,
kind="you_authored",
payload={"name": "Me", "pronouns": "they/them", "persona": "engineer"},
)
append_event(
conn,
kind="chat_created",
payload={
"id": "chat_bot_a",
"host_bot_id": "bot_a",
"initial_time": "2026-04-26T20:00:00+00:00",
"narrative_anchor": "Day 1",
"weather": "",
},
)
append_event(
conn,
kind="container_created",
payload={
"chat_id": "chat_bot_a",
"name": "office",
"type": "workplace",
"properties": {},
},
)
append_event(
conn,
kind="scene_opened",
payload={
"chat_id": "chat_bot_a",
"container_id": 1,
"started_at": "2026-04-26T20:00:00+00:00",
"participants": ["you", "bot_a"],
},
)
append_event(
conn,
kind="edge_update",
payload={
"source_id": "bot_a",
"target_id": "you",
"chat_id": "chat_bot_a",
},
)
append_event(
conn,
kind="user_turn",
payload={
"chat_id": "chat_bot_a",
"prose": "Quick chat about the deadline",
"segments": [],
},
)
append_event(
conn,
kind="assistant_turn",
payload={
"chat_id": "chat_bot_a",
"speaker_id": "bot_a",
"text": "It's going to be okay.",
"truncated": False,
"user_turn_id": 1,
},
)
def _seed_memory(conn, *, pov_summary: str, significance: int) -> None:
append_event(
conn,
kind="memory_written",
payload={
"owner_id": "bot_a",
"chat_id": "chat_bot_a",
"scene_id": 1,
"pov_summary": pov_summary,
"witness_you": 1,
"witness_host": 1,
"witness_guest": 0,
"significance": significance,
},
)
@pytest.mark.asyncio
async def test_low_significance_scene_omits_quotes(tmp_path):
"""When the scene's max-turn-significance is < 2, the per-POV summary
rewrite collapses fully — no "Key quotes:" suffix is appended."""
db = tmp_path / "t.db"
apply_migrations(db)
canned = json.dumps(
{
"summary": "BotA had a low-key chat with you.",
"knowledge_facts": [],
"relationship_summary": "Nothing major shifted.",
}
)
no_threads = json.dumps({"candidates": []})
with open_db(db) as conn:
_seed_single_bot_scene_no_memory(conn)
_seed_memory(conn, pov_summary="Maya rambled about coffee", significance=1)
_seed_memory(conn, pov_summary="Maya glanced at the clock", significance=0)
project(conn)
client = MockLLMClient(canned=[canned, no_threads])
await apply_scene_close_summary(
conn,
client,
classifier_model="x",
chat_id="chat_bot_a",
scene_id=1,
host_bot_id="bot_a",
)
rows = conn.execute(
"SELECT pov_summary FROM memories WHERE scene_id = 1"
).fetchall()
assert rows
for (pov,) in rows:
assert "Key quotes:" not in pov
assert "BotA had a low-key chat" in pov
@pytest.mark.asyncio
async def test_high_significance_scene_includes_top_3_quotes(tmp_path):
"""When max-turn-significance is >= 2, each per-POV summary text gains
a "Key quotes:" suffix listing the top-3 highest-significance memory
rows verbatim, ordered by (significance DESC, id ASC)."""
db = tmp_path / "t.db"
apply_migrations(db)
canned = json.dumps(
{
"summary": "BotA had a heavy talk with you.",
"knowledge_facts": [],
"relationship_summary": "Things shifted.",
}
)
no_threads = json.dumps({"candidates": []})
with open_db(db) as conn:
_seed_single_bot_scene_no_memory(conn)
# Insertion order matches id ASC. Top-3 by (sig DESC, id ASC):
# quote 1 (sig 3) -> quote 2 (sig 2, lower id) -> quote 4 (sig 2,
# higher id). quote 3 (sig 1) is dropped.
_seed_memory(conn, pov_summary="Maya quote one", significance=3)
_seed_memory(conn, pov_summary="Maya quote two", significance=2)
_seed_memory(conn, pov_summary="Maya quote three", significance=1)
_seed_memory(conn, pov_summary="Maya quote four", significance=2)
project(conn)
client = MockLLMClient(canned=[canned, no_threads])
await apply_scene_close_summary(
conn,
client,
classifier_model="x",
chat_id="chat_bot_a",
scene_id=1,
host_bot_id="bot_a",
)
rows = conn.execute(
"SELECT pov_summary FROM memories WHERE scene_id = 1"
).fetchall()
assert rows
for (pov,) in rows:
assert "Key quotes:" in pov
assert '"Maya quote one"' in pov
assert '"Maya quote two"' in pov
assert '"Maya quote four"' in pov
# The sig-1 quote falls outside the top-3 cap.
assert '"Maya quote three"' not in pov
# Ordering: sig 3 first, then the two sig-2s by id ASC.
i_one = pov.index('"Maya quote one"')
i_two = pov.index('"Maya quote two"')
i_four = pov.index('"Maya quote four"')
assert i_one < i_two < i_four
@pytest.mark.asyncio
async def test_thread_detection_emits_events(tmp_path, monkeypatch):
"""On scene close, ``detect_threads`` is invoked and each "open"
candidate yields a ``thread_opened`` event with a fresh thread_id."""
from chat.services import thread_detection as td_mod
canned = json.dumps(
{
"summary": "BotA noticed something unresolved.",
"knowledge_facts": [],
"relationship_summary": "Tension lingered.",
}
)
async def fake_detect_threads(client, **kwargs):
return td_mod.ThreadDetectionResult(
candidates=[
td_mod.ThreadCandidate(
action="open",
title="Test thread",
summary="A test",
existing_thread_id=None,
),
]
)
monkeypatch.setattr(td_mod, "detect_threads", fake_detect_threads)
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_single_bot_scene(conn)
project(conn)
client = MockLLMClient(canned=[canned])
await apply_scene_close_summary(
conn,
client,
classifier_model="x",
chat_id="chat_bot_a",
scene_id=1,
host_bot_id="bot_a",
)
rows = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'thread_opened'"
).fetchall()
assert len(rows) == 1
payload = json.loads(rows[0][0])
assert payload["title"] == "Test thread"
assert payload["summary"] == "A test"
assert payload["chat_id"] == "chat_bot_a"
assert payload["thread_id"].startswith("thr_")
# The threads-table projection ran via append_and_apply.
from chat.state.threads import list_open_threads
open_threads = list_open_threads(conn, "chat_bot_a")
assert len(open_threads) == 1
assert open_threads[0]["title"] == "Test thread"
+117
View File
@@ -0,0 +1,117 @@
"""Skip narration service tests (T53).
The skip-narration service generates short transition prose between an
in-progress moment and a post-skip moment. Two flavors:
* ``elision`` — collapses an in-progress activity to its expected
end-state in 1-2 sentences (e.g. "skip ahead to when we arrive").
* ``jump`` — bridges a longer fiction-time delta in 2-3 sentences
(e.g. "next morning", "a week later").
Output is free-form prose, not structured JSON, so the service goes
through ``client.generate`` directly rather than the classifier path.
A deterministic template fallback fires on any LLM failure so the skip
flow never blocks even when the model is down.
"""
from __future__ import annotations
from typing import AsyncIterator, Sequence
import pytest
from chat.llm.client import Message
from chat.llm.mock import MockLLMClient
from chat.services.skip_narration import narrate_skip
_SPEAKER = {
"id": "bot1",
"name": "Aria",
"persona": "thoughtful, observant",
}
@pytest.mark.asyncio
async def test_narrate_elision_returns_classifier_output():
canned = (
"She closes her laptop and slings her bag over her shoulder. "
"The office shrinks behind her as she steps into the late "
"afternoon light."
)
mock = MockLLMClient(canned=[canned])
result = await narrate_skip(
mock,
narrative_model="x",
skip_kind="elision",
speaker_bot=_SPEAKER,
you_name="Me",
current_time="3:42 PM",
new_time="5:10 PM",
current_activity="finishing up at her desk",
landing_state_hint="walking out into the parking lot",
)
assert "office" in result or result == canned
@pytest.mark.asyncio
async def test_narrate_jump_returns_classifier_output():
canned = (
"Morning light spills through the kitchen window. The coffee "
"maker hums. She's already at the table, scrolling her phone."
)
mock = MockLLMClient(canned=[canned])
result = await narrate_skip(
mock,
narrative_model="x",
skip_kind="jump",
speaker_bot=_SPEAKER,
you_name="Me",
current_time="late evening",
new_time="next morning",
current_activity="winding down for the night",
landing_state_hint="having coffee in the kitchen",
)
assert result
lower = result.lower()
assert "morning" in lower or "coffee" in lower
class _RaisingMock:
"""Mock LLMClient whose ``generate`` always raises.
``MockLLMClient.generate`` raises ``IndexError`` once the canned
list is empty, but the test wants a clear, unambiguous failure
regardless of canned-list state, so we ship a tiny dedicated mock
instead.
"""
async def generate(
self, messages: Sequence[Message], *, model: str, **params
) -> str:
raise RuntimeError("LLM is down")
async def stream(
self, messages: Sequence[Message], *, model: str, **params
) -> AsyncIterator[str]:
raise RuntimeError("LLM is down")
yield # pragma: no cover - make this a generator
@pytest.mark.asyncio
async def test_narrate_falls_back_on_generation_failure():
new_time = "next morning"
result = await narrate_skip(
_RaisingMock(),
narrative_model="x",
skip_kind="jump",
speaker_bot=_SPEAKER,
you_name="Me",
current_time="late evening",
new_time=new_time,
current_activity="winding down for the night",
landing_state_hint="having coffee in the kitchen",
)
# Fallback template includes the new_time so callers can see *what*
# we skipped to even when the LLM never answered.
assert new_time in result
+98
View File
@@ -0,0 +1,98 @@
"""Tests for the synthesized-memories service (T54).
When the user jump-skips ("a week later") they are prompted "anything
notable happen?" If they answer with prose, this service parses it into
1-N synthesized memories per present bot. Each memory carries
``source="synthesized"`` and ``reliability=0.7`` (the caller — T62 skip
flow — applies those tags when persisting; this service just produces
the structured digest).
These tests cover:
* The happy path: a canned classifier response parses cleanly into a
populated :class:`SynthesizedDigest` with one memory.
* Empty prose short-circuits before any classifier call — the mock has
no canned responses, so an accidental call would raise
``IndexError``.
* Classifier failure (3 bad responses, exhausting :func:`classify`'s
retry budget) falls back to an empty default digest.
"""
from __future__ import annotations
import json
import pytest
from chat.llm.mock import MockLLMClient
from chat.services.synthesized_memories import (
SynthesizedDigest,
SynthesizedMemory,
synthesize_memories,
)
@pytest.mark.asyncio
async def test_synthesize_parses_canned_prose():
canned = json.dumps(
{
"memories": [
{
"text": "Maya started a new pottery class.",
"significance": 1,
"affinity_delta": 0,
"trust_delta": 0,
}
]
}
)
mock = MockLLMClient(canned=[canned])
result = await synthesize_memories(
mock,
classifier_model="x",
prose="we saw each other at her pottery class once",
bot_name="Maya",
bot_persona="warm potter, mid-30s",
you_name="Sam",
)
assert isinstance(result, SynthesizedDigest)
assert len(result.memories) == 1
mem = result.memories[0]
assert isinstance(mem, SynthesizedMemory)
assert mem.text == "Maya started a new pottery class."
assert mem.significance == 1
assert mem.affinity_delta == 0
assert mem.trust_delta == 0
@pytest.mark.asyncio
async def test_empty_prose_returns_empty_digest():
"""Empty prose short-circuits — the classifier must not be called."""
mock = MockLLMClient(canned=[])
result = await synthesize_memories(
mock,
classifier_model="x",
prose="",
bot_name="Maya",
bot_persona="warm potter, mid-30s",
you_name="Sam",
)
assert result == SynthesizedDigest()
assert result.memories == []
@pytest.mark.asyncio
async def test_classifier_failure_returns_empty_default():
"""Three bad responses exhaust the classifier's retry budget; the
service then returns the empty default digest."""
mock = MockLLMClient(canned=["bad", "bad", "bad"])
result = await synthesize_memories(
mock,
classifier_model="x",
prose="we saw each other at her pottery class once",
bot_name="Maya",
bot_persona="warm potter, mid-30s",
you_name="Sam",
)
assert result == SynthesizedDigest()
assert result.memories == []
+128
View File
@@ -0,0 +1,128 @@
"""Tests for the thread-detection service (T55).
On scene close, the transcript is classified to detect open threads
(unresolved arcs, dangling questions, promises made). The service can
also signal **update** to an existing thread when the scene developed
it, or **close** when the scene resolved it.
These tests cover:
* A new thread the scene introduced — action="open" with a fresh title.
* An update to an existing thread — action="update" with
``existing_thread_id`` referencing the prior thread.
* Classifier failure — three bad responses degrade to an empty
candidates list (graceful degradation, §3.3).
* Empty transcript short-circuits before any classifier call.
"""
from __future__ import annotations
import json
import pytest
from chat.llm.mock import MockLLMClient
from chat.services.thread_detection import (
ThreadCandidate,
ThreadDetectionResult,
detect_threads,
)
@pytest.mark.asyncio
async def test_detects_new_thread_open():
canned = json.dumps(
{
"candidates": [
{
"action": "open",
"title": "Maya's job hunt",
"summary": "Maya is looking for a new job",
"existing_thread_id": None,
}
]
}
)
mock = MockLLMClient(canned=[canned])
result = await detect_threads(
mock,
classifier_model="x",
scene_transcript=[
{"speaker": "Maya", "text": "I need to find a new job soon."},
{"speaker": "Sam", "text": "What kind of role are you looking for?"},
],
open_threads=[],
)
assert isinstance(result, ThreadDetectionResult)
assert len(result.candidates) == 1
cand = result.candidates[0]
assert isinstance(cand, ThreadCandidate)
assert cand.action == "open"
assert cand.title == "Maya's job hunt"
assert cand.summary == "Maya is looking for a new job"
assert cand.existing_thread_id is None
@pytest.mark.asyncio
async def test_detects_update_to_existing_thread():
canned = json.dumps(
{
"candidates": [
{
"action": "update",
"title": "",
"summary": "Maya interviewed at Acme today",
"existing_thread_id": "thr_jobhunt",
}
]
}
)
mock = MockLLMClient(canned=[canned])
result = await detect_threads(
mock,
classifier_model="x",
scene_transcript=[
{"speaker": "Maya", "text": "I had the Acme interview today."},
{"speaker": "Sam", "text": "How did it go?"},
],
open_threads=[
{
"thread_id": "thr_jobhunt",
"title": "Maya's job hunt",
"summary": "Maya is looking for a new job",
}
],
)
assert len(result.candidates) == 1
cand = result.candidates[0]
assert cand.action == "update"
assert cand.existing_thread_id == "thr_jobhunt"
assert cand.summary == "Maya interviewed at Acme today"
@pytest.mark.asyncio
async def test_classifier_failure_returns_empty():
"""Three malformed classifier responses → empty candidates list."""
mock = MockLLMClient(canned=["not json", "still not json", "{bad"])
result = await detect_threads(
mock,
classifier_model="x",
scene_transcript=[
{"speaker": "Maya", "text": "Anything could happen here."},
],
open_threads=[],
)
assert result.candidates == []
@pytest.mark.asyncio
async def test_empty_transcript_short_circuits():
"""Empty transcript short-circuits — classifier must not be called."""
mock = MockLLMClient(canned=[])
result = await detect_threads(
mock,
classifier_model="x",
scene_transcript=[],
open_threads=[],
)
assert result.candidates == []
+181
View File
@@ -0,0 +1,181 @@
from __future__ import annotations
from chat.db.connection import open_db
from chat.db.migrate import apply_migrations
from chat.eventlog.log import append_and_apply, append_event
from chat.eventlog.projector import project
import chat.state.entities # registers handlers
import chat.state.world # registers handlers
import chat.state.threads # registers handlers
from chat.state.threads import get_thread, list_open_threads
def _bot_payload(bot_id: str, name: str) -> dict:
return {
"id": bot_id,
"name": name,
"persona": "thoughtful, observant",
"voice_samples": [],
"traits": [],
"backstory": "",
"initial_relationship_to_you": "coworker",
"kickoff_prose": "",
}
def _chat_payload(chat_id: str = "chat_bot_a") -> dict:
return {
"id": chat_id,
"host_bot_id": "bot_a",
"guest_bot_id": "bot_b",
"initial_time": "2026-04-26T20:00:00+00:00",
"narrative_anchor": "Day 1 evening",
"weather": "clear",
}
def test_thread_opened_creates_row(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
append_event(conn, kind="chat_created", payload=_chat_payload())
append_event(
conn,
kind="thread_opened",
payload={
"thread_id": "thr_abc",
"chat_id": "chat_bot_a",
"title": "Maya's job hunt",
"summary": "Maya is looking for a new job",
},
)
project(conn)
t = get_thread(conn, "thr_abc")
assert t is not None
assert t["thread_id"] == "thr_abc"
assert t["chat_id"] == "chat_bot_a"
assert t["title"] == "Maya's job hunt"
assert t["summary"] == "Maya is looking for a new job"
assert t["status"] == "open"
assert t["closed_at"] is None
assert t["last_referenced_scene_id"] is None
def test_thread_updated_changes_summary_and_last_referenced(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
append_event(conn, kind="chat_created", payload=_chat_payload())
append_event(
conn,
kind="thread_opened",
payload={
"thread_id": "thr_abc",
"chat_id": "chat_bot_a",
"title": "Maya's job hunt",
"summary": "Maya is looking for a new job",
},
)
append_event(
conn,
kind="thread_updated",
payload={
"thread_id": "thr_abc",
"summary": "Maya landed an interview at a startup",
"last_referenced_scene_id": 42,
},
)
project(conn)
t = get_thread(conn, "thr_abc")
assert t is not None
assert t["summary"] == "Maya landed an interview at a startup"
assert t["last_referenced_scene_id"] == 42
assert t["status"] == "open"
def test_thread_closed_terminal(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
append_event(conn, kind="chat_created", payload=_chat_payload())
append_event(
conn,
kind="thread_opened",
payload={
"thread_id": "thr_abc",
"chat_id": "chat_bot_a",
"title": "Maya's job hunt",
"summary": "Maya is looking for a new job",
},
)
append_event(
conn,
kind="thread_closed",
payload={
"thread_id": "thr_abc",
"closed_at": "2026-04-26T21:00:00+00:00",
},
)
project(conn)
t = get_thread(conn, "thr_abc")
assert t is not None
assert t["status"] == "closed"
assert t["closed_at"] == "2026-04-26T21:00:00+00:00"
# Subsequent updates to a closed thread are no-ops.
append_and_apply(
conn,
kind="thread_updated",
payload={
"thread_id": "thr_abc",
"summary": "should not be applied",
},
)
t2 = get_thread(conn, "thr_abc")
assert t2 is not None
assert t2["summary"] == "Maya is looking for a new job"
assert t2["status"] == "closed"
def test_list_open_threads_filters_to_open_only(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
append_event(conn, kind="chat_created", payload=_chat_payload())
for tid, title in [
("thr_1", "Arc 1"),
("thr_2", "Arc 2"),
("thr_3", "Arc 3"),
]:
append_event(
conn,
kind="thread_opened",
payload={
"thread_id": tid,
"chat_id": "chat_bot_a",
"title": title,
"summary": "",
},
)
append_event(
conn,
kind="thread_closed",
payload={
"thread_id": "thr_3",
"closed_at": "2026-04-26T21:00:00+00:00",
},
)
project(conn)
open_threads = list_open_threads(conn, "chat_bot_a")
assert len(open_threads) == 2
ids = {t["thread_id"] for t in open_threads}
assert ids == {"thr_1", "thr_2"}
+132
View File
@@ -0,0 +1,132 @@
from __future__ import annotations
from chat.db.connection import open_db
from chat.db.migrate import apply_migrations
from chat.eventlog.log import append_event
from chat.eventlog.projector import project
import chat.state.world # registers handlers
from chat.state.world import get_activity, get_chat
def _chat_payload(**overrides):
payload = {
"id": "chat_bot_a",
"host_bot_id": "bot_a",
"guest_bot_id": None,
"initial_time": "2026-04-26T20:00:00+00:00",
"narrative_anchor": "Day 1 evening",
"weather": "clear",
}
payload.update(overrides)
return payload
def _container_payload(**overrides):
payload = {
"chat_id": "chat_bot_a",
"name": "office",
"type": "workplace",
"properties": {
"public": True,
"moving": False,
"audible_range": "normal",
"slots": [],
},
"parent_id": None,
}
payload.update(overrides)
return payload
def _activity_payload(**overrides):
payload = {
"entity_id": "bot_a",
"container_id": 1,
"slot": "desk_chair",
"posture": "sitting",
"action": {"verb": "writing email"},
"attention": "the screen",
"holding": ["pen"],
"status": {"hungry": False},
}
payload.update(overrides)
return payload
def _seed_events(conn):
"""Append seed events but do NOT project — caller appends more then projects once."""
append_event(conn, kind="chat_created", payload=_chat_payload())
append_event(conn, kind="container_created", payload=_container_payload())
append_event(conn, kind="activity_change", payload=_activity_payload())
def test_elision_advances_chat_clock_only(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_events(conn)
append_event(conn, kind="time_skip_elision", payload={
"chat_id": "chat_bot_a",
"new_time": "2026-04-26T20:30:00+00:00",
})
project(conn)
chat = get_chat(conn, "chat_bot_a")
assert chat["time"] == "2026-04-26T20:30:00+00:00"
# Activity row preserved with the same fields it was seeded with.
a = get_activity(conn, "bot_a")
assert a is not None
assert a["entity_id"] == "bot_a"
assert a["container_id"] == 1
assert a["slot"] == "desk_chair"
assert a["posture"] == "sitting"
assert a["action"] == {"verb": "writing email"}
assert a["attention"] == "the screen"
assert a["holding"] == ["pen"]
assert a["status"] == {"hungry": False}
def test_jump_with_reset_clears_activity(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_events(conn)
append_event(conn, kind="time_skip_jump", payload={
"chat_id": "chat_bot_a",
"new_time": "2026-04-27T08:00:00+00:00",
"reset_activity": True,
})
project(conn)
chat = get_chat(conn, "chat_bot_a")
assert chat["time"] == "2026-04-27T08:00:00+00:00"
count = conn.execute(
"SELECT COUNT(*) FROM activity "
"WHERE container_id IN (SELECT id FROM containers WHERE chat_id = ?)",
("chat_bot_a",),
).fetchone()[0]
assert count == 0
assert get_activity(conn, "bot_a") is None
def test_jump_without_reset_preserves_activity(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_events(conn)
append_event(conn, kind="time_skip_jump", payload={
"chat_id": "chat_bot_a",
"new_time": "2026-04-27T08:00:00+00:00",
"reset_activity": False,
})
project(conn)
chat = get_chat(conn, "chat_bot_a")
assert chat["time"] == "2026-04-27T08:00:00+00:00"
a = get_activity(conn, "bot_a")
assert a is not None
assert a["posture"] == "sitting"
assert a["action"]["verb"] == "writing email"
+2 -2
View File
@@ -324,11 +324,11 @@ def test_get_scene_returns_none_for_missing(tmp_path):
assert active_scene(conn, "chat_missing") is None
def test_schema_version_after_migration_is_8(tmp_path):
def test_schema_version_after_migration_is_10(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
row = conn.execute(
"SELECT value FROM meta WHERE key = 'schema_version'"
).fetchone()
assert int(row[0]) == 8
assert int(row[0]) == 10