32 Commits

Author SHA1 Message Date
Joseph Doherty 819803da84 merge: T62 natural-language skip command flow + shared skip controllers 2026-04-26 20:47:07 -04:00
Joseph Doherty a7eedb8037 feat: natural-language skip detection + skip command flow (T62)
Extend ParsedTurn with intent/landing_state_hint so the classifier can
flag skip-elision and skip-jump prose. The post_turn handler short-
circuits the regular narrative path when intent != "narrative":
elision runs through the shared controller in chat/web/skip.py;
jump returns 422 directing the user to the drawer's structured form
(simpler Phase 3 path — natural-language fiction-time delta parsing
is too fragile for v1 without a structured surface).

Extract the elision/jump logic that previously lived in drawer.py
into chat/web/skip.py so both the drawer T59 routes and the new
natural-language path share one canonical implementation. The drawer
routes become thin HTTP wrappers that translate ValueError to 400
and refresh the drawer partial; the existing drawer skip tests pass
unchanged.

The new natural-language elision derives ``new_time`` by bumping the
chat clock by 1 hour (Phase 3 stub) — the drawer's structured form
remains the path for picking a specific landing time.
2026-04-26 20:45:05 -04:00
Joseph Doherty e236bcadcd merge: T61 per-turn event-lifecycle detection + completion promotion 2026-04-26 20:37:21 -04:00
Joseph Doherty 3678bcaca6 merge: T60 prompt assembly active events + open threads 2026-04-26 20:37:21 -04:00
Joseph Doherty b582567521 feat: per-turn event-lifecycle detection + completion promotion (T61) 2026-04-26 20:35:34 -04:00
Joseph Doherty 21c4ffa63c feat: prompt assembly renders active events + open threads (T60) 2026-04-26 20:34:26 -04:00
Joseph Doherty 83f94a4325 merge: T59 drawer events / threads / skip controls 2026-04-26 20:29:40 -04:00
Joseph Doherty 2d14197553 feat: drawer events / threads / skip controls (T59) 2026-04-26 20:27:47 -04:00
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
33 changed files with 4449 additions and 26 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"]
+127 -5
View File
@@ -37,8 +37,10 @@ import tiktoken
from chat.llm.client import Message
from chat.state.edges import get_edge, list_edges_for
from chat.state.entities import get_bot, get_you
from chat.state.events import list_active_events
from chat.state.group_node import get_group_node
from chat.state.memory import search_memories
from chat.state.threads import list_open_threads
from chat.state.world import (
active_scene,
get_activity,
@@ -227,6 +229,76 @@ def _build_group_node_block(group_node: dict | None) -> str | None:
return "\n".join(lines)
def _props_excerpt(props: dict | None, limit: int = 80) -> str:
"""Return a one-line excerpt of an event's ``props`` dict.
Renders ``key=value`` pairs separated by ", " (deterministic by dict
insertion order) and truncates to ~``limit`` characters with a
trailing ellipsis. Returns empty string for falsy/empty props so the
caller can omit the line entirely.
"""
if not props:
return ""
pieces: list[str] = []
for k, v in props.items():
pieces.append(f"{k}={v}")
rendered = ", ".join(pieces)
if len(rendered) > limit:
# Reserve 1 char for the ellipsis so the total never exceeds limit.
rendered = rendered[: max(0, limit - 1)] + ""
return rendered
def _build_active_events_block(events: list[dict]) -> str | None:
"""Render the ``Active events:`` block for Phase 3 Task 60.
One bullet per event. The sub-label depends on status:
- ``planned`` → ``(planned for {planned_for})``
- ``active`` → ``(active, started_at={started_at})``
A second indented line carries a one-line excerpt of the event's
``props`` (truncated ~80 chars) when non-empty. Returns ``None`` when
there are no active events so the caller can omit the entire block.
"""
if not events:
return None
lines = ["Active events:"]
for ev in events:
kind = ev.get("kind") or "?"
status = ev.get("status") or ""
if status == "active":
started = ev.get("started_at") or ""
lines.append(f"- {kind} (active, started_at={started})")
else:
planned = ev.get("planned_for") or ""
lines.append(f"- {kind} (planned for {planned})")
excerpt = _props_excerpt(ev.get("props"))
if excerpt:
lines.append(f" {excerpt}")
return "\n".join(lines)
def _build_open_threads_block(threads: list[dict]) -> str | None:
"""Render the ``Open threads:`` block for Phase 3 Task 60.
One bullet per thread, formatted as ``- {title}: {summary}`` with the
summary truncated to ~120 characters. Returns ``None`` when there are
no open threads so the caller can omit the entire block.
"""
if not threads:
return None
lines = ["Open threads:"]
for t in threads:
title = t.get("title") or "?"
summary = t.get("summary") or ""
if len(summary) > 120:
summary = summary[:119] + ""
if summary:
lines.append(f"- {title}: {summary}")
else:
lines.append(f"- {title}")
return "\n".join(lines)
def _closing_instruction(speaker_name: str, addressee_name: str) -> str:
return (
f"Continue the scene as {speaker_name}, in their voice, responding "
@@ -436,6 +508,17 @@ def assemble_narrative_prompt(
if required.issubset(members):
group_node_block = _build_group_node_block(gn)
# SHOULD-tier active events + open threads (Phase 3 / Task 60).
# Auto-detect both from the chat_id per the Phase 2 T43 precedent —
# no new function parameter. Both blocks are omit-when-empty so a
# Phase 1 chat with no events/threads renders identically to before.
active_events_block = _build_active_events_block(
list_active_events(conn, chat_id)
)
open_threads_block = _build_open_threads_block(
list_open_threads(conn, chat_id)
)
container = None
if chat.get("active_scene_id"):
scene = get_scene(conn, chat["active_scene_id"])
@@ -531,6 +614,8 @@ def assemble_narrative_prompt(
include_you_activity: bool = True,
include_guest_activity: bool = True,
include_group_node: bool = True,
include_active_events: bool = True,
include_open_threads: bool = True,
) -> tuple[str, int, list[dict]]:
# dialogue: keep the last `dialogue_keep` turns verbatim; older
# turns become an "earlier:" placeholder line.
@@ -566,6 +651,8 @@ def assemble_narrative_prompt(
scene_block,
activity_block,
group_node_block if include_group_node else None,
active_events_block if include_active_events else None,
open_threads_block if include_open_threads else None,
prev_block,
memories_block,
dialogue_block,
@@ -585,9 +672,12 @@ def assemble_narrative_prompt(
include_you_activity = you_activity is not None
include_guest_activity = guest_activity is not None
include_group_node = group_node_block is not None
include_active_events = active_events_block is not None
include_open_threads = open_threads_block is not None
def _build(*, prev: bool, mem_k: int, dlg: int, other: bool,
you_act: bool, guest_act: bool, group: bool) -> tuple[str, int]:
you_act: bool, guest_act: bool, group: bool,
events: bool, threads: bool) -> tuple[str, int]:
body, total, _ = assemble(
include_other_edges=other,
include_previous_scene=prev,
@@ -596,6 +686,8 @@ def assemble_narrative_prompt(
include_you_activity=you_act,
include_guest_activity=guest_act,
include_group_node=group,
include_active_events=events,
include_open_threads=threads,
)
return body, total
@@ -603,6 +695,7 @@ def assemble_narrative_prompt(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
# If under soft, we're done.
@@ -637,6 +730,7 @@ def assemble_narrative_prompt(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
if total <= budget_soft:
return _emit(body, user_turn_prose)
@@ -647,6 +741,7 @@ def assemble_narrative_prompt(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
if total <= budget_soft:
return _emit(body, user_turn_prose)
@@ -657,6 +752,7 @@ def assemble_narrative_prompt(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
if total <= budget_soft:
return _emit(body, user_turn_prose)
@@ -668,21 +764,44 @@ def assemble_narrative_prompt(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
# Drop SHOULD-tier extras in order:
# 1. guest activity bullet (T71.2: bullet-level trim within the
# 1. open threads block (T60: SHOULD-tier; least critical to the
# speaker's immediate voice — drop first among SHOULD)
# 2. active events block (T60: same tier, drops next)
# 3. guest activity bullet (T71.2: bullet-level trim within the
# single ACTIVITIES: block — guest goes first per Task 43 spec)
# 2. group node block
# 3. you activity bullet (still SHOULD-tier; speaker bullet is the
# 4. group node block
# 5. you activity bullet (still SHOULD-tier; speaker bullet is the
# MUST-tier floor and never dropped)
# 4. other edges
# 6. other edges
if include_open_threads and total > budget_hard:
include_open_threads = False
body, total = _build(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
if include_active_events and total > budget_hard:
include_active_events = False
body, total = _build(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
if include_guest_activity and total > budget_hard:
include_guest_activity = False
body, total = _build(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
if include_group_node and total > budget_hard:
@@ -691,6 +810,7 @@ def assemble_narrative_prompt(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
if include_you_activity and total > budget_hard:
@@ -699,6 +819,7 @@ def assemble_narrative_prompt(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
if include_other and total > budget_hard:
@@ -707,6 +828,7 @@ def assemble_narrative_prompt(
prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
other=include_other, you_act=include_you_activity,
guest_act=include_guest_activity, group=include_group_node,
events=include_active_events, threads=include_open_threads,
)
if total > budget_hard:
+64
View File
@@ -73,12 +73,15 @@ from sqlite3 import Connection
from chat.config import Settings
from chat.eventlog.log import append_and_apply, append_event
from chat.services.event_lifecycle import detect_event_transitions
from chat.services.event_promotion import promote_completed_event
from chat.services.interjection import detect_interjection
from chat.services.memory_write import record_turn_memory_for_present
from chat.services.multi_state_update import compute_state_updates_for_present
from chat.services.prompt import assemble_narrative_prompt
from chat.state.edges import get_edge
from chat.state.entities import get_bot, get_you
from chat.state.events import list_active_events
from chat.state.world import active_scene, get_chat
from chat.web.pubsub import publish
from chat.web.render import render_turn_html
@@ -617,6 +620,67 @@ async def regenerate_assistant_turn(
(new_assistant_event_id, original_interjection_event_id),
)
# 10. Event-lifecycle detection (Phase 3, T61). Mirrors the post_turn
# block: classify whether any active events transitioned in the
# regenerated narrative and append the corresponding event_started /
# event_completed / event_cancelled. ``promote_completed_event``
# runs inline after a completion so promotion artifacts land in the
# same regenerate path.
#
# Phase 3.5 follow-up: when a regenerate replaces a turn that had
# already produced event transitions, those original transitions are
# NOT undone here. The superseded ``assistant_turn`` group keeps its
# prior ``event_started`` / ``event_completed`` events in the log
# (they remain projected onto the events table). Phase 3.5 will add
# an "undo lifecycle" step to roll back the prior transitions before
# re-classifying the regenerated text. For v3 we accept that a
# regenerate-after-completion will double-emit promotion artifacts
# if the new text re-completes the same event — narratively rare,
# and a true fix needs the lifecycle-undo pass.
new_active_events = list_active_events(conn, chat_id)
if new_active_events:
lifecycle_decision = await detect_event_transitions(
client,
classifier_model=settings.classifier_model,
narrative_text=new_text,
active_events=new_active_events,
timeout_s=settings.classifier_timeout_s,
)
for transition in lifecycle_decision.transitions:
if transition.new_status == "active":
append_and_apply(
conn,
kind="event_started",
payload={
"event_id": transition.event_id,
"started_at": chat.get("time"),
},
)
elif transition.new_status == "completed":
append_and_apply(
conn,
kind="event_completed",
payload={
"event_id": transition.event_id,
"completed_at": chat.get("time"),
},
)
promote_completed_event(
conn,
event_id=transition.event_id,
chat_id=chat_id,
chat_clock_at=chat.get("time"),
)
elif transition.new_status == "cancelled":
append_and_apply(
conn,
kind="event_cancelled",
payload={
"event_id": transition.event_id,
"completed_at": chat.get("time"),
},
)
return new_text
+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"]
+39 -8
View File
@@ -16,6 +16,16 @@ nested quotes, mixed punctuation), so v1 delegates the segmentation to
the classifier. The configurable ``Settings.ooc_marker`` is *not* read
here: the classifier figures OOC out from ``((`` ``))`` regardless of
config-time choice; marker-based stripping is a downstream concern.
T62 extends the parser with an ``intent`` field so the turn flow can
short-circuit time-skip phrases before the regular narrative path.
``intent`` defaults to ``"narrative"``; the classifier may set it to
``"skip_elision"`` when prose like "skip to when we arrive" or
``"skip_jump"`` when prose like "next morning" / "a week later" is
detected. ``landing_state_hint`` carries the residual descriptor for
elision skips (the "to when we ..." phrase). Existing callers that
don't read ``intent`` continue to work because the default keeps the
narrative path intact.
"""
from __future__ import annotations
@@ -39,9 +49,19 @@ class TurnSegment(BaseModel):
class ParsedTurn(BaseModel):
"""A turn split into ordered, typed segments."""
"""A turn split into ordered, typed segments.
``intent`` distinguishes a regular narrative beat (the default) from
a natural-language time-skip command (T62). ``landing_state_hint``
captures the descriptor following "skip to when we ..." for elision
skips so the downstream skip controller can pass it to the
narration helper. Both fields are optional and default-empty so
older fixtures and tests that don't supply them keep working.
"""
segments: list[TurnSegment]
intent: str = "narrative" # "narrative" | "skip_elision" | "skip_jump"
landing_state_hint: str = ""
_SYSTEM_PROMPT = (
@@ -52,13 +72,24 @@ _SYSTEM_PROMPT = (
"- ((text in double parens)) is an OOC (out-of-character) segment — "
"the author talking to the system, not the in-fiction bot.\n\n"
"Output a JSON object with shape "
'{"segments": [{"kind": "...", "text": "..."}, ...]} '
"where each ``kind`` is exactly one of: dialogue, action, ooc. "
"Preserve the original substring text as ``text``: do not rewrite, "
"translate, or normalize punctuation — strip only the marker "
"characters (asterisks, surrounding quotes, double parens) so "
"``text`` is the inner content. Emit segments in the order they "
"appear in the input."
'{"segments": [{"kind": "...", "text": "..."}, ...], '
'"intent": "...", "landing_state_hint": "..."} '
"where each segment ``kind`` is exactly one of: dialogue, action, "
"ooc. Preserve the original substring text as ``text``: do not "
"rewrite, translate, or normalize punctuation — strip only the "
"marker characters (asterisks, surrounding quotes, double parens) "
"so ``text`` is the inner content. Emit segments in the order they "
"appear in the input.\n\n"
"``intent`` is exactly one of: narrative, skip_elision, skip_jump. "
"Default to ``narrative``. Use ``skip_elision`` when the prose is a "
"directive to fast-forward an in-progress activity to a near-term "
"landing state — e.g. 'skip to when we arrive', 'fast-forward to "
"after dinner'. Use ``skip_jump`` when the prose denotes a longer "
"fiction-time bridge — e.g. 'next morning', 'a week later', 'the "
"following day'.\n"
"``landing_state_hint`` is a short descriptor of the landing state "
"for ``skip_elision`` (e.g. 'we arrive at the park'). Empty string "
"for ``skip_jump`` and ``narrative``."
)
+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
+115
View File
@@ -41,6 +41,121 @@
{% endif %}
</div>
{% endfor %}
<details class="skip-controls">
<summary>Elision skip</summary>
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/skip/elision"
hx-target="#drawer" hx-swap="innerHTML">
<label>
Landing state hint:
<input type="text" name="landing_state_hint"
placeholder="e.g. arriving at the office">
</label>
<label>
New time (ISO 8601):
<input type="text" name="new_time" required
placeholder="2026-04-26T20:30:00+00:00">
</label>
<button type="submit">Skip ahead</button>
</form>
</details>
<details class="skip-controls">
<summary>Jump skip</summary>
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/skip/jump"
hx-target="#drawer" hx-swap="innerHTML">
<label>
New time (ISO 8601):
<input type="text" name="new_time" required
placeholder="2026-04-27T08:00:00+00:00">
</label>
<label>
Anything notable happen? (optional)
<textarea name="notable_prose" rows="3"
placeholder="leave blank to jump without synthesizing memories"></textarea>
</label>
<label>
<input type="checkbox" name="reset_activity" value="1">
Reset activity at landing
</label>
<button type="submit">Jump ahead</button>
</form>
</details>
</section>
<section class="drawer-section">
<h3>Events</h3>
{% if active_events %}
<ul class="event-list">
{% for ev in active_events %}
<li class="event-row">
<strong>{{ ev.kind }}</strong>
<span class="muted"> ({{ ev.status }})</span>
{% if ev.planned_for %}
<p class="muted">planned for: {{ ev.planned_for }}</p>
{% endif %}
{% if ev.props %}
<p class="muted">{{ ev.props|tojson }}</p>
{% endif %}
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/event/cancel/{{ ev.event_id }}"
hx-target="#drawer" hx-swap="innerHTML">
<button type="submit">Cancel</button>
</form>
</li>
{% endfor %}
</ul>
{% else %}
<p class="muted">No active events.</p>
{% endif %}
<details>
<summary>Plan event</summary>
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/event/plan"
hx-target="#drawer" hx-swap="innerHTML">
<label>
Kind:
<input type="text" name="kind" required
placeholder="e.g. dinner_reservation">
</label>
<label>
Planned for (ISO 8601):
<input type="text" name="planned_for" required
placeholder="2026-04-26T19:00:00+00:00">
</label>
<label>
Props (JSON):
<textarea name="props_json" rows="3"
placeholder='{"location": "Bistro X"}'>{}</textarea>
</label>
<button type="submit">Plan event</button>
</form>
</details>
</section>
<section class="drawer-section">
<h3>Threads</h3>
{% if open_threads %}
<ul class="thread-list">
{% for th in open_threads %}
<li class="thread-row">
<strong>{{ th.title }}</strong>
{% if th.summary %}
<p>{{ th.summary }}</p>
{% endif %}
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/thread/close/{{ th.thread_id }}"
hx-target="#drawer" hx-swap="innerHTML">
<button type="submit">Close</button>
</form>
</li>
{% endfor %}
</ul>
{% else %}
<p class="muted">No open threads.</p>
{% endif %}
</section>
{% if guest_bot %}
+236
View File
@@ -27,6 +27,8 @@ one so a later inverse edit can restore state (§6.4 final paragraph).
from __future__ import annotations
import json
import uuid
from pathlib import Path
from fastapi import APIRouter, Depends, Form, HTTPException, Request
@@ -38,11 +40,18 @@ from chat.services.relationship_seed import seed_inter_bot_edges
from chat.services.scene_summarize import apply_scene_close_summary
from chat.state.edges import get_edge
from chat.state.entities import get_bot, get_you, list_bots
from chat.state.events import list_active_events
from chat.state.group_node import get_group_node
from chat.state.memory import get_pinned
from chat.state.threads import list_open_threads
from chat.state.world import active_scene, get_activity, get_chat, get_container
from chat.web.bots import get_conn
from chat.web.kickoff import get_llm_client
from chat.web.skip import (
_now_iso,
process_elision_skip,
process_jump_skip,
)
TEMPLATES = Jinja2Templates(
directory=str(Path(__file__).resolve().parent.parent / "templates")
@@ -155,6 +164,10 @@ async def drawer(chat_id: str, request: Request, conn=Depends(get_conn)):
pinned = get_pinned(conn, chat["host_bot_id"])
# T59: active events + open threads for the new drawer sections.
active_events = list_active_events(conn, chat_id)
open_threads = list_open_threads(conn, chat_id)
return TEMPLATES.TemplateResponse(
request,
"_drawer.html",
@@ -180,6 +193,8 @@ async def drawer(chat_id: str, request: Request, conn=Depends(get_conn)):
"recent_memories": recent_memories,
"pinned": pinned,
"pin_cap": PIN_CAP,
"active_events": active_events,
"open_threads": open_threads,
},
)
@@ -839,3 +854,224 @@ async def remove_guest(
)
return await drawer(chat_id, request, conn)
# --- T59 events / threads / skip controls --------------------------------
#
# Five drawer-driven endpoints that emit Phase 3 event-log entries:
#
# * ``event_planned`` / ``event_cancelled`` for the events panel — props
# arrive as a JSON-encoded form field so the user can author arbitrary
# structured side-info without a custom HTMX widget per kind.
# * ``time_skip_elision`` / ``time_skip_jump`` for the skip panel —
# each emits the projector event AND an ``assistant_turn`` carrying the
# narration prose from :mod:`chat.services.skip_narration`. Jump skips
# ALSO write per-bot synthesized memories from any user-supplied
# ``notable_prose`` via :func:`synthesize_memories` +
# :func:`record_turn_memory_for_present`.
# * ``thread_closed`` for the threads panel.
#
# Skip narration is appended via plain ``append_event`` (assistant_turn
# has no projector handler — it's a transcript-only kind, see
# :func:`chat.web.turns._read_recent_dialogue`). The user will see the
# new turn on the next chat-detail page load; we do NOT broadcast via
# ``publish`` here because the SSE channel is scoped to the chat-detail
# page and the drawer partial is the response body — adding cross-cutting
# SSE here would require dragging the publish import + chat-channel state
# into the drawer module without a meaningful UX gain (the drawer only
# rerenders itself on these submissions).
@router.post(
"/chats/{chat_id}/drawer/event/plan",
response_class=HTMLResponse,
)
async def plan_event(
chat_id: str,
request: Request,
kind: str = Form(...),
planned_for: str = Form(...),
props_json: str = Form("{}"),
conn=Depends(get_conn),
):
"""Append an ``event_planned`` row from the drawer's "Plan event" form.
``props_json`` is parsed into a dict before being attached to the
payload so the projector can treat it as structured data. Bad JSON
yields ``400`` — the form template renders an inline error in that
case so the user can fix-and-resubmit without losing their input.
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
try:
props = json.loads(props_json) if props_json.strip() else {}
except json.JSONDecodeError as exc:
raise HTTPException(
status_code=400, detail=f"props_json must be valid JSON: {exc}"
)
if not isinstance(props, dict):
raise HTTPException(
status_code=400, detail="props_json must encode a JSON object"
)
event_id = f"evt_{uuid.uuid4().hex[:12]}"
append_and_apply(
conn,
kind="event_planned",
payload={
"event_id": event_id,
"chat_id": chat_id,
"kind": kind,
"props": props,
"planned_for": planned_for,
},
)
return await drawer(chat_id, request, conn)
@router.post(
"/chats/{chat_id}/drawer/event/cancel/{event_id}",
response_class=HTMLResponse,
)
async def cancel_event(
chat_id: str,
event_id: str,
request: Request,
conn=Depends(get_conn),
):
"""Append an ``event_cancelled`` row for ``event_id``.
``completed_at`` is sourced from the chat clock (so cancellations
timeline-align with the rest of the fiction) with a UTC-now fallback
when the clock isn't set. The projector is idempotent on terminal
statuses so a stale double-submit is harmless.
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
completed_at = chat.get("time") or _now_iso()
append_and_apply(
conn,
kind="event_cancelled",
payload={
"event_id": event_id,
"completed_at": completed_at,
},
)
return await drawer(chat_id, request, conn)
@router.post(
"/chats/{chat_id}/drawer/skip/elision",
response_class=HTMLResponse,
)
async def skip_elision(
chat_id: str,
request: Request,
landing_state_hint: str = Form(""),
new_time: str = Form(...),
conn=Depends(get_conn),
client=Depends(get_llm_client),
):
"""Elision skip: collapse in-progress activity into its end-state.
Thin HTTP wrapper around :func:`chat.web.skip.process_elision_skip`
(T62 extracted the controller). Validation failures surface as
``400`` and the route still returns the refreshed drawer partial on
success so HTMX swaps in the new chat clock.
"""
settings = request.app.state.settings
try:
await process_elision_skip(
conn,
client,
settings,
chat_id=chat_id,
new_time=new_time,
landing_state_hint=landing_state_hint,
)
except ValueError as exc:
# ``process_elision_skip`` raises on missing-chat or malformed /
# backwards new_time. The drawer used to 404 / 400 these
# separately — preserve the 404-vs-400 split by sniffing the
# error message so existing tests keep passing without changes.
if str(exc).startswith("chat not found"):
raise HTTPException(status_code=404, detail=str(exc))
raise HTTPException(status_code=400, detail=str(exc))
return await drawer(chat_id, request, conn)
@router.post(
"/chats/{chat_id}/drawer/skip/jump",
response_class=HTMLResponse,
)
async def skip_jump(
chat_id: str,
request: Request,
new_time: str = Form(...),
notable_prose: str = Form(""),
reset_activity: str = Form(""),
conn=Depends(get_conn),
client=Depends(get_llm_client),
):
"""Jump skip: bridge a longer fiction-time delta.
Thin HTTP wrapper around :func:`chat.web.skip.process_jump_skip`
(T62 extracted the controller). ``reset_activity`` is parsed
permissively here ("1" / "true" / "on" / "yes" — same shape as the
add-guest reseed flag) since HTML checkboxes typically post the
literal "1" or omit the field entirely.
"""
reset_flag = reset_activity.lower() in ("1", "true", "on", "yes")
settings = request.app.state.settings
try:
await process_jump_skip(
conn,
client,
settings,
chat_id=chat_id,
new_time=new_time,
notable_prose=notable_prose,
reset_activity=reset_flag,
)
except ValueError as exc:
if str(exc).startswith("chat not found"):
raise HTTPException(status_code=404, detail=str(exc))
raise HTTPException(status_code=400, detail=str(exc))
return await drawer(chat_id, request, conn)
@router.post(
"/chats/{chat_id}/drawer/thread/close/{thread_id}",
response_class=HTMLResponse,
)
async def close_thread(
chat_id: str,
thread_id: str,
request: Request,
conn=Depends(get_conn),
):
"""Append a ``thread_closed`` row for ``thread_id``.
Mirrors :func:`cancel_event` — chat-clock-or-now timestamp, projector
handles idempotency. The drawer's open-threads list is sourced from
``list_open_threads`` which filters by ``status='open'`` so a stale
double-submit is a no-op visually.
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
closed_at = chat.get("time") or _now_iso()
append_and_apply(
conn,
kind="thread_closed",
payload={
"thread_id": thread_id,
"closed_at": closed_at,
},
)
return await drawer(chat_id, request, conn)
+287
View File
@@ -0,0 +1,287 @@
"""Shared skip-flow controllers (T62).
Both the drawer skip routes (T59) and the natural-language skip parse
(T62) call into these controllers. Keep the controllers free of HTTP
concerns — they take ``conn`` + ``client`` + ``settings`` and structured
args, append events, and return a small result dict the caller can map
to whatever response shape it owes (drawer partial, 204, 422, etc.).
``ValueError`` is the controller-level signal for caller-mappable
validation failure (bad ISO timestamp, backwards skip). The drawer
routes translate it to ``HTTP 400``; the natural-language path either
swallows it (the parser handed us a degenerate hint) or surfaces it the
same way. Anything else (LLM failure, unexpected exception) propagates
uncaught — :func:`narrate_skip` already has its own deterministic
fallback for the routine LLM-down case, so a real exception here means
something we want to see.
The two controllers mirror the drawer T59 logic closely so the v1
guarantees (``time_skip_*`` lands first → memory writes ride the
post-skip clock → narration ``assistant_turn`` is appended last) hold
identically across the two entry points.
"""
from __future__ import annotations
from datetime import datetime, timezone
from sqlite3 import Connection
from chat.config import Settings
from chat.eventlog.log import append_and_apply, append_event
from chat.llm.client import LLMClient
from chat.services.memory_write import record_turn_memory_for_present
from chat.services.skip_narration import narrate_skip
from chat.services.synthesized_memories import synthesize_memories
from chat.state.entities import get_bot, get_you
from chat.state.world import get_activity, get_chat
def _parse_iso_time(value: str) -> datetime | None:
"""Permissive ISO 8601 parser shared with the drawer routes (T59).
``datetime.fromisoformat`` doesn't accept a trailing ``Z`` until
Python 3.11; we normalize it to ``+00:00`` so older interpreters
parse the same set of strings the drawer accepts.
"""
if not value:
return None
try:
v = value.strip()
if v.endswith("Z"):
v = v[:-1] + "+00:00"
return datetime.fromisoformat(v)
except (TypeError, ValueError):
return None
def _validate_new_time(chat: dict, new_time: str) -> None:
"""Raise ``ValueError`` if ``new_time`` is unparseable or backwards.
The drawer route maps the raised error to ``HTTP 400``; the
natural-language path may also surface it as a ``400``. Centralizing
the rule here means both entry points enforce the same invariant
(no causality-corrupting backwards jumps).
"""
new_dt = _parse_iso_time(new_time)
if new_dt is None:
raise ValueError(f"new_time must be ISO 8601, got {new_time!r}")
cur_dt = _parse_iso_time(chat.get("time") or "")
if cur_dt is not None and new_dt < cur_dt:
raise ValueError(
"new_time must not be earlier than the current chat clock"
)
async def process_elision_skip(
conn: Connection,
client: LLMClient,
settings: Settings,
*,
chat_id: str,
new_time: str,
landing_state_hint: str = "",
) -> dict:
"""Run an elision skip end-to-end.
Validates ``new_time`` against the current chat clock, appends a
``time_skip_elision`` event (chat clock advances), generates a
transition narration via :func:`narrate_skip`, and appends an
``assistant_turn`` carrying the narration. ``narrate_skip`` has its
own deterministic fallback so this never blocks on the model.
Returns ``{"assistant_text": ..., "speaker_id": ..., "skip_event_id":
..., "assistant_event_id": ...}`` so callers can introspect the
generated turn (e.g. for SSE rebroadcast or test assertions).
Raises ``ValueError`` on validation failure or when the chat row
can't be located (the drawer maps it to ``HTTP 400`` / ``404``
respectively; the natural-language path follows the same shape).
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise ValueError(f"chat not found: {chat_id}")
_validate_new_time(chat, new_time)
host_bot = get_bot(conn, chat["host_bot_id"]) or {
"id": chat["host_bot_id"],
"name": "host",
"persona": "",
}
you_entity = get_you(conn) or {"name": "you"}
# The drawer route reaches into the host bot's current activity to
# surface the verb to the narration helper — we do the same so both
# entry points produce the same prose for the same chat state.
bot_activity = get_activity(conn, chat["host_bot_id"]) or {}
current_activity = (bot_activity.get("action") or {}).get("verb") or ""
narration = await narrate_skip(
client,
narrative_model=settings.narrative_model,
skip_kind="elision",
speaker_bot=host_bot,
you_name=you_entity.get("name") or "you",
current_time=chat.get("time") or "",
new_time=new_time,
current_activity=current_activity,
landing_state_hint=landing_state_hint,
timeout_s=settings.classifier_timeout_s,
)
skip_event_id = append_and_apply(
conn,
kind="time_skip_elision",
payload={"chat_id": chat_id, "new_time": new_time},
)
speaker_id = host_bot.get("id") or chat["host_bot_id"]
assistant_event_id = append_event(
conn,
kind="assistant_turn",
payload={
"chat_id": chat_id,
"speaker_id": speaker_id,
"text": narration,
"truncated": False,
},
)
return {
"assistant_text": narration,
"speaker_id": speaker_id,
"skip_event_id": skip_event_id,
"assistant_event_id": assistant_event_id,
}
async def process_jump_skip(
conn: Connection,
client: LLMClient,
settings: Settings,
*,
chat_id: str,
new_time: str,
notable_prose: str = "",
reset_activity: bool = False,
) -> dict:
"""Run a jump skip end-to-end.
Same validations as :func:`process_elision_skip`. Emits
``time_skip_jump`` *before* synthesizing memories so per-bot writes
record the post-jump chat clock (mirroring how a regular turn's
memory carries the chat clock). When ``notable_prose`` is non-empty,
runs :func:`synthesize_memories` once per present bot witness, then
fans the resulting memories out via
:func:`record_turn_memory_for_present` with ``source="synthesized"``.
Finally appends the narration ``assistant_turn``.
Returns ``{"assistant_text": ..., "speaker_id": ..., "skip_event_id":
..., "assistant_event_id": ...}``.
Raises ``ValueError`` on validation failure (caller maps to ``400``).
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise ValueError(f"chat not found: {chat_id}")
_validate_new_time(chat, new_time)
host_bot = get_bot(conn, chat["host_bot_id"]) or {
"id": chat["host_bot_id"],
"name": "host",
"persona": "",
}
you_entity = get_you(conn) or {"name": "you"}
you_name = you_entity.get("name") or "you"
guest_bot_id = chat.get("guest_bot_id")
guest_bot = get_bot(conn, guest_bot_id) if guest_bot_id else None
# Emit time_skip_jump up front so subsequent memory writes ride the
# post-jump chat clock (matches the drawer T59 behavior pinned by
# test_post_skip_jump_with_notable_prose_writes_synthesized_memories).
skip_event_id = append_and_apply(
conn,
kind="time_skip_jump",
payload={
"chat_id": chat_id,
"new_time": new_time,
"reset_activity": reset_activity,
},
)
# Synthesize per-bot memories when prose is non-empty. The helper
# short-circuits on whitespace prose, but gating the loop here keeps
# the canned-LLM-queue accounting predictable for tests.
if notable_prose.strip():
present_bots: list[dict] = [host_bot]
if guest_bot is not None:
present_bots.append(guest_bot)
for bot in present_bots:
digest = await synthesize_memories(
client,
classifier_model=settings.classifier_model,
prose=notable_prose,
bot_name=bot.get("name") or "",
bot_persona=bot.get("persona") or "",
you_name=you_name,
timeout_s=settings.classifier_timeout_s,
)
for mem in digest.memories:
# ``record_turn_memory_for_present`` writes one row per
# present bot per call — we already iterate by bot here,
# so guest_bot_id=None avoids double-writing the guest's
# row when bot==guest.
record_turn_memory_for_present(
conn,
chat_id=chat_id,
host_bot_id=bot["id"],
guest_bot_id=None,
narrative_text=mem.text,
chat_clock_at=new_time,
source="synthesized",
significance=mem.significance,
)
narration = await narrate_skip(
client,
narrative_model=settings.narrative_model,
skip_kind="jump",
speaker_bot=host_bot,
you_name=you_name,
current_time=chat.get("time") or "",
new_time=new_time,
current_activity="",
landing_state_hint=notable_prose,
timeout_s=settings.classifier_timeout_s,
)
speaker_id = host_bot.get("id") or chat["host_bot_id"]
assistant_event_id = append_event(
conn,
kind="assistant_turn",
payload={
"chat_id": chat_id,
"speaker_id": speaker_id,
"text": narration,
"truncated": False,
},
)
return {
"assistant_text": narration,
"speaker_id": speaker_id,
"skip_event_id": skip_event_id,
"assistant_event_id": assistant_event_id,
}
def _now_iso() -> str:
"""UTC ISO timestamp used by callers as a chat-clock fallback."""
return datetime.now(timezone.utc).isoformat()
__all__ = [
"process_elision_skip",
"process_jump_skip",
"_now_iso",
"_parse_iso_time",
]
+136 -1
View File
@@ -51,12 +51,16 @@ import html
import json
import re
from datetime import timedelta
from fastapi import APIRouter, Depends, Form, HTTPException, Request
from fastapi.responses import HTMLResponse, RedirectResponse, Response
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response
from chat.eventlog.log import append_and_apply, append_event
from chat.services.addressee import detect_addressee
from chat.services.background import SignificanceJob
from chat.services.event_lifecycle import detect_event_transitions
from chat.services.event_promotion import promote_completed_event
from chat.services.interjection import detect_interjection
from chat.services.memory_write import record_turn_memory_for_present
from chat.services.multi_state_update import compute_state_updates_for_present
@@ -67,11 +71,13 @@ from chat.services.scene_summarize import apply_scene_close_summary
from chat.services.turn_parse import ParsedTurn, parse_turn
from chat.state.edges import get_edge
from chat.state.entities import get_bot, get_you
from chat.state.events import list_active_events
from chat.state.world import active_scene, get_chat, get_container
from chat.web.bots import get_conn
from chat.web.kickoff import get_llm_client
from chat.web.pubsub import publish
from chat.web.render import render_turn_html as _render_turn_html
from chat.web.skip import _parse_iso_time, process_elision_skip
router = APIRouter()
@@ -251,6 +257,65 @@ async def post_turn(
)
prompt_prose = _strip_ooc_for_prompt(parsed)
# 1a. Skip-command short-circuit (T62). The parser may classify the
# prose as a time-skip directive — in which case the regular
# narrative path (addressee detection, narrative stream, post-turn
# state-update + scene-close passes) is skipped entirely. Elision
# runs through the shared controller in :mod:`chat.web.skip`; jump
# is drawer-only for Phase 3 (the natural-language path returns
# 422 directing the user to the drawer's jump form, where they can
# supply structured ``notable_prose`` and a target time). Anything
# not matching these intents falls through to the narrative branch.
intent = getattr(parsed, "intent", "narrative") or "narrative"
if intent == "skip_jump":
# Drawer-only jump for Phase 3: parsing a free-form fiction-time
# delta out of natural language ("next morning" -> ?) is fragile
# enough that we'd rather route the user to the drawer form,
# where they pick a concrete ISO time and an optional notable-
# prose field. 422 = "request shape is understood, but the
# required structured input lives on a different surface".
return JSONResponse(
{
"error": (
"Jump skip requires the drawer's jump form for "
"notable_prose."
)
},
status_code=422,
)
if intent == "skip_elision":
# Derive ``new_time`` from the chat clock. Phase 3 stub: bump by
# 1 hour. The drawer's elision form is the structured path when
# the author wants a specific landing time; here the goal is
# "elide the dull bit" and any sensible forward step is fine —
# ``narrate_skip`` weaves the landing-state hint into the
# transition prose so the prose carries the semantic time, not
# the timestamp itself.
cur_dt = _parse_iso_time(chat.get("time") or "")
new_time = (
(cur_dt + timedelta(hours=1)).isoformat()
if cur_dt is not None
else (chat.get("time") or "")
)
try:
await process_elision_skip(
conn,
client,
settings,
chat_id=chat_id,
new_time=new_time,
landing_state_hint=getattr(parsed, "landing_state_hint", "")
or "",
)
except ValueError as exc:
# The controller raises on missing chat / bad new_time.
# Missing chat is already handled above (we'd have 404'd);
# a bad new_time here is a stub-derivation bug rather than
# user input — surface as 400 with the controller message.
raise HTTPException(status_code=400, detail=str(exc))
return Response(status_code=204)
# 2. Append user_turn event.
user_turn_event_id = append_event(
conn,
@@ -654,6 +719,76 @@ async def post_turn(
)
)
# 8a. Event-lifecycle detection (Phase 3, T61). Runs after the post-turn
# classifier passes (memory write + state update + optional
# interjection) and BEFORE scene-close detection. The classifier reads
# ``primary_text`` against the chat's currently-active events and
# returns a (usually empty) list of transitions. Each transition lands
# an ``event_started`` / ``event_completed`` / ``event_cancelled``
# event via ``append_and_apply`` so the events projection updates
# synchronously. A completion is followed inline by
# ``promote_completed_event`` so any structured artifacts the event
# carries (knowledge_facts, relationship_change, acquired_objects)
# land in state in the same turn — see chat/services/event_promotion.
#
# ``detect_event_transitions`` short-circuits when ``active_events``
# is empty (per T52), so chats without active events don't pay a
# classifier round-trip and existing fixtures need no extra canned
# slots.
active_events = list_active_events(conn, chat_id)
if active_events:
lifecycle_decision = await detect_event_transitions(
client,
classifier_model=settings.classifier_model,
narrative_text=primary_text,
active_events=active_events,
timeout_s=settings.classifier_timeout_s,
)
for transition in lifecycle_decision.transitions:
if transition.new_status == "active":
append_and_apply(
conn,
kind="event_started",
payload={
"event_id": transition.event_id,
"started_at": chat.get("time"),
},
)
elif transition.new_status == "completed":
append_and_apply(
conn,
kind="event_completed",
payload={
"event_id": transition.event_id,
"completed_at": chat.get("time"),
},
)
# Run promotion inline so the artifact-emitting events
# (edge_update / manual_edit) land synchronously after
# the completion. ``promote_completed_event`` is
# synchronous (no await) and skips silently when the
# event row's status isn't 'completed' — a safety net
# for races, not expected to trigger in practice.
promote_completed_event(
conn,
event_id=transition.event_id,
chat_id=chat_id,
chat_clock_at=chat.get("time"),
)
elif transition.new_status == "cancelled":
append_and_apply(
conn,
kind="event_cancelled",
payload={
"event_id": transition.event_id,
"completed_at": chat.get("time"),
},
)
# Any other ``new_status`` value falls through silently —
# the lifecycle service constrains the schema to the three
# valid transitions, and a defensive no-op here keeps the
# turn flow tolerant of unexpected outputs.
# 9. Scene-close detection (Plan §7.2, T26). Runs AFTER assistant_turn
# and the optional interjection so the bots' responses are part of
# the closing scene's final beat — closing before narrative would
+452
View File
@@ -0,0 +1,452 @@
"""T59: drawer events / threads / skip controls.
Extends the chat drawer with three new sections (Events, Threads, Skip)
and five new POST endpoints:
* ``POST /chats/{chat_id}/drawer/event/plan`` — emits ``event_planned``.
* ``POST /chats/{chat_id}/drawer/event/cancel/{event_id}`` — emits
``event_cancelled``.
* ``POST /chats/{chat_id}/drawer/skip/elision`` — validates new_time,
emits ``time_skip_elision`` plus an ``assistant_turn`` carrying the
narrated transition prose from :mod:`chat.services.skip_narration`.
* ``POST /chats/{chat_id}/drawer/skip/jump`` — validates new_time, emits
``time_skip_jump`` plus per-bot synthesized ``memory_written`` events
derived from the user-supplied "anything notable" prose, and an
``assistant_turn`` carrying the narration.
* ``POST /chats/{chat_id}/drawer/thread/close/{thread_id}`` — emits
``thread_closed``.
Each route returns the refreshed drawer partial (HTMX swap target) so
the tests assert both the persisted event_log effect AND the rendered
section content. Wire-up follows the T42 ``MockLLMClient`` pattern.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from chat.app import app
from chat.db.connection import open_db
from chat.eventlog.log import append_and_apply, append_event
from chat.eventlog.projector import project
from chat.llm.mock import MockLLMClient
@pytest.fixture
def client(tmp_path, monkeypatch):
cfg = tmp_path / "config.toml"
cfg.write_text('featherless_api_key = "test"\n')
monkeypatch.setenv("CHAT_CONFIG_PATH", str(cfg))
db = tmp_path / "test.db"
monkeypatch.setenv("CHAT_DB_PATH", str(db))
with TestClient(app) as c:
if hasattr(app.state, "background_worker"):
app.state.background_worker.enabled = False
yield c
def _bot_payload(bot_id: str, name: str) -> dict:
return {
"id": bot_id,
"name": name,
"persona": "...",
"voice_samples": [],
"traits": [],
"backstory": "",
"initial_relationship_to_you": "",
"kickoff_prose": "",
}
def _seed_chat(db: Path, *, with_scene: bool = True) -> None:
"""Seed a chat hosted by ``bot_a`` (with ``bot_b`` authored as a
candidate guest) so the skip-jump path can write per-bot synthesized
memories when a guest is present.
"""
with open_db(db) as conn:
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="you_authored",
payload={"name": "Me", "pronouns": "they/them", "persona": ""},
)
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": "",
},
)
if with_scene:
append_event(
conn,
kind="scene_opened",
payload={
"chat_id": "chat_bot_a",
"container_id": None,
"started_at": "2026-04-26T20:00:00+00:00",
"participants": ["you", "bot_a"],
},
)
project(conn)
def _override_llm(canned: list[str]):
"""Wire a ``MockLLMClient`` into the drawer's LLM dependency."""
from chat.web.kickoff import get_llm_client
app.dependency_overrides[get_llm_client] = lambda: MockLLMClient(
canned=list(canned)
)
# ---------------------------------------------------------------------------
# 1. Empty drawer state — Events + Threads sections render but show
# empty-state copy (no row markup) when no events / threads exist.
# ---------------------------------------------------------------------------
def test_get_drawer_with_no_events_or_threads_omits_sections(client, tmp_path):
_seed_chat(tmp_path / "test.db")
response = client.get("/chats/chat_bot_a/drawer")
assert response.status_code == 200
body = response.text
# Sections render with empty-state copy — deterministic markers.
assert "<h3>Events</h3>" in body
assert "<h3>Threads</h3>" in body
assert "No active events" in body
assert "No open threads" in body
# Skip controls always render under Activity (gated by chat clock).
assert "Elision skip" in body
assert "Jump skip" in body
# ---------------------------------------------------------------------------
# 2. POST event/plan — event_planned lands and the drawer lists it.
# ---------------------------------------------------------------------------
def test_post_event_plan_appends_event_planned_and_renders(client, tmp_path):
_seed_chat(tmp_path / "test.db")
response = client.post(
"/chats/chat_bot_a/drawer/event/plan",
data={
"kind": "dinner_reservation",
"planned_for": "2026-04-26T19:00:00+00:00",
"props_json": json.dumps({"restaurant": "Bistro X"}),
},
)
assert response.status_code == 200
with open_db(tmp_path / "test.db") as conn:
rows = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'event_planned'"
).fetchall()
assert len(rows) == 1
payload = json.loads(rows[0][0])
assert payload["kind"] == "dinner_reservation"
assert payload["chat_id"] == "chat_bot_a"
assert payload["planned_for"] == "2026-04-26T19:00:00+00:00"
assert payload["props"] == {"restaurant": "Bistro X"}
assert payload["event_id"].startswith("evt_")
# Refreshed partial lists the new event by kind.
body = response.text
assert "dinner_reservation" in body
def test_post_event_plan_invalid_props_json_returns_400(client, tmp_path):
_seed_chat(tmp_path / "test.db")
response = client.post(
"/chats/chat_bot_a/drawer/event/plan",
data={
"kind": "dinner_reservation",
"planned_for": "2026-04-26T19:00:00+00:00",
"props_json": "not valid json {",
},
)
assert response.status_code == 400
# ---------------------------------------------------------------------------
# 3. POST event/cancel — event_cancelled lands and the active list drops it.
# ---------------------------------------------------------------------------
def test_post_event_cancel_appends_event_cancelled(client, tmp_path):
_seed_chat(tmp_path / "test.db")
# Plan first via the route so the test exercises both sides.
plan_resp = client.post(
"/chats/chat_bot_a/drawer/event/plan",
data={
"kind": "doctor_visit",
"planned_for": "2026-04-27T09:00:00+00:00",
"props_json": "{}",
},
)
assert plan_resp.status_code == 200
with open_db(tmp_path / "test.db") as conn:
row = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'event_planned' "
"ORDER BY id DESC LIMIT 1"
).fetchone()
event_id = json.loads(row[0])["event_id"]
cancel_resp = client.post(
f"/chats/chat_bot_a/drawer/event/cancel/{event_id}"
)
assert cancel_resp.status_code == 200
with open_db(tmp_path / "test.db") as conn:
cancelled = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'event_cancelled'"
).fetchall()
assert len(cancelled) == 1
cp = json.loads(cancelled[0][0])
assert cp["event_id"] == event_id
# Active-events query should no longer surface this event.
from chat.state.events import list_active_events
assert list_active_events(conn, "chat_bot_a") == []
# ---------------------------------------------------------------------------
# 4. POST skip/elision — emits time_skip_elision + assistant_turn narration.
# ---------------------------------------------------------------------------
def test_post_skip_elision_advances_clock_and_emits_narration(client, tmp_path):
_seed_chat(tmp_path / "test.db")
canned_narration = "We pull up to the curb just before sunset."
_override_llm([canned_narration])
try:
response = client.post(
"/chats/chat_bot_a/drawer/skip/elision",
data={
"landing_state_hint": "arriving at the venue",
"new_time": "2026-04-26T20:30:00+00:00",
},
)
assert response.status_code == 200
finally:
app.dependency_overrides.clear()
with open_db(tmp_path / "test.db") as conn:
from chat.state.world import get_chat
chat = get_chat(conn, "chat_bot_a")
assert chat["time"] == "2026-04-26T20:30:00+00:00"
skip_rows = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'time_skip_elision'"
).fetchall()
assert len(skip_rows) == 1
sp = json.loads(skip_rows[0][0])
assert sp["chat_id"] == "chat_bot_a"
assert sp["new_time"] == "2026-04-26T20:30:00+00:00"
# An assistant_turn event landed with the narration text.
turn_rows = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'assistant_turn'"
).fetchall()
assert len(turn_rows) == 1
tp = json.loads(turn_rows[0][0])
assert tp["chat_id"] == "chat_bot_a"
assert tp["text"].strip() # non-empty narration
assert tp["speaker_id"] == "bot_a"
def test_post_skip_elision_invalid_time_returns_400(client, tmp_path):
_seed_chat(tmp_path / "test.db")
_override_llm([])
try:
# Garbled ISO timestamp.
bad_resp = client.post(
"/chats/chat_bot_a/drawer/skip/elision",
data={"landing_state_hint": "x", "new_time": "not-a-time"},
)
assert bad_resp.status_code == 400
# Backwards-in-time skip: chat seeded at 20:00, asking 19:00.
backwards_resp = client.post(
"/chats/chat_bot_a/drawer/skip/elision",
data={
"landing_state_hint": "x",
"new_time": "2026-04-26T19:00:00+00:00",
},
)
assert backwards_resp.status_code == 400
finally:
app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# 5. POST skip/jump — synthesized memories per present bot + narration.
# ---------------------------------------------------------------------------
def test_post_skip_jump_with_notable_prose_writes_synthesized_memories(
client, tmp_path
):
_seed_chat(tmp_path / "test.db")
# Single host present (no guest) — exactly one synthesize call,
# one narration call. The synthesize digest carries two memories so
# we can assert N writes lands the right shape.
digest_json = json.dumps(
{
"memories": [
{
"text": "We bumped into an old friend at the cafe.",
"significance": 1,
"affinity_delta": 0,
"trust_delta": 0,
},
{
"text": "It started raining on the walk home.",
"significance": 1,
"affinity_delta": 0,
"trust_delta": 0,
},
]
}
)
narration = "The afternoon slipped by quickly."
_override_llm([digest_json, narration])
try:
response = client.post(
"/chats/chat_bot_a/drawer/skip/jump",
data={
"new_time": "2026-04-27T08:00:00+00:00",
"notable_prose": (
"We ran into an old friend, and it rained on the way back."
),
"reset_activity": "1",
},
)
assert response.status_code == 200
finally:
app.dependency_overrides.clear()
with open_db(tmp_path / "test.db") as conn:
from chat.state.world import get_chat
chat = get_chat(conn, "chat_bot_a")
assert chat["time"] == "2026-04-27T08:00:00+00:00"
jump_rows = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'time_skip_jump'"
).fetchall()
assert len(jump_rows) == 1
jp = json.loads(jump_rows[0][0])
assert jp["chat_id"] == "chat_bot_a"
assert jp["new_time"] == "2026-04-27T08:00:00+00:00"
assert jp["reset_activity"] is True
# Two synthesized memories land for the lone host bot
# (record_turn_memory_for_present writes one row per present bot
# per call — host only here, so 2 memories x 1 bot = 2 events).
mem_rows = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'memory_written'"
).fetchall()
synth_payloads = [
json.loads(r[0])
for r in mem_rows
if json.loads(r[0]).get("source") == "synthesized"
]
assert len(synth_payloads) == 2
for p in synth_payloads:
assert p["owner_id"] == "bot_a"
assert p["chat_id"] == "chat_bot_a"
# And the assistant_turn narration landed.
turn_rows = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'assistant_turn'"
).fetchall()
assert len(turn_rows) == 1
tp = json.loads(turn_rows[0][0])
assert tp["text"].strip()
assert tp["speaker_id"] == "bot_a"
def test_post_skip_jump_with_empty_prose_skips_memory_writes(client, tmp_path):
_seed_chat(tmp_path / "test.db")
# Empty prose short-circuits in synthesize_memories before any LLM call,
# so the canned queue only needs the narration.
narration = "(next morning: still in the kitchen.)"
_override_llm([narration])
try:
response = client.post(
"/chats/chat_bot_a/drawer/skip/jump",
data={
"new_time": "2026-04-27T08:00:00+00:00",
"notable_prose": " ",
"reset_activity": "",
},
)
assert response.status_code == 200
finally:
app.dependency_overrides.clear()
with open_db(tmp_path / "test.db") as conn:
synth = conn.execute(
"SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written' "
"AND payload_json LIKE '%synthesized%'"
).fetchone()[0]
assert synth == 0
# ---------------------------------------------------------------------------
# 6. POST thread/close — thread_closed lands and the open list drops it.
# ---------------------------------------------------------------------------
def test_post_thread_close_appends_thread_closed(client, tmp_path):
_seed_chat(tmp_path / "test.db")
# Open a thread directly via append_and_apply so the test focuses on
# the close route's effect.
with open_db(tmp_path / "test.db") as conn:
append_and_apply(
conn,
kind="thread_opened",
payload={
"thread_id": "thr_alpha",
"chat_id": "chat_bot_a",
"title": "the missing key",
"summary": "Couldn't find the key.",
},
)
response = client.post("/chats/chat_bot_a/drawer/thread/close/thr_alpha")
assert response.status_code == 200
with open_db(tmp_path / "test.db") as conn:
closed = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'thread_closed'"
).fetchall()
assert len(closed) == 1
cp = json.loads(closed[0][0])
assert cp["thread_id"] == "thr_alpha"
from chat.state.threads import list_open_threads
assert list_open_threads(conn, "chat_bot_a") == []
+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"
+92 -1
View File
@@ -12,12 +12,14 @@ import pytest
from chat.db.connection import open_db
from chat.db.migrate import apply_migrations
from chat.eventlog.log import append_event
from chat.eventlog.log import append_and_apply, append_event
from chat.eventlog.projector import project
import chat.state.entities # noqa: F401 (registers handlers)
import chat.state.edges # noqa: F401
import chat.state.memory # noqa: F401
import chat.state.world # noqa: F401
import chat.state.events # noqa: F401
import chat.state.threads # noqa: F401
from chat.llm.client import Message
from chat.services.prompt import assemble_narrative_prompt
@@ -761,3 +763,92 @@ def test_assemble_with_tight_budget_drops_guest_activity_first(tmp_path):
import tiktoken
enc = tiktoken.get_encoding("cl100k_base")
assert len(enc.encode(body)) <= 340
# ---------------------------------------------------------------------------
# Task 60: Active events + open threads in prompt assembly
# ---------------------------------------------------------------------------
def test_assemble_with_no_events_or_threads_omits_blocks(tmp_path):
"""Regression: with the basic 2-entity scenario (no events seeded, no
threads seeded), the assembled prompt must NOT contain the
``Active events:`` or ``Open threads:`` headers both blocks are
omit-when-empty."""
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_basic(conn)
msgs = assemble_narrative_prompt(
conn,
chat_id="chat_bot_a",
speaker_bot_id="bot_a",
recent_dialogue=[],
retrieved_memory_summaries=[],
)
body = msgs[0].content
assert "Active events:" not in body
assert "Open threads:" not in body
def test_assemble_with_active_events_renders_block(tmp_path):
"""Seed a planned event then transition it to active; the assembled
prompt should render the ``Active events:`` block listing the event
by kind."""
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_basic(conn)
# event_planned then event_started → status="active". Use
# append_and_apply because _seed_basic already projected; calling
# project() again would replay every prior event (and trip
# UNIQUE constraints on chat_created etc.).
append_and_apply(conn, kind="event_planned", payload={
"event_id": "evt_park",
"chat_id": "chat_bot_a",
"kind": "date_at_park",
"props": {"location": "Riverside Park", "vibe": "casual"},
"planned_for": "2026-04-30T18:00:00+00:00",
})
append_and_apply(conn, kind="event_started", payload={
"event_id": "evt_park",
"started_at": "2026-04-30T18:05:00+00:00",
})
msgs = assemble_narrative_prompt(
conn,
chat_id="chat_bot_a",
speaker_bot_id="bot_a",
recent_dialogue=[],
retrieved_memory_summaries=[],
)
body = msgs[0].content
assert "Active events:" in body
assert "date_at_park" in body
def test_assemble_with_open_thread_renders_block(tmp_path):
"""Seed a single open thread; the assembled prompt should render the
``Open threads:`` block listing the thread by title."""
db = tmp_path / "t.db"
apply_migrations(db)
with open_db(db) as conn:
_seed_basic(conn)
# _seed_basic already projected; use append_and_apply for the
# post-seed event so we don't re-trigger UNIQUE constraint
# collisions on the prior chat_created/etc. events.
append_and_apply(conn, kind="thread_opened", payload={
"thread_id": "thr_job",
"chat_id": "chat_bot_a",
"title": "Maya's job hunt",
"summary": "Maya is looking for a new job",
})
msgs = assemble_narrative_prompt(
conn,
chat_id="chat_bot_a",
speaker_bot_id="bot_a",
recent_dialogue=[],
retrieved_memory_summaries=[],
)
body = msgs[0].content
assert "Open threads:" in body
assert "Maya's job hunt" in body
+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"
+422 -1
View File
@@ -19,7 +19,7 @@ from fastapi.testclient import TestClient
from chat.app import app
from chat.db.connection import open_db
from chat.eventlog.log import append_event
from chat.eventlog.log import append_and_apply, append_event
from chat.eventlog.projector import project
from chat.llm.mock import MockLLMClient
@@ -896,3 +896,424 @@ def test_interjection_enqueues_significance_job(app_state_setup, tmp_path):
# The two narrative texts should be the two streamed beats.
narrative_texts = sorted(job.narrative_text for job in captured_jobs)
assert narrative_texts == ["Interjection beat!", "Primary beat."]
# ---------------------------------------------------------------------------
# Phase 3 (T61) — per-turn event-lifecycle detection + completion promotion.
#
# After the post-turn classifier passes (memory write, state update,
# interjection check) and BEFORE scene-close detection, ``post_turn``
# calls :func:`detect_event_transitions`. Each transition becomes one
# of ``event_started`` / ``event_completed`` / ``event_cancelled``. A
# completed event is followed inline by ``promote_completed_event`` so
# the props it carries (knowledge_facts, etc.) land in state
# synchronously.
#
# When no active events are seeded the classifier short-circuits without
# an LLM call (per T52) — the canned queue therefore needs ZERO extra
# slots in that case.
# ---------------------------------------------------------------------------
def test_turn_with_event_transition_appends_started_event(
app_state_setup, tmp_path
):
"""A planned event becomes active when the classifier reports a
``new_status='active'`` transition for that event_id.
Canned queue (5 calls single-bot, no scene seeded):
1. parse_turn
2. narrative stream
3. state-update bot_a -> you
4. state-update you -> bot_a
5. detect_event_transitions -> 1 transition (active)
"""
_seed(tmp_path / "test.db")
# Seed a planned event so list_active_events returns 1 row. Use
# append_and_apply so we don't re-replay the prior chat_created event
# (whose handler is INSERT-not-IGNORE and would 409 on replay).
with open_db(tmp_path / "test.db") as conn:
append_and_apply(
conn,
kind="event_planned",
payload={
"event_id": "evt_1",
"chat_id": "chat_bot_a",
"kind": "story_event",
"props": {},
"planned_for": "2026-04-30T18:00:00+00:00",
},
)
canned_parse = json.dumps(
{"segments": [{"kind": "dialogue", "text": "they arrived"}]}
)
canned_event_decision = json.dumps(
{
"transitions": [
{
"event_id": "evt_1",
"new_status": "active",
"reason": "they arrived",
}
]
}
)
mock = _override_llm(
[
canned_parse,
"They walk in.",
_zero_state(),
_zero_state(),
canned_event_decision,
]
)
try:
response = app_state_setup.post(
"/chats/chat_bot_a/turns", data={"prose": "they arrived"}
)
assert response.status_code == 204
finally:
app.dependency_overrides.clear()
# All 5 canned slots consumed.
assert mock._canned == []
with open_db(tmp_path / "test.db") as conn:
# event_started landed in event_log.
rows = conn.execute(
"SELECT payload_json FROM event_log "
"WHERE kind = 'event_started' ORDER BY id"
).fetchall()
assert len(rows) == 1
started_payload = json.loads(rows[0][0])
assert started_payload["event_id"] == "evt_1"
assert started_payload["started_at"] == "2026-04-26T20:00:00+00:00"
# The events projection row reflects the active status.
ev_row = conn.execute(
"SELECT status, started_at FROM events WHERE event_id = ?",
("evt_1",),
).fetchone()
assert ev_row is not None
assert ev_row[0] == "active"
assert ev_row[1] == "2026-04-26T20:00:00+00:00"
def test_turn_with_event_completion_runs_promotion(app_state_setup, tmp_path):
"""An active event with knowledge_facts in props completes; the
inline call to ``promote_completed_event`` emits the corresponding
``edge_update``.
"""
_seed(tmp_path / "test.db")
# Seed: planned -> started so the event is currently active. Props
# carry a knowledge_fact that promotion will turn into an edge_update.
# Use append_and_apply (not project) to avoid re-replaying chat_created.
with open_db(tmp_path / "test.db") as conn:
append_and_apply(
conn,
kind="event_planned",
payload={
"event_id": "evt_2",
"chat_id": "chat_bot_a",
"kind": "story_event",
"props": {
"knowledge_facts": [
{
"owner_id": "bot_a",
"target_id": "you",
"fact": "Maya likes pottery",
}
]
},
"planned_for": "2026-04-30T18:00:00+00:00",
},
)
append_and_apply(
conn,
kind="event_started",
payload={
"event_id": "evt_2",
"started_at": "2026-04-30T19:00:00+00:00",
},
)
# Snapshot the max event_log id so we can assert on rows AFTER the turn.
with open_db(tmp_path / "test.db") as conn:
before_id = conn.execute(
"SELECT COALESCE(MAX(id), 0) FROM event_log"
).fetchone()[0]
canned_parse = json.dumps(
{"segments": [{"kind": "dialogue", "text": "we wrap it up"}]}
)
canned_event_decision = json.dumps(
{
"transitions": [
{
"event_id": "evt_2",
"new_status": "completed",
"reason": "wrapped",
}
]
}
)
mock = _override_llm(
[
canned_parse,
"They wrap it up.",
_zero_state(),
_zero_state(),
canned_event_decision,
]
)
try:
response = app_state_setup.post(
"/chats/chat_bot_a/turns", data={"prose": "we wrap it up"}
)
assert response.status_code == 204
finally:
app.dependency_overrides.clear()
assert mock._canned == []
with open_db(tmp_path / "test.db") as conn:
# event_completed landed.
completed_rows = conn.execute(
"SELECT id, payload_json FROM event_log "
"WHERE kind = 'event_completed' AND id > ? ORDER BY id",
(before_id,),
).fetchall()
assert len(completed_rows) == 1
completed_payload = json.loads(completed_rows[0][1])
assert completed_payload["event_id"] == "evt_2"
completed_id = completed_rows[0][0]
# promote_completed_event ran inline AFTER event_completed: the
# follow-on edge_update carries the knowledge fact and is tagged
# with source=event_promotion.
promo_rows = conn.execute(
"SELECT payload_json FROM event_log "
"WHERE kind = 'edge_update' AND id > ? ORDER BY id",
(completed_id,),
).fetchall()
promo_facts: list[str] = []
for (payload_json,) in promo_rows:
p = json.loads(payload_json)
if p.get("source") == "event_promotion":
promo_facts.extend(p.get("knowledge_facts") or [])
assert "Maya likes pottery" in promo_facts
def test_turn_with_no_active_events_skips_classifier(app_state_setup, tmp_path):
"""When no active events are seeded, ``detect_event_transitions``
short-circuits without an LLM call (per T52). The canned queue must
therefore have ZERO event-detection slots same shape as the
Phase 2 no-guest baseline.
"""
_seed(tmp_path / "test.db")
canned_parse = json.dumps(
{"segments": [{"kind": "dialogue", "text": "hello"}]}
)
# Only 4 slots: parse + narrative + 2 state-updates. NO extra slot for
# event-detection — non-existent active_events causes the helper to
# short-circuit before pulling from the queue.
mock = _override_llm(
[canned_parse, "Hi there.", _zero_state(), _zero_state()]
)
try:
response = app_state_setup.post(
"/chats/chat_bot_a/turns", data={"prose": "hello"}
)
assert response.status_code == 204
finally:
app.dependency_overrides.clear()
# Queue fully drained — no canned slot was consumed by event detection.
assert mock._canned == []
with open_db(tmp_path / "test.db") as conn:
for kind in ("event_started", "event_completed", "event_cancelled"):
count = conn.execute(
"SELECT COUNT(*) FROM event_log WHERE kind = ?", (kind,)
).fetchone()[0]
assert count == 0, f"expected zero {kind} events, got {count}"
# ---------------------------------------------------------------------------
# Phase 3 (T62) — natural-language skip-command surface.
#
# The classifier may flag prose as a time-skip directive via
# ``ParsedTurn.intent``. Elision runs through the shared controller in
# :mod:`chat.web.skip` and short-circuits the regular narrative path;
# jump returns 422 directing the user to the drawer's structured form
# (Phase 3 simpler path — natural-language jump time derivation is too
# fragile for v1 without the structured surface).
# ---------------------------------------------------------------------------
def test_elision_skip_via_natural_language(app_state_setup, tmp_path):
"""User prose 'skip to when we arrive at the park' classifies as
``intent='skip_elision'``. The post_turn handler short-circuits the
narrative path, advances the chat clock by an hour stub, appends a
``time_skip_elision`` event AND an ``assistant_turn`` carrying the
canned narration. No ``user_turn`` is emitted on the skip path.
Canned queue: 1 parse_turn (intent=skip_elision) + 1 narration
string (consumed by ``narrate_skip``). No state-update / scene-close
/ event-detection slots those branches are bypassed entirely.
"""
_seed(tmp_path / "test.db")
canned_parse = json.dumps(
{
"segments": [
{"kind": "dialogue", "text": "skip to when we arrive at the park"}
],
"intent": "skip_elision",
"landing_state_hint": "we arrive at the park",
}
)
canned_narration = "We pull up to the park entrance, sun low in the sky."
mock = _override_llm([canned_parse, canned_narration])
try:
response = app_state_setup.post(
"/chats/chat_bot_a/turns",
data={"prose": "skip to when we arrive at the park"},
)
assert response.status_code == 204
finally:
app.dependency_overrides.clear()
# Both canned slots drained — no other classifier branches ran.
assert mock._canned == []
with open_db(tmp_path / "test.db") as conn:
# time_skip_elision landed.
skip_rows = conn.execute(
"SELECT payload_json FROM event_log "
"WHERE kind = 'time_skip_elision' ORDER BY id"
).fetchall()
assert len(skip_rows) == 1
sp = json.loads(skip_rows[0][0])
assert sp["chat_id"] == "chat_bot_a"
# 1-hour stub from the seeded chat clock (20:00 -> 21:00).
assert sp["new_time"].startswith("2026-04-26T21:00:00")
# Chat clock advanced via the projector.
from chat.state.world import get_chat
chat = get_chat(conn, "chat_bot_a")
assert chat["time"].startswith("2026-04-26T21:00:00")
# An assistant_turn carrying the canned narration was appended.
turn_rows = conn.execute(
"SELECT payload_json FROM event_log "
"WHERE kind = 'assistant_turn' ORDER BY id"
).fetchall()
assert len(turn_rows) == 1
tp = json.loads(turn_rows[0][0])
assert tp["chat_id"] == "chat_bot_a"
assert tp["text"] == canned_narration
assert tp["speaker_id"] == "bot_a"
assert tp["truncated"] is False
# No user_turn lands on the skip path — the natural-language
# skip is a command, not a beat the bots should remember.
user_count = conn.execute(
"SELECT COUNT(*) FROM event_log WHERE kind = 'user_turn'"
).fetchone()[0]
assert user_count == 0
def test_jump_skip_via_natural_language_returns_422(app_state_setup, tmp_path):
"""User prose 'next morning' classifies as ``intent='skip_jump'``.
The handler returns 422 with a guidance payload pointing the author
at the drawer's structured jump form. No event is emitted — the
drawer form is the only entry point for jump skips in Phase 3.
"""
_seed(tmp_path / "test.db")
canned_parse = json.dumps(
{
"segments": [{"kind": "dialogue", "text": "next morning"}],
"intent": "skip_jump",
"landing_state_hint": "",
}
)
# Only one canned slot — parse — because the 422 fallback short-
# circuits before any other classifier runs.
mock = _override_llm([canned_parse])
try:
response = app_state_setup.post(
"/chats/chat_bot_a/turns", data={"prose": "next morning"}
)
assert response.status_code == 422
body = response.json()
# Guidance payload mentions the drawer so the client can surface
# the right CTA; we don't pin the exact wording.
assert "drawer" in body.get("error", "").lower()
finally:
app.dependency_overrides.clear()
# Parse slot consumed; no follow-on classifier calls.
assert mock._canned == []
with open_db(tmp_path / "test.db") as conn:
for kind in (
"user_turn",
"assistant_turn",
"time_skip_elision",
"time_skip_jump",
):
count = conn.execute(
"SELECT COUNT(*) FROM event_log WHERE kind = ?", (kind,)
).fetchone()[0]
assert count == 0, f"expected zero {kind} on jump-via-NL, got {count}"
def test_skip_command_does_not_run_narrative_classifier(
app_state_setup, tmp_path, monkeypatch
):
"""The skip dispatch branch must bypass the narrative-prompt assembly
entirely. We monkeypatch ``assemble_narrative_prompt`` (re-bound on
the ``chat.web.turns`` module since the handler imports it by name)
and assert the call count is zero after the elision skip lands.
"""
_seed(tmp_path / "test.db")
canned_parse = json.dumps(
{
"segments": [
{"kind": "dialogue", "text": "skip to when we arrive at the park"}
],
"intent": "skip_elision",
"landing_state_hint": "we arrive at the park",
}
)
canned_narration = "We arrive moments later."
mock = _override_llm([canned_parse, canned_narration])
call_counter = {"n": 0}
def _spy(*args, **kwargs):
call_counter["n"] += 1
return []
# Patch the symbol at the handler's import site so we can assert
# the skip path bypasses prompt assembly even when the symbol still
# exists in the module namespace.
from chat.web import turns as turns_mod
monkeypatch.setattr(turns_mod, "assemble_narrative_prompt", _spy)
try:
response = app_state_setup.post(
"/chats/chat_bot_a/turns",
data={"prose": "skip to when we arrive at the park"},
)
assert response.status_code == 204
finally:
app.dependency_overrides.clear()
assert mock._canned == []
assert call_counter["n"] == 0, (
"assemble_narrative_prompt was called on the skip path; the "
"natural-language skip dispatch must bypass narrative assembly."
)
+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