merge: T45 per-POV summaries on close for each present witness

chat/db/migrations/0008_group_node.sql

@@ -0,0 +1,8 @@
+CREATE TABLE group_node (
+    chat_id TEXT PRIMARY KEY,
+    members_json TEXT NOT NULL,
+    summary TEXT NOT NULL DEFAULT '',
+    dynamic TEXT NOT NULL DEFAULT '',
+    threads_json TEXT NOT NULL DEFAULT '[]',
+    updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);

chat/services/interjection.py

@@ -0,0 +1,100 @@
+"""Interjection classifier service (T39).
+
+Per Requirements §6.2, when a guest is present and the addressee bot has
+just spoken, the *non-addressee* bot may follow on with a brief
+interjection beat. This service decides whether that interjection
+fires. Conservative bias: most turns return ``should_interject=False``
+— the addressee has the floor and an interjection is the exception.
+Trigger ``True`` only when the silent witness's character, given their
+persona and edges, would plausibly speak up: jealousy, surprise, strong
+agreement worth voicing, correcting a factual falsehood, urgency.
+
+T44 (turn flow) calls this and, on ``True``, generates the brief
+follow-on response as the silent witness. Classifier failure falls back
+to ``should_interject=False`` with ``reason="fallback"`` so the chat
+keeps moving (§3.3 graceful-degradation rule); callers that care can
+distinguish a real "no" from a degraded "no" by the reason string.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+from chat.llm.classify import classify
+from chat.llm.client import LLMClient
+
+
+class InterjectionDecision(BaseModel):
+    """Whether the silent witness interjects, plus a short reason.
+
+    Defaults are a deliberate no-op: ``should_interject=False`` with an
+    empty reason. The classifier-failure fallback uses
+    ``reason="fallback"`` so it's distinguishable from a real "no".
+    """
+
+    should_interject: bool = False
+    reason: str = ""
+
+
+_SYSTEM = (
+    "You decide whether a silent witness character interjects after the "
+    "addressee character finishes speaking. STRONGLY default to false — "
+    "the addressee has the floor and most turns should NOT have an "
+    "interjection. Only return true when the silent witness's character, "
+    "given their persona and edges, would plausibly speak up: jealousy, "
+    "surprise, strong agreement worth voicing, correcting a factual "
+    "falsehood, urgency. Output strict JSON matching the schema."
+)
+
+
+async def detect_interjection(
+    client: LLMClient,
+    *,
+    classifier_model: str,
+    addressee_name: str,
+    addressee_just_said: str,
+    silent_witness_name: str,
+    silent_witness_persona: str,
+    silent_witness_edge_to_addressee: dict,  # {affinity, trust, summary}
+    silent_witness_edge_to_you: dict,
+    you_just_said: str,
+    timeout_s: float = 30.0,
+) -> InterjectionDecision:
+    """Decide whether the silent witness bot interjects after the addressee
+    finishes speaking.
+
+    The two ``silent_witness_edge_*`` dicts carry the silent witness's
+    directed edges toward the addressee and toward the user ("you"),
+    each shaped ``{affinity: int, trust: int, summary: str}``. Missing
+    keys fall back to a 50/50 baseline with an empty summary so this
+    function tolerates partially-populated edge state without raising.
+    """
+    user = (
+        f"You said: {you_just_said}\n\n"
+        f"{addressee_name} just said: {addressee_just_said}\n\n"
+        f"Silent witness: {silent_witness_name}\n"
+        f"Persona: {silent_witness_persona}\n"
+        f"Edge {silent_witness_name} -> {addressee_name}: "
+        f"affinity={silent_witness_edge_to_addressee.get('affinity', 50)}, "
+        f"trust={silent_witness_edge_to_addressee.get('trust', 50)}, "
+        f"summary={silent_witness_edge_to_addressee.get('summary', '')}\n"
+        f"Edge {silent_witness_name} -> you: "
+        f"affinity={silent_witness_edge_to_you.get('affinity', 50)}, "
+        f"trust={silent_witness_edge_to_you.get('trust', 50)}, "
+        f"summary={silent_witness_edge_to_you.get('summary', '')}\n\n"
+        f"Should {silent_witness_name} interject?"
+    )
+    return await classify(
+        client,
+        model=classifier_model,
+        system=_SYSTEM,
+        user=user,
+        schema=InterjectionDecision,
+        default=InterjectionDecision(
+            should_interject=False, reason="fallback"
+        ),
+        timeout_s=timeout_s,
+    )
+
+
+__all__ = ["InterjectionDecision", "detect_interjection"]

chat/services/memory_write.py

@@ -76,3 +76,103 @@ def record_turn_memory(
     ).fetchone()
     memory_id = row[0] if row else None
     return event_id, memory_id
+
+
+def _write_one_memory(
+    conn: Connection,
+    *,
+    owner_id: str,
+    chat_id: str,
+    narrative_text: str,
+    witness_you: int,
+    witness_host: int,
+    witness_guest: int,
+    scene_id: int | None,
+    chat_clock_at: str | None,
+    source: str,
+    significance: int,
+) -> tuple[int, int | None]:
+    """Append a single ``memory_written`` event for ``owner_id`` and return
+    ``(event_id, memory_id)`` for the projected row."""
+    payload: dict = {
+        "owner_id": owner_id,
+        "chat_id": chat_id,
+        "pov_summary": narrative_text,
+        "witness_you": witness_you,
+        "witness_host": witness_host,
+        "witness_guest": witness_guest,
+        "source": source,
+        "reliability": 1.0,
+        "significance": significance,
+        "pinned": 0,
+        "auto_pinned": 0,
+    }
+    if scene_id is not None:
+        payload["scene_id"] = scene_id
+    if chat_clock_at is not None:
+        payload["chat_clock_at"] = chat_clock_at
+
+    event_id = append_and_apply(conn, kind="memory_written", payload=payload)
+    row = conn.execute(
+        "SELECT id FROM memories "
+        "WHERE owner_id = ? AND chat_id = ? "
+        "ORDER BY id DESC LIMIT 1",
+        (owner_id, chat_id),
+    ).fetchone()
+    memory_id = row[0] if row else None
+    return event_id, memory_id
+
+
+def record_turn_memory_for_present(
+    conn: Connection,
+    *,
+    chat_id: str,
+    host_bot_id: str,
+    guest_bot_id: str | None,
+    narrative_text: str,
+    scene_id: int | None = None,
+    chat_clock_at: str | None = None,
+    source: str = "direct",
+    significance: int = 1,
+) -> dict[str, tuple[int, int | None]]:
+    """Write a ``memory_written`` event for each present bot witness.
+
+    Host is always written. Guest is written iff ``guest_bot_id is not
+    None``. Witness flags are ``[you=1, host=1, guest=1]`` when a guest
+    is present, ``[you=1, host=1, guest=0]`` otherwise.
+
+    Returns a mapping ``{bot_id: (event_id, memory_id)}`` so callers can
+    look up the freshly-projected memory id per owner without re-querying
+    the database.
+    """
+    witness_guest = 1 if guest_bot_id is not None else 0
+
+    result: dict[str, tuple[int, int | None]] = {}
+    result[host_bot_id] = _write_one_memory(
+        conn,
+        owner_id=host_bot_id,
+        chat_id=chat_id,
+        narrative_text=narrative_text,
+        witness_you=1,
+        witness_host=1,
+        witness_guest=witness_guest,
+        scene_id=scene_id,
+        chat_clock_at=chat_clock_at,
+        source=source,
+        significance=significance,
+    )
+    if guest_bot_id is not None:
+        result[guest_bot_id] = _write_one_memory(
+            conn,
+            owner_id=guest_bot_id,
+            chat_id=chat_id,
+            narrative_text=narrative_text,
+            witness_you=1,
+            witness_host=1,
+            witness_guest=1,
+            scene_id=scene_id,
+            chat_clock_at=chat_clock_at,
+            source=source,
+            significance=significance,
+        )
+    return result

chat/services/multi_state_update.py

@@ -0,0 +1,62 @@
+"""Multi-entity state-update coordinator (T40).
+
+Wraps single-pair compute_state_update to run state updates for ALL
+directed pairs of present entities. With 3 present entities (you, host,
+guest) that's 6 directed pairs. With 2 present (you, host) it's 2 pairs.
+
+Calls run sequentially to respect Featherless's 2-connection cap (the
+client-level semaphore would serialize them anyway, but doing it here
+keeps the failure surface clean — a hung pair doesn't queue behind
+itself).
+"""
+
+from __future__ import annotations
+
+from chat.llm.client import LLMClient
+from chat.services.state_update import StateUpdate, compute_state_update
+
+
+async def compute_state_updates_for_present(
+    client: LLMClient,
+    *,
+    classifier_model: str,
+    present_ids: list[str],
+    present_names: dict[str, str],
+    personas: dict[str, str],
+    prior_edges: dict[tuple[str, str], dict],
+    recent_dialogue: list[dict],
+    timeout_s: float = 30.0,
+) -> list[tuple[str, str, StateUpdate]]:
+    """Run compute_state_update for every directed pair (src != tgt) over
+    ``present_ids``. Returns list of ``(source_id, target_id, update)``
+    tuples in the natural iteration order over ``present_ids x present_ids``.
+
+    A single failing pair falls back to the schema-default StateUpdate
+    (zero deltas, empty facts) inside ``compute_state_update``; the batch
+    keeps going.
+    """
+    out: list[tuple[str, str, StateUpdate]] = []
+    for src in present_ids:
+        for tgt in present_ids:
+            if src == tgt:
+                continue
+            edge = prior_edges.get((src, tgt), {})
+            update = await compute_state_update(
+                client,
+                model=classifier_model,
+                source_id=src,
+                target_id=tgt,
+                source_name=present_names.get(src, src),
+                source_persona=personas.get(src, "") or "",
+                target_name=present_names.get(tgt, tgt),
+                prior_affinity=int(edge.get("affinity", 50)),
+                prior_trust=int(edge.get("trust", 50)),
+                prior_summary=edge.get("summary", "") or "",
+                recent_dialogue=recent_dialogue,
+                timeout_s=timeout_s,
+            )
+            out.append((src, tgt, update))
+    return out
+
+
+__all__ = ["compute_state_updates_for_present"]

chat/services/prompt.py

@@ -37,6 +37,7 @@ 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.group_node import get_group_node
 from chat.state.memory import search_memories
 from chat.state.world import (
     active_scene,
@@ -206,6 +207,26 @@ def _build_previous_scene_block(pov_summary: str | None) -> str | None:
     return "PREVIOUS SCENE SUMMARY:\n" + pov_summary
 
 
+def _build_group_node_block(group_node: dict | None) -> str | None:
+    """Render the group-node summary + dynamic as a SHOULD-tier block.
+
+    Used only in 3-entity scenes (you + host + guest). Returns None when
+    the row is missing or both summary and dynamic are empty.
+    """
+    if not group_node:
+        return None
+    summary = (group_node.get("summary") or "").strip()
+    dynamic = (group_node.get("dynamic") or "").strip()
+    if not summary and not dynamic:
+        return None
+    lines = ["Group dynamic:"]
+    if summary:
+        lines.append(f"- Summary: {summary}")
+    if dynamic:
+        lines.append(f"- Dynamic: {dynamic}")
+    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 "
@@ -287,6 +308,7 @@ def assemble_narrative_prompt(
     budget_soft: int = 6000,
     budget_hard: int = 8000,
     encoding_name: str = "cl100k_base",
+    guest_id: str | None = None,
 ) -> list[Message]:
     """Assemble the narrative prompt for ``speaker_bot_id`` to respond.
 
@@ -313,6 +335,15 @@ def assemble_narrative_prompt(
     if chat is None:
         raise ValueError(f"chat_id {chat_id!r} not found")
 
+    # Auto-detect guest from chat state when caller didn't pass one.
+    # Phase 1 chats have ``guest_bot_id is None``; the auto-detect is a
+    # no-op there and the function behaves exactly as before.
+    if guest_id is None:
+        guest_id = chat.get("guest_bot_id")
+    # A speaker addressing themself as guest doesn't add a third party.
+    if guest_id is not None and guest_id == speaker_bot_id:
+        guest_id = None
+
     you = get_you(conn)
     addressee_id, addressee_name = _resolve_addressee(conn, addressee, you)
 
@@ -325,9 +356,10 @@ def assemble_narrative_prompt(
         addressee_name,
     )
 
-    # Activity for present entities. Phase 1: you + speaker bot. (When a
-    # guest is added in Phase 1+, callers that know about it can pass
-    # extra activities via a future hook; for now we keep it strict.)
+    # Activity for present entities. Core (MUST): you + speaker bot.
+    # Phase 2 (SHOULD-tier): when a third party (guest) is present in
+    # the chat, append their activity in a separate block so it can be
+    # trimmed independently under tight budget.
     activities: list[dict] = []
     you_act = get_activity(conn, "you")
     if you_act is not None:
@@ -341,6 +373,34 @@ def assemble_narrative_prompt(
         activities.append(bot_act)
     activity_block = _build_activity_block(activities)
 
+    # SHOULD-tier guest activity extension (Phase 2 / Task 43).
+    guest_activity_block: str | None = None
+    if guest_id is not None:
+        guest_act = get_activity(conn, guest_id)
+        if guest_act is not None:
+            guest_act = dict(guest_act)
+            guest_bot = get_bot(conn, guest_id)
+            guest_act["_display_name"] = (
+                guest_bot["name"] if guest_bot else guest_id
+            )
+            guest_activity_block = _build_activity_block([guest_act])
+
+    # SHOULD-tier group-node block (Phase 2 / Task 43): rendered only
+    # when the group_node row is present AND it covers all three of
+    # you + host + guest (per the Task 43 spec).
+    group_node_block: str | None = None
+    if guest_id is not None:
+        gn = get_group_node(conn, chat_id)
+        if gn is not None:
+            members = set(gn.get("members") or [])
+            host_id = chat.get("host_bot_id")
+            required = {"you"}
+            if host_id is not None:
+                required.add(host_id)
+            required.add(guest_id)
+            if required.issubset(members):
+                group_node_block = _build_group_node_block(gn)
+
     container = None
     if chat.get("active_scene_id"):
         scene = get_scene(conn, chat["active_scene_id"])
@@ -421,6 +481,8 @@ def assemble_narrative_prompt(
         include_previous_scene: bool,
         include_memories_top_k: int,
         dialogue_keep: int,
+        include_guest_activity: bool = True,
+        include_group_node: bool = True,
     ) -> tuple[str, int, list[dict]]:
         # dialogue: keep the last `dialogue_keep` turns verbatim; older
         # turns become an "earlier:" placeholder line.
@@ -447,6 +509,8 @@ def assemble_narrative_prompt(
             other_edges_block if include_other_edges else None,
             scene_block,
             activity_block,
+            guest_activity_block if include_guest_activity else None,
+            group_node_block if include_group_node else None,
             prev_block,
             memories_block,
             dialogue_block,
@@ -463,12 +527,25 @@ def assemble_narrative_prompt(
     nice_memories_k = min(4, len(memory_summaries))
     include_prev = previous_scene_summary is not None
     include_other = other_edges_block is not None
+    include_guest_activity = guest_activity_block is not None
+    include_group_node = group_node_block is not None
 
-    body, total, _ = assemble(
-        include_other_edges=include_other,
-        include_previous_scene=include_prev,
-        include_memories_top_k=nice_memories_k,
-        dialogue_keep=nice_dialogue_keep,
+    def _build(*, prev: bool, mem_k: int, dlg: int, other: bool,
+               guest_act: bool, group: bool) -> tuple[str, int]:
+        body, total, _ = assemble(
+            include_other_edges=other,
+            include_previous_scene=prev,
+            include_memories_top_k=mem_k,
+            dialogue_keep=dlg,
+            include_guest_activity=guest_act,
+            include_group_node=group,
+        )
+        return body, total
+
+    body, total = _build(
+        prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
+        other=include_other, guest_act=include_guest_activity,
+        group=include_group_node,
     )
 
     # If under soft, we're done.
@@ -478,34 +555,31 @@ def assemble_narrative_prompt(
     # Drop NICE in order: previous scene → memories beyond top-2 →
     # older dialogue turns (collapse to 4).
     if include_prev:
-        body, total, _ = assemble(
-            include_other_edges=include_other,
-            include_previous_scene=False,
-            include_memories_top_k=nice_memories_k,
-            dialogue_keep=nice_dialogue_keep,
-        )
         include_prev = False
+        body, total = _build(
+            prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
+            other=include_other, guest_act=include_guest_activity,
+            group=include_group_node,
+        )
         if total <= budget_soft:
             return _emit(body, user_turn_prose)
 
     if nice_memories_k > 2:
         nice_memories_k = 2
-        body, total, _ = assemble(
-            include_other_edges=include_other,
-            include_previous_scene=False,
-            include_memories_top_k=nice_memories_k,
-            dialogue_keep=nice_dialogue_keep,
+        body, total = _build(
+            prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
+            other=include_other, guest_act=include_guest_activity,
+            group=include_group_node,
         )
         if total <= budget_soft:
             return _emit(body, user_turn_prose)
 
     if nice_dialogue_keep > baseline_keep:
         nice_dialogue_keep = baseline_keep
-        body, total, _ = assemble(
-            include_other_edges=include_other,
-            include_previous_scene=False,
-            include_memories_top_k=nice_memories_k,
-            dialogue_keep=nice_dialogue_keep,
+        body, total = _build(
+            prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
+            other=include_other, guest_act=include_guest_activity,
+            group=include_group_node,
         )
         if total <= budget_soft:
             return _emit(body, user_turn_prose)
@@ -513,21 +587,37 @@ def assemble_narrative_prompt(
     # Drop more NICE until we're under hard: memories all the way to 0.
     while nice_memories_k > 0 and total > budget_hard:
         nice_memories_k = max(0, nice_memories_k - 1)
-        body, total, _ = assemble(
-            include_other_edges=include_other,
-            include_previous_scene=False,
-            include_memories_top_k=nice_memories_k,
-            dialogue_keep=nice_dialogue_keep,
+        body, total = _build(
+            prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
+            other=include_other, guest_act=include_guest_activity,
+            group=include_group_node,
+        )
+
+    # Drop SHOULD-tier blocks in order: guest activity → group node →
+    # other edges. (Guest activity goes first per Task 43 spec — it's
+    # the most expendable additive context.)
+    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, guest_act=include_guest_activity,
+            group=include_group_node,
+        )
+
+    if include_group_node and total > budget_hard:
+        include_group_node = False
+        body, total = _build(
+            prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
+            other=include_other, guest_act=include_guest_activity,
+            group=include_group_node,
         )
 
-    # Drop SHOULD: other edges.
     if include_other and total > budget_hard:
         include_other = False
-        body, total, _ = assemble(
-            include_other_edges=False,
-            include_previous_scene=False,
-            include_memories_top_k=nice_memories_k,
-            dialogue_keep=nice_dialogue_keep,
+        body, total = _build(
+            prev=include_prev, mem_k=nice_memories_k, dlg=nice_dialogue_keep,
+            other=include_other, guest_act=include_guest_activity,
+            group=include_group_node,
         )
 
     if total > budget_hard:

chat/services/relationship_seed.py

@@ -0,0 +1,107 @@
+"""Parse user-supplied "have they met?" prose into per-direction seed
+content for two bots' edges (T38).
+
+Per Requirements §5.2, when two bots first co-appear in a chat, the user
+is offered a small drawer asking "Have they met before? If yes, write a
+short prose seed describing how." That prose lands here and is parsed
+into a :class:`RelationshipSeed` whose two halves populate the
+``botA -> botB`` and ``botB -> botA`` edges respectively (summary,
+initial knowledge facts, and small affinity/trust deltas around the
+default 50/50 baseline).
+
+The two directions can differ — A may know more about B than B knows
+about A, or A may trust B less than the reverse — so the schema carries
+both halves independently.
+
+Empty/whitespace-only prose short-circuits to a default
+``RelationshipSeed`` (all zeroes, empty strings); the caller treats
+that as "they haven't met" and writes no edge content. The wrapper uses
+:func:`chat.llm.classify.classify` with ``default=RelationshipSeed()``
+so a flapping classifier degrades to the same no-op rather than
+blocking the chat-creation flow (§3.3 graceful-degradation rule).
+
+T42 (the inter-bot relationship drawer) calls this from the route layer.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from chat.llm.classify import classify
+from chat.llm.client import LLMClient
+
+
+class RelationshipSeed(BaseModel):
+    """Structured per-direction seed for two bots' edges.
+
+    Defaults are a deliberate no-op: empty summaries, empty knowledge
+    lists, zero deltas. Both the empty-prose short-circuit and the
+    classifier-failure fallback return this default so the caller can
+    treat them identically.
+    """
+
+    a_to_b_summary: str = ""
+    a_to_b_knowledge_facts: list[str] = Field(default_factory=list)
+    a_to_b_affinity_delta: int = 0  # signed, -10..+10 typical
+    a_to_b_trust_delta: int = 0
+    b_to_a_summary: str = ""
+    b_to_a_knowledge_facts: list[str] = Field(default_factory=list)
+    b_to_a_affinity_delta: int = 0
+    b_to_a_trust_delta: int = 0
+
+
+_SYSTEM = (
+    "You parse a short prose seed describing how two characters know each "
+    "other into structured per-direction edge content. For each direction "
+    "(A -> B, B -> A) extract: summary (one sentence from that POV), "
+    "knowledge_facts (list of factual claims that direction can carry "
+    "into future scenes), affinity_delta (-10..+10 — small adjustments to "
+    "the default 50/50 baseline), trust_delta (-10..+10). Default deltas "
+    "to 0 when prose is neutral. The two directions can differ — A may "
+    "trust B more than B trusts A. Output strict JSON matching the schema."
+)
+
+
+async def seed_inter_bot_edges(
+    client: LLMClient,
+    *,
+    classifier_model: str,
+    bot_a_id: str,
+    bot_a_name: str,
+    bot_b_id: str,
+    bot_b_name: str,
+    relationship_prose: str,
+    timeout_s: float = 30.0,
+) -> RelationshipSeed:
+    """Parse user-supplied prose into structured edge content for both
+    directed pairs.
+
+    Empty/whitespace prose short-circuits to an empty
+    :class:`RelationshipSeed` (the caller treats this as "they haven't
+    met" and writes no edge content). Classifier failure also returns
+    the default — see module docstring for the rationale.
+
+    The ``bot_a_id`` / ``bot_b_id`` arguments are accepted for symmetry
+    with the caller (T42's drawer route uses them when emitting
+    ``edge_update`` events); they're embedded in the prompt alongside
+    the names so the classifier can disambiguate when names collide.
+    """
+    if not relationship_prose or not relationship_prose.strip():
+        return RelationshipSeed()
+    user = (
+        f"Bot A: {bot_a_name} (id={bot_a_id})\n"
+        f"Bot B: {bot_b_name} (id={bot_b_id})\n\n"
+        f"Prose seed:\n{relationship_prose.strip()}"
+    )
+    return await classify(
+        client,
+        model=classifier_model,
+        system=_SYSTEM,
+        user=user,
+        schema=RelationshipSeed,
+        default=RelationshipSeed(),
+        timeout_s=timeout_s,
+    )
+
+
+__all__ = ["RelationshipSeed", "seed_inter_bot_edges"]

chat/services/scene_summarize.py

@@ -156,64 +156,50 @@ def _read_recent_dialogue(
     return out
 
 
-async def apply_scene_close_summary(
+async def _summarize_and_apply_for_witness(
     conn: Connection,
     client: LLMClient,
     *,
     classifier_model: str,
     chat_id: str,
     scene_id: int,
-    host_bot_id: str,
-    timeout_s: float = 10.0,
+    bot_id: str,
+    you_name: str,
+    dialogue: list[dict],
+    timeout_s: float,
 ) -> ScenePOVSummary:
-    """Drive the per-POV summary pipeline after ``scene_closed``.
+    """Run :func:`summarize_scene` for one bot witness and apply the
+    three projected updates (memory pov_summary rewrite, edge summary
+    overwrite, edge knowledge_facts append).
 
-    Steps (Phase 1, single-bot):
-      1. Gather the closing scene's dialogue from the event_log.
-      2. Run :func:`summarize_scene` for the host bot.
-      3. Rewrite each scene-bound memory's ``pov_summary`` via
-         ``manual_edit`` (target_kind ``memory_pov_summary``), capturing
-         the prior value for §6.4 reversibility.
-      4. Update the bot->you edge summary via ``manual_edit`` with the
-         new ``edge_summary`` target_kind. v1 combines prior + new by
-         concatenation — the classifier's ``relationship_summary`` is
-         already phrased as a continuation.
-      5. Append any new knowledge_facts to the same edge via
-         ``edge_update``.
-
-    Tolerant of missing pieces: no memories -> skip step 3 silently;
-    no edge row -> skip step 4; empty knowledge_facts -> skip step 5.
-    The classifier's empty default flows through harmlessly.
+    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).
     """
-    # Local imports to keep the module-level surface tight and avoid
-    # any chance of a circular dep through chat.state.*.
     from chat.state.edges import get_edge
-    from chat.state.entities import get_bot, get_you
+    from chat.state.entities import get_bot
 
-    host_bot = get_bot(conn, host_bot_id) or {"name": host_bot_id, "persona": ""}
-    you_entity = get_you(conn) or {"name": "you", "persona": ""}
+    bot = get_bot(conn, bot_id) or {"name": bot_id, "persona": ""}
 
-    dialogue = _read_recent_dialogue(conn, chat_id)
-
-    edge_b2y = get_edge(conn, host_bot_id, "you")
+    edge_b2y = get_edge(conn, bot_id, "you")
     prior_summary = (edge_b2y or {}).get("summary", "") or ""
 
     pov = await summarize_scene(
         client,
         model=classifier_model,
-        bot_name=host_bot.get("name", host_bot_id),
-        bot_persona=host_bot.get("persona", "") or "",
-        you_name=you_entity.get("name", "you") or "you",
+        bot_name=bot.get("name", bot_id),
+        bot_persona=bot.get("persona", "") or "",
+        you_name=you_name,
         prior_edge_summary=prior_summary,
         dialogue=dialogue,
         timeout_s=timeout_s,
     )
 
-    # Update memories belonging to the closed scene for the host bot.
+    # Update memories belonging to the closed scene for this witness.
     cur = conn.execute(
         "SELECT id, pov_summary FROM memories "
         "WHERE scene_id = ? AND owner_id = ?",
-        (scene_id, host_bot_id),
+        (scene_id, bot_id),
     )
     for memory_id, prior_pov in cur.fetchall():
         if not pov.summary:
@@ -231,7 +217,7 @@ async def apply_scene_close_summary(
             },
         )
 
-    # Update the bot->you edge summary if we have an edge row and a
+    # Update this bot->you edge summary if we have an edge row and a
     # non-empty relationship_summary to merge.
     if edge_b2y is not None and pov.relationship_summary:
         new_summary = (
@@ -245,7 +231,7 @@ async def apply_scene_close_summary(
             payload={
                 "target_kind": "edge_summary",
                 "target_id": {
-                    "source_id": host_bot_id,
+                    "source_id": bot_id,
                     "target_id": "you",
                 },
                 "prior_value": prior_summary,
@@ -253,13 +239,13 @@ async def apply_scene_close_summary(
             },
         )
 
-    # Append knowledge_facts to the bot->you edge if present.
+    # Append knowledge_facts to this bot->you edge if present.
     if pov.knowledge_facts:
         append_and_apply(
             conn,
             kind="edge_update",
             payload={
-                "source_id": host_bot_id,
+                "source_id": bot_id,
                 "target_id": "you",
                 "chat_id": chat_id,
                 "knowledge_facts": list(pov.knowledge_facts),
@@ -267,3 +253,107 @@ async def apply_scene_close_summary(
         )
 
     return pov
+
+
+async def apply_scene_close_summary(
+    conn: Connection,
+    client: LLMClient,
+    *,
+    classifier_model: str,
+    chat_id: str,
+    scene_id: int,
+    host_bot_id: str,
+    timeout_s: float = 10.0,
+) -> ScenePOVSummary:
+    """Drive the per-POV summary pipeline after ``scene_closed``.
+
+    Phase 1 (single-bot) behavior — the host bot is summarized once and
+    the result drives memory + edge rewrites — is preserved exactly when
+    the chat has no guest. T45 extends this to fan out across each
+    present bot witness when a guest is also in the room:
+
+      1. Gather the closing scene's dialogue from the event_log.
+      2. For each present witness (host + guest if any), run
+         :func:`summarize_scene` once with that witness's persona and
+         their own prior ``bot -> you`` edge summary.
+      3. For each witness independently:
+         a. Rewrite each scene-bound memory's ``pov_summary`` via
+            ``manual_edit`` (target_kind ``memory_pov_summary``).
+         b. Update that witness's ``bot -> you`` edge summary via
+            ``manual_edit`` (target_kind ``edge_summary``). v2 combines
+            prior + classifier ``relationship_summary`` by simple
+            concatenation.
+         c. Append any ``knowledge_facts`` to the same edge via
+            ``edge_update``.
+      4. If a ``group_node`` row exists for this chat, append a
+         ``group_node_updated`` event whose ``summary`` is the naive
+         per-POV concat ``f"{name}: {summary}\\n\\n..."``. A true
+         LLM-merged group view is deferred to Phase 2.5; ``dynamic``
+         is left empty here for v2 (Phase 3 polishes it).
+
+    The host's :class:`ScenePOVSummary` is returned to preserve the
+    Phase 1 callers' contract.
+    """
+    # Local imports to keep the module-level surface tight and avoid
+    # any chance of a circular dep through chat.state.*.
+    from chat.state.entities import get_bot, get_you
+    from chat.state.group_node import get_group_node
+    from chat.state.world import get_chat
+
+    you_entity = get_you(conn) or {"name": "you", "persona": ""}
+    you_name = you_entity.get("name", "you") or "you"
+
+    chat = get_chat(conn, chat_id) or {}
+    guest_bot_id = chat.get("guest_bot_id")
+
+    dialogue = _read_recent_dialogue(conn, chat_id)
+
+    host_pov = await _summarize_and_apply_for_witness(
+        conn,
+        client,
+        classifier_model=classifier_model,
+        chat_id=chat_id,
+        scene_id=scene_id,
+        bot_id=host_bot_id,
+        you_name=you_name,
+        dialogue=dialogue,
+        timeout_s=timeout_s,
+    )
+
+    guest_pov: ScenePOVSummary | None = None
+    if guest_bot_id is not None:
+        guest_pov = await _summarize_and_apply_for_witness(
+            conn,
+            client,
+            classifier_model=classifier_model,
+            chat_id=chat_id,
+            scene_id=scene_id,
+            bot_id=guest_bot_id,
+            you_name=you_name,
+            dialogue=dialogue,
+            timeout_s=timeout_s,
+        )
+
+    # Group node update: naive per-POV concat for v2. Only fires when
+    # both POVs ran (i.e. the guest is present) and a group_node row
+    # exists for this chat.
+    if guest_pov is not None and get_group_node(conn, chat_id) is not None:
+        host_bot = get_bot(conn, host_bot_id) or {"name": host_bot_id}
+        guest_bot = get_bot(conn, guest_bot_id) or {"name": guest_bot_id}
+        host_name = host_bot.get("name", host_bot_id) or host_bot_id
+        guest_name = guest_bot.get("name", guest_bot_id) or guest_bot_id
+        group_summary = (
+            f"{host_name}: {host_pov.summary}\n\n"
+            f"{guest_name}: {guest_pov.summary}"
+        )
+        append_and_apply(
+            conn,
+            kind="group_node_updated",
+            payload={
+                "chat_id": chat_id,
+                "summary": group_summary,
+                "dynamic": "",
+            },
+        )
+
+    return host_pov

chat/state/group_node.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+import json
+from sqlite3 import Connection
+from chat.eventlog.projector import on
+from chat.eventlog.log import Event
+
+
+@on("group_node_initialized")
+def _apply_group_node_initialized(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "INSERT OR REPLACE INTO group_node "
+        "(chat_id, members_json, summary, dynamic, threads_json) "
+        "VALUES (?, ?, ?, ?, ?)",
+        (
+            p["chat_id"],
+            json.dumps(p["members"]),
+            p.get("summary", ""),
+            p.get("dynamic", ""),
+            json.dumps(p.get("threads", [])),
+        ),
+    )
+
+
+@on("group_node_updated")
+def _apply_group_node_updated(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "UPDATE group_node SET summary = ?, dynamic = ?, updated_at = datetime('now') "
+        "WHERE chat_id = ?",
+        (p.get("summary", ""), p.get("dynamic", ""), p["chat_id"]),
+    )
+
+
+def get_group_node(conn: Connection, chat_id: str) -> dict | None:
+    row = conn.execute(
+        "SELECT chat_id, members_json, summary, dynamic, threads_json, updated_at "
+        "FROM group_node WHERE chat_id = ?",
+        (chat_id,),
+    ).fetchone()
+    if not row:
+        return None
+    return {
+        "chat_id": row[0],
+        "members": json.loads(row[1]),
+        "summary": row[2],
+        "dynamic": row[3],
+        "threads": json.loads(row[4]),
+        "updated_at": row[5],
+    }

chat/state/world.py

@@ -29,6 +29,24 @@ def _apply_chat_created(conn: Connection, e: Event) -> None:
     )
 
 
+@on("guest_added")
+def _apply_guest_added(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "UPDATE chats SET guest_bot_id = ? WHERE id = ?",
+        (p["guest_bot_id"], p["chat_id"]),
+    )
+
+
+@on("guest_removed")
+def _apply_guest_removed(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "UPDATE chats SET guest_bot_id = NULL WHERE id = ?",
+        (p["chat_id"],),
+    )
+
+
 @on("container_created")
 def _apply_container_created(conn: Connection, e: Event) -> None:
     p = e.payload

chat/templates/_drawer.html

@@ -43,6 +43,101 @@
     {% endfor %}
   </section>
 
+  {% if guest_bot %}
+  <section class="drawer-section">
+    <h3>Guest</h3>
+    <p><strong>{{ guest_bot.name }}</strong></p>
+    {% if guest_activity %}
+      <p>{{ guest_activity.posture or "—" }} / {{ (guest_activity.action or {}).verb or "—" }}</p>
+      {% if guest_activity.attention %}<p class="muted">attention: {{ guest_activity.attention }}</p>{% endif %}
+      {% if guest_activity.holding %}<p class="muted">holding: {{ guest_activity.holding|join(", ") }}</p>{% endif %}
+    {% else %}
+      <p class="muted">No activity recorded.</p>
+    {% endif %}
+
+    {% if edge_h2g %}
+      <div class="edge-row">
+        <strong>{{ host_bot.name }} &rarr; {{ guest_bot.name }}</strong>
+        <p>Affinity: {{ edge_h2g.affinity }}/100 &middot; Trust: {{ edge_h2g.trust }}/100</p>
+        {% if edge_h2g.knowledge %}
+          <details><summary>Knowledge ({{ edge_h2g.knowledge|length }})</summary>
+            <ul>{% for fact in edge_h2g.knowledge %}<li>{{ fact }}</li>{% endfor %}</ul>
+          </details>
+        {% endif %}
+      </div>
+    {% endif %}
+    {% if edge_g2h %}
+      <div class="edge-row">
+        <strong>{{ guest_bot.name }} &rarr; {{ host_bot.name }}</strong>
+        <p>Affinity: {{ edge_g2h.affinity }}/100 &middot; Trust: {{ edge_g2h.trust }}/100</p>
+        {% if edge_g2h.knowledge %}
+          <details><summary>Knowledge ({{ edge_g2h.knowledge|length }})</summary>
+            <ul>{% for fact in edge_g2h.knowledge %}<li>{{ fact }}</li>{% endfor %}</ul>
+          </details>
+        {% endif %}
+      </div>
+    {% endif %}
+    {% if edge_y2g %}
+      <div class="edge-row">
+        <strong>you &rarr; {{ guest_bot.name }}</strong>
+        <p>Affinity: {{ edge_y2g.affinity }}/100 &middot; Trust: {{ edge_y2g.trust }}/100</p>
+      </div>
+    {% endif %}
+    {% if edge_g2y %}
+      <div class="edge-row">
+        <strong>{{ guest_bot.name }} &rarr; you</strong>
+        <p>Affinity: {{ edge_g2y.affinity }}/100 &middot; Trust: {{ edge_g2y.trust }}/100</p>
+      </div>
+    {% endif %}
+
+    <form class="inline-edit"
+          hx-post="/chats/{{ chat.id }}/drawer/guest/remove"
+          hx-target="#drawer" hx-swap="innerHTML">
+      <button type="submit">Remove guest</button>
+    </form>
+  </section>
+  {% else %}
+  <section class="drawer-section">
+    <h3>Add guest</h3>
+    {% if available_guests %}
+      <form class="inline-edit"
+            hx-post="/chats/{{ chat.id }}/drawer/guest/add"
+            hx-target="#drawer" hx-swap="innerHTML">
+        <label>
+          Bot:
+          <select name="guest_bot_id" required>
+            {% for b in available_guests %}
+              <option value="{{ b.id }}">{{ b.name }}</option>
+            {% endfor %}
+          </select>
+        </label>
+        <label>
+          Have they met before? Describe how (leave blank if not):
+          <textarea name="relationship_prose" rows="3"
+                    placeholder="e.g. Old college friends who studied physics together."></textarea>
+        </label>
+        <button type="submit">Add guest</button>
+      </form>
+    {% else %}
+      <p class="muted">No other bots authored yet.</p>
+    {% endif %}
+  </section>
+  {% endif %}
+
+  {% if group_node %}
+  <section class="drawer-section">
+    <h3>Group</h3>
+    {% if group_node.summary %}
+      <p>{{ group_node.summary }}</p>
+    {% else %}
+      <p class="muted">No group summary yet.</p>
+    {% endif %}
+    {% if group_node.dynamic %}
+      <p class="muted">Dynamic: {{ group_node.dynamic }}</p>
+    {% endif %}
+  </section>
+  {% endif %}
+
   <section class="drawer-section">
     <h3>Edges</h3>
     {% if edge_b2y %}

chat/web/drawer.py

@@ -32,9 +32,11 @@ from fastapi.responses import HTMLResponse
 from fastapi.templating import Jinja2Templates
 
 from chat.eventlog.log import append_and_apply
+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
+from chat.state.entities import get_bot, get_you, list_bots
+from chat.state.group_node import get_group_node
 from chat.state.memory import get_pinned
 from chat.state.world import active_scene, get_activity, get_chat, get_container
 from chat.web.bots import get_conn
@@ -78,6 +80,32 @@ async def drawer(chat_id: str, request: Request, conn=Depends(get_conn)):
     edge_b2y = get_edge(conn, chat["host_bot_id"], "you")
     edge_y2b = get_edge(conn, "you", chat["host_bot_id"])
 
+    # T42: guest + group context. Empty defaults keep the template happy
+    # when no guest is present (the relevant sections render conditionally).
+    guest_bot = None
+    guest_activity = None
+    edge_h2g = None
+    edge_g2h = None
+    edge_y2g = None
+    edge_g2y = None
+    available_guests: list[dict] = []
+    group_node = None
+    if chat.get("guest_bot_id"):
+        guest_bot_id = chat["guest_bot_id"]
+        guest_bot = get_bot(conn, guest_bot_id)
+        guest_activity = get_activity(conn, guest_bot_id)
+        edge_h2g = get_edge(conn, chat["host_bot_id"], guest_bot_id)
+        edge_g2h = get_edge(conn, guest_bot_id, chat["host_bot_id"])
+        edge_y2g = get_edge(conn, "you", guest_bot_id)
+        edge_g2y = get_edge(conn, guest_bot_id, "you")
+    else:
+        # Candidates for the "Add guest" dropdown — every authored bot
+        # except the host (and "you", which is implicit, never a bot row).
+        available_guests = [
+            b for b in list_bots(conn) if b["id"] != chat["host_bot_id"]
+        ]
+    group_node = get_group_node(conn, chat_id)
+
     # Recent memories from host's POV (witness_host = 1), most recent first.
     # Raw query keeps this read self-contained — no projector helper exposes
     # "latest N for an owner" yet and the drawer is the only consumer.
@@ -117,6 +145,14 @@ async def drawer(chat_id: str, request: Request, conn=Depends(get_conn)):
             "bot_activity": bot_activity,
             "edge_b2y": edge_b2y,
             "edge_y2b": edge_y2b,
+            "guest_bot": guest_bot,
+            "guest_activity": guest_activity,
+            "edge_h2g": edge_h2g,
+            "edge_g2h": edge_g2h,
+            "edge_y2g": edge_y2g,
+            "edge_g2y": edge_g2y,
+            "available_guests": available_guests,
+            "group_node": group_node,
             "recent_memories": recent_memories,
             "pinned": pinned,
             "pin_cap": PIN_CAP,
@@ -304,3 +340,187 @@ async def toggle_memory_pin(
         },
     )
     return await drawer(chat_id, request, conn)
+
+
+# --- T42 guest add/remove -------------------------------------------------
+#
+# Adding a guest fans out into up to four events: a ``guest_added`` to flip
+# ``chats.guest_bot_id``, two ``edge_update`` events seeded from the
+# user-supplied prose (skipped when the prose is empty / the seed comes back
+# default), and a ``group_node_initialized`` if no row exists yet — three
+# entities now share the chat so the §8.4 group node becomes meaningful.
+#
+# Removing a guest first emits ``scene_closed`` for the active scene (so any
+# host -> you scene closes cleanly with the guest still in scope) before
+# clearing the guest_bot_id; per spec the next user message implicitly opens
+# a fresh you+host scene via Phase 1's mid-chat reset behavior.
+
+
+def _seed_is_default(seed) -> bool:
+    """Treat a seed as a no-op when both summaries are empty AND both
+    delta pairs are zero AND both fact lists are empty.
+    """
+    return (
+        not seed.a_to_b_summary
+        and not seed.b_to_a_summary
+        and seed.a_to_b_affinity_delta == 0
+        and seed.a_to_b_trust_delta == 0
+        and seed.b_to_a_affinity_delta == 0
+        and seed.b_to_a_trust_delta == 0
+        and not seed.a_to_b_knowledge_facts
+        and not seed.b_to_a_knowledge_facts
+    )
+
+
+@router.post(
+    "/chats/{chat_id}/drawer/guest/add",
+    response_class=HTMLResponse,
+)
+async def add_guest(
+    chat_id: str,
+    request: Request,
+    guest_bot_id: str = Form(...),
+    relationship_prose: str = Form(""),
+    conn=Depends(get_conn),
+    client=Depends(get_llm_client),
+):
+    chat = get_chat(conn, chat_id)
+    if chat is None:
+        raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
+
+    if chat.get("guest_bot_id") is not None:
+        raise HTTPException(
+            status_code=400,
+            detail="a guest is already present in this chat",
+        )
+
+    if guest_bot_id == chat["host_bot_id"]:
+        raise HTTPException(
+            status_code=400, detail="guest must differ from host"
+        )
+
+    guest_bot = get_bot(conn, guest_bot_id)
+    if guest_bot is None:
+        raise HTTPException(
+            status_code=404, detail=f"guest bot not found: {guest_bot_id}"
+        )
+
+    host_bot = get_bot(conn, chat["host_bot_id"])
+    if host_bot is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"host bot not found: {chat['host_bot_id']}",
+        )
+
+    settings = request.app.state.settings
+    seed = await seed_inter_bot_edges(
+        client,
+        classifier_model=settings.classifier_model,
+        bot_a_id=chat["host_bot_id"],
+        bot_a_name=host_bot["name"],
+        bot_b_id=guest_bot_id,
+        bot_b_name=guest_bot["name"],
+        relationship_prose=relationship_prose,
+        timeout_s=settings.classifier_timeout_s,
+    )
+
+    append_and_apply(
+        conn,
+        kind="guest_added",
+        payload={"chat_id": chat_id, "guest_bot_id": guest_bot_id},
+    )
+
+    # Emit edge_update only when the seed carries content. Empty prose
+    # short-circuits inside ``seed_inter_bot_edges`` to a default seed,
+    # so this skips the two extra log entries on the no-prose path.
+    # NOTE: ``_apply_edge_update`` does not accept a ``summary`` field —
+    # per-direction summary is set via the per-pov scene-close path
+    # (T27), not direct edge_update. We therefore drop seed.*_summary
+    # here; the deltas + knowledge_facts are what materializes.
+    if not _seed_is_default(seed):
+        append_and_apply(
+            conn,
+            kind="edge_update",
+            payload={
+                "source_id": chat["host_bot_id"],
+                "target_id": guest_bot_id,
+                "chat_id": chat_id,
+                "affinity_delta": seed.a_to_b_affinity_delta,
+                "trust_delta": seed.a_to_b_trust_delta,
+                "knowledge_facts": seed.a_to_b_knowledge_facts,
+                "last_interaction_at": chat.get("time"),
+                "last_interaction_chat_id": chat_id,
+            },
+        )
+        append_and_apply(
+            conn,
+            kind="edge_update",
+            payload={
+                "source_id": guest_bot_id,
+                "target_id": chat["host_bot_id"],
+                "chat_id": chat_id,
+                "affinity_delta": seed.b_to_a_affinity_delta,
+                "trust_delta": seed.b_to_a_trust_delta,
+                "knowledge_facts": seed.b_to_a_knowledge_facts,
+                "last_interaction_at": chat.get("time"),
+                "last_interaction_chat_id": chat_id,
+            },
+        )
+
+    # Three entities now share the chat (you, host, guest) — initialize
+    # the group node row if Wave 1's reader doesn't see one yet.
+    if get_group_node(conn, chat_id) is None:
+        append_and_apply(
+            conn,
+            kind="group_node_initialized",
+            payload={
+                "chat_id": chat_id,
+                "members": ["you", chat["host_bot_id"], guest_bot_id],
+                "summary": "",
+                "dynamic": "",
+                "threads": [],
+            },
+        )
+
+    return await drawer(chat_id, request, conn)
+
+
+@router.post(
+    "/chats/{chat_id}/drawer/guest/remove",
+    response_class=HTMLResponse,
+)
+async def remove_guest(
+    chat_id: str,
+    request: Request,
+    conn=Depends(get_conn),
+):
+    chat = get_chat(conn, chat_id)
+    if chat is None:
+        raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
+
+    if chat.get("guest_bot_id") is None:
+        raise HTTPException(
+            status_code=400, detail="no guest present in this chat"
+        )
+
+    # Close the active scene (if any) before flipping guest_bot_id so
+    # the scene record carries the guest as a participant.
+    scene = active_scene(conn, chat_id)
+    if scene is not None:
+        append_and_apply(
+            conn,
+            kind="scene_closed",
+            payload={
+                "scene_id": scene["id"],
+                "ended_at": chat.get("time"),
+                "significance": 0,
+            },
+        )
+
+    append_and_apply(
+        conn,
+        kind="guest_removed",
+        payload={"chat_id": chat_id},
+    )
+
+    return await drawer(chat_id, request, conn)

tests/test_drawer_guest.py

@@ -0,0 +1,322 @@
+"""T42: drawer guest add/remove + render.
+
+The drawer grows a "Guest" section (when a guest bot is present in the
+chat), a "Group" section sourced from the ``group_node`` row, an
+"Add guest" form (visible while no guest is present), and a "Remove
+guest" button (visible while one is). The two new POST endpoints emit
+``guest_added`` / ``guest_removed`` events plus ancillary updates:
+
+* ``POST /chats/{chat_id}/drawer/guest/add`` runs the relationship-seed
+  classifier (T38) over the user-supplied prose and emits an
+  ``edge_update`` per direction when the seed comes back non-default.
+  It also seeds a ``group_node_initialized`` row when none exists yet.
+* ``POST /chats/{chat_id}/drawer/guest/remove`` first emits
+  ``scene_closed`` for the active scene so the host -> you scene closes
+  cleanly before the guest leaves.
+"""
+
+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_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) and, by default, an open scene so the
+    ``guest_removed`` flow has something to close.
+    """
+    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)
+    )
+
+
+def test_drawer_no_guest_omits_guest_section(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
+    # No guest-section header; the "Add guest" form should be visible instead.
+    assert "<h3>Guest</h3>" not in body
+    assert "Add guest" in body
+
+
+def test_drawer_add_guest_seeds_edges_and_group_node(client, tmp_path):
+    _seed_chat(tmp_path / "test.db")
+    canned = json.dumps(
+        {
+            "a_to_b_summary": "old college friend",
+            "a_to_b_knowledge_facts": ["studied physics together"],
+            "a_to_b_affinity_delta": 4,
+            "a_to_b_trust_delta": -1,
+            "b_to_a_summary": "former roommate",
+            "b_to_a_knowledge_facts": ["lived together junior year"],
+            "b_to_a_affinity_delta": 3,
+            "b_to_a_trust_delta": 0,
+        }
+    )
+    _override_llm([canned])
+    try:
+        response = client.post(
+            "/chats/chat_bot_a/drawer/guest/add",
+            data={
+                "guest_bot_id": "bot_b",
+                "relationship_prose": (
+                    "Alice and Bob met in college and studied physics together."
+                ),
+            },
+        )
+        assert response.status_code == 200
+    finally:
+        app.dependency_overrides.clear()
+
+    with open_db(tmp_path / "test.db") as conn:
+        from chat.state.edges import get_edge
+        from chat.state.group_node import get_group_node
+        from chat.state.world import get_chat
+
+        chat = get_chat(conn, "chat_bot_a")
+        assert chat["guest_bot_id"] == "bot_b"
+
+        edge_a_to_b = get_edge(conn, "bot_a", "bot_b")
+        edge_b_to_a = get_edge(conn, "bot_b", "bot_a")
+        # Seed deltas applied around the 50/50 default.
+        assert edge_a_to_b["affinity"] == 54
+        assert edge_a_to_b["trust"] == 49
+        assert "studied physics together" in edge_a_to_b["knowledge"]
+        assert edge_b_to_a["affinity"] == 53
+        assert edge_b_to_a["trust"] == 50
+        assert "lived together junior year" in edge_b_to_a["knowledge"]
+
+        group = get_group_node(conn, "chat_bot_a")
+        assert group is not None
+        assert set(group["members"]) == {"you", "bot_a", "bot_b"}
+
+
+def test_drawer_add_guest_empty_prose_skips_edge_update(client, tmp_path):
+    _seed_chat(tmp_path / "test.db")
+    # No canned responses: the seed function short-circuits on empty prose
+    # so no LLM call should happen.
+    _override_llm([])
+    try:
+        response = client.post(
+            "/chats/chat_bot_a/drawer/guest/add",
+            data={"guest_bot_id": "bot_b", "relationship_prose": "   "},
+        )
+        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["guest_bot_id"] == "bot_b"
+
+        # guest_added fires but no edge_update events between bot_a and bot_b.
+        added = conn.execute(
+            "SELECT COUNT(*) FROM event_log WHERE kind = 'guest_added'"
+        ).fetchone()[0]
+        assert added == 1
+
+        edge_updates = conn.execute(
+            "SELECT payload_json FROM event_log WHERE kind = 'edge_update'"
+        ).fetchall()
+        for (payload_json,) in edge_updates:
+            payload = json.loads(payload_json)
+            pair = {payload.get("source_id"), payload.get("target_id")}
+            assert pair != {"bot_a", "bot_b"}, (
+                "no edge_update should be emitted between host and guest "
+                "when prose is empty"
+            )
+
+
+def test_drawer_add_guest_when_already_present_returns_400(client, tmp_path):
+    _seed_chat(tmp_path / "test.db")
+    # Pre-attach a guest directly via append_and_apply so we don't replay
+    # the prior chat_created (which would violate UNIQUE on chats.id).
+    from chat.eventlog.log import append_and_apply
+
+    with open_db(tmp_path / "test.db") as conn:
+        append_and_apply(
+            conn,
+            kind="bot_authored",
+            payload=_bot_payload("bot_c", "BotC"),
+        )
+        append_and_apply(
+            conn,
+            kind="guest_added",
+            payload={"chat_id": "chat_bot_a", "guest_bot_id": "bot_b"},
+        )
+
+    _override_llm([])
+    try:
+        response = client.post(
+            "/chats/chat_bot_a/drawer/guest/add",
+            data={"guest_bot_id": "bot_c", "relationship_prose": ""},
+        )
+        assert response.status_code == 400
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_drawer_remove_guest_clears_and_closes_scene(client, tmp_path):
+    _seed_chat(tmp_path / "test.db")
+    from chat.eventlog.log import append_and_apply
+
+    with open_db(tmp_path / "test.db") as conn:
+        append_and_apply(
+            conn,
+            kind="guest_added",
+            payload={"chat_id": "chat_bot_a", "guest_bot_id": "bot_b"},
+        )
+
+    response = client.post("/chats/chat_bot_a/drawer/guest/remove")
+    assert response.status_code == 200
+
+    with open_db(tmp_path / "test.db") as conn:
+        from chat.state.world import active_scene, get_chat
+
+        chat = get_chat(conn, "chat_bot_a")
+        assert chat["guest_bot_id"] is None
+        assert active_scene(conn, "chat_bot_a") is None
+
+        kinds = [
+            row[0]
+            for row in conn.execute(
+                "SELECT kind FROM event_log ORDER BY id"
+            ).fetchall()
+        ]
+        # scene_closed must precede guest_removed in the log.
+        assert "scene_closed" in kinds
+        assert "guest_removed" in kinds
+        assert kinds.index("scene_closed") < kinds.index("guest_removed")
+
+
+def test_drawer_with_guest_renders_guest_and_group_sections(client, tmp_path):
+    _seed_chat(tmp_path / "test.db")
+    from chat.eventlog.log import append_and_apply
+
+    with open_db(tmp_path / "test.db") as conn:
+        append_and_apply(
+            conn,
+            kind="guest_added",
+            payload={"chat_id": "chat_bot_a", "guest_bot_id": "bot_b"},
+        )
+        # Activity for the guest so the section has content to render.
+        append_and_apply(
+            conn,
+            kind="activity_change",
+            payload={
+                "entity_id": "bot_b",
+                "posture": "leaning",
+                "action": {"verb": "smirking"},
+                "attention": "BotA",
+            },
+        )
+        # Edges in all four directions involving the guest.
+        for src, tgt in (("bot_a", "bot_b"), ("bot_b", "bot_a"), ("you", "bot_b"), ("bot_b", "you")):
+            append_and_apply(
+                conn,
+                kind="edge_update",
+                payload={
+                    "source_id": src,
+                    "target_id": tgt,
+                    "chat_id": "chat_bot_a",
+                    "affinity_delta": 1,
+                },
+            )
+        append_and_apply(
+            conn,
+            kind="group_node_initialized",
+            payload={
+                "chat_id": "chat_bot_a",
+                "members": ["you", "bot_a", "bot_b"],
+                "summary": "Three friends catching up over drinks.",
+                "dynamic": "warm and conspiratorial",
+            },
+        )
+
+    response = client.get("/chats/chat_bot_a/drawer")
+    assert response.status_code == 200
+    body = response.text
+    assert "<h3>Guest</h3>" in body
+    assert "BotB" in body
+    assert "smirking" in body
+    assert "<h3>Group</h3>" in body
+    assert "Three friends catching up over drinks." in body
+    assert "warm and conspiratorial" in body
+    # "Remove guest" button is visible when a guest is present.
+    assert "Remove guest" in body

tests/test_group_node.py

@@ -0,0 +1,101 @@
+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.entities  # registers handlers
+import chat.state.world  # registers handlers
+import chat.state.group_node  # registers handlers
+from chat.state.group_node import get_group_node
+
+
+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_group_node_initialized_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="bot_authored", payload=_bot_payload("bot_b", "BotB"))
+        append_event(conn, kind="chat_created", payload=_chat_payload())
+        append_event(
+            conn,
+            kind="group_node_initialized",
+            payload={
+                "chat_id": "chat_bot_a",
+                "members": ["you", "bot_a", "bot_b"],
+            },
+        )
+        project(conn)
+
+        gn = get_group_node(conn, "chat_bot_a")
+        assert gn is not None
+        assert gn["chat_id"] == "chat_bot_a"
+        assert gn["members"] == ["you", "bot_a", "bot_b"]
+        assert gn["summary"] == ""
+        assert gn["dynamic"] == ""
+        assert gn["threads"] == []
+
+
+def test_group_node_updated_changes_summary_and_dynamic(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="bot_authored", payload=_bot_payload("bot_b", "BotB"))
+        append_event(conn, kind="chat_created", payload=_chat_payload())
+        append_event(
+            conn,
+            kind="group_node_initialized",
+            payload={
+                "chat_id": "chat_bot_a",
+                "members": ["you", "bot_a", "bot_b"],
+            },
+        )
+        append_event(
+            conn,
+            kind="group_node_updated",
+            payload={
+                "chat_id": "chat_bot_a",
+                "summary": "Three coworkers chatting about the project.",
+                "dynamic": "Tense but cordial.",
+            },
+        )
+        project(conn)
+
+        gn = get_group_node(conn, "chat_bot_a")
+        assert gn is not None
+        assert gn["summary"] == "Three coworkers chatting about the project."
+        assert gn["dynamic"] == "Tense but cordial."
+        # Members preserved across update
+        assert gn["members"] == ["you", "bot_a", "bot_b"]
+
+
+def test_get_group_node_returns_none_for_missing_chat(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        assert get_group_node(conn, "chat_missing") is None

tests/test_guest_events.py

@@ -0,0 +1,96 @@
+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.entities  # registers bot_authored handler
+import chat.state.world  # registers chat_created / guest_added / guest_removed
+from chat.state.world import get_chat
+
+
+def _bot_payload(bot_id: str, name: str) -> dict:
+    return {
+        "id": bot_id,
+        "name": name,
+        "persona": "...",
+        "voice_samples": ["sample"],
+        "traits": ["shy"],
+        "backstory": "...",
+        "initial_relationship_to_you": "coworker",
+        "kickoff_prose": "you stay late",
+    }
+
+
+def _chat_payload(**overrides) -> dict:
+    payload = {
+        "id": "chat_bot_a",
+        "host_bot_id": "bot_a",
+        "initial_time": "2026-04-26T20:00:00+00:00",
+        "narrative_anchor": "Day 1 evening",
+        "weather": "clear",
+    }
+    payload.update(overrides)
+    return payload
+
+
+def test_guest_added_sets_guest_bot_id(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="bot_authored", payload=_bot_payload("bot_b", "BotB"))
+        append_event(conn, kind="chat_created", payload=_chat_payload())
+        append_event(conn, kind="guest_added", payload={
+            "chat_id": "chat_bot_a",
+            "guest_bot_id": "bot_b",
+        })
+        project(conn)
+
+        chat = get_chat(conn, "chat_bot_a")
+        assert chat is not None
+        assert chat["guest_bot_id"] == "bot_b"
+
+
+def test_guest_removed_clears_guest_bot_id(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="bot_authored", payload=_bot_payload("bot_b", "BotB"))
+        append_event(conn, kind="chat_created", payload=_chat_payload())
+        append_event(conn, kind="guest_added", payload={
+            "chat_id": "chat_bot_a",
+            "guest_bot_id": "bot_b",
+        })
+        append_event(conn, kind="guest_removed", payload={
+            "chat_id": "chat_bot_a",
+        })
+        project(conn)
+
+        chat = get_chat(conn, "chat_bot_a")
+        assert chat is not None
+        assert chat["guest_bot_id"] is None
+
+
+def test_guest_added_idempotent_overwrite(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="bot_authored", payload=_bot_payload("bot_b", "BotB"))
+        append_event(conn, kind="bot_authored", payload=_bot_payload("bot_c", "BotC"))
+        append_event(conn, kind="chat_created", payload=_chat_payload())
+        append_event(conn, kind="guest_added", payload={
+            "chat_id": "chat_bot_a",
+            "guest_bot_id": "bot_b",
+        })
+        append_event(conn, kind="guest_added", payload={
+            "chat_id": "chat_bot_a",
+            "guest_bot_id": "bot_c",
+        })
+        project(conn)
+
+        chat = get_chat(conn, "chat_bot_a")
+        assert chat is not None
+        assert chat["guest_bot_id"] == "bot_c"

tests/test_interjection.py

@@ -0,0 +1,89 @@
+"""Tests for the interjection classifier service (T39).
+
+Per Requirements §6.2, when a guest is present and the addressee bot has
+just spoken, the *non-addressee* bot may interject with a brief follow-on
+beat. The classifier wrapped here decides whether that interjection
+should fire. The default bias is strongly toward False — the addressee
+has the floor — so an interjection only fires when the silent witness's
+character would plausibly speak up.
+
+These tests cover:
+
+* The classifier returning ``should_interject=True`` is honored.
+* The classifier returning ``should_interject=False`` is honored.
+* Repeated invalid JSON exhausts the classifier retries and falls back
+  to ``should_interject=False`` with ``reason="fallback"``.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from chat.llm.mock import MockLLMClient
+from chat.services.interjection import (
+    InterjectionDecision,
+    detect_interjection,
+)
+
+
+def _kwargs() -> dict:
+    """Reasonable, non-empty kwargs for ``detect_interjection``."""
+    return dict(
+        classifier_model="x",
+        addressee_name="Alice",
+        addressee_just_said="I think we should leave now.",
+        silent_witness_name="Bob",
+        silent_witness_persona="Skeptical engineer, blunt, protective of the user.",
+        silent_witness_edge_to_addressee={
+            "affinity": 40,
+            "trust": 30,
+            "summary": "old rival; mild distrust",
+        },
+        silent_witness_edge_to_you={
+            "affinity": 70,
+            "trust": 80,
+            "summary": "long-time confidant",
+        },
+        you_just_said="Where do you both think we should go?",
+    )
+
+
+@pytest.mark.asyncio
+async def test_interjection_returns_true_when_classifier_decides_yes():
+    canned = json.dumps({"should_interject": True, "reason": "jealousy"})
+    mock = MockLLMClient(canned=[canned])
+    result = await detect_interjection(mock, **_kwargs())
+    assert isinstance(result, InterjectionDecision)
+    assert result.should_interject is True
+    assert result.reason == "jealousy"
+
+
+@pytest.mark.asyncio
+async def test_interjection_returns_false_when_classifier_decides_no():
+    canned = json.dumps(
+        {"should_interject": False, "reason": "addressee has the floor"}
+    )
+    mock = MockLLMClient(canned=[canned])
+    result = await detect_interjection(mock, **_kwargs())
+    assert isinstance(result, InterjectionDecision)
+    assert result.should_interject is False
+    assert result.reason == "addressee has the floor"
+
+
+@pytest.mark.asyncio
+async def test_interjection_falls_back_to_false_on_classifier_failure():
+    """``classify`` retries 3 times; after all fail it returns the default.
+
+    The default carries ``should_interject=False`` and
+    ``reason="fallback"`` so callers can tell a real "no" from a
+    classifier-degraded "no" if they ever care to.
+    """
+    mock = MockLLMClient(
+        canned=["this is not json", "still not json", "still not json"]
+    )
+    result = await detect_interjection(mock, **_kwargs())
+    assert isinstance(result, InterjectionDecision)
+    assert result.should_interject is False
+    assert result.reason == "fallback"

tests/test_memory_write.py

@@ -22,7 +22,7 @@ from chat.db.migrate import apply_migrations
 from chat.eventlog.log import append_event
 from chat.eventlog.projector import project
 from chat.llm.mock import MockLLMClient
-from chat.services.memory_write import record_turn_memory
+from chat.services.memory_write import record_turn_memory, record_turn_memory_for_present
 import chat.state.entities  # noqa: F401  - register handlers
 import chat.state.memory  # noqa: F401
 import chat.state.world  # noqa: F401
@@ -295,3 +295,152 @@ def test_post_turn_writes_memory_for_host_bot(client, tmp_path):
         assert w_guest == 0
         assert source == "direct"
         assert sig == 1
+
+
+# ---------------------------------------------------------------------------
+# T41: record_turn_memory_for_present — multi-witness helper.
+# ---------------------------------------------------------------------------
+
+
+def _seed_two_bots(db_path: Path) -> None:
+    """Author host + guest bots and create a two-bot chat."""
+    with open_db(db_path) as conn:
+        for bot_id, name in (("bot_a", "BotA"), ("bot_b", "BotB")):
+            append_event(
+                conn,
+                kind="bot_authored",
+                payload={
+                    "id": bot_id,
+                    "name": name,
+                    "persona": "...",
+                    "voice_samples": [],
+                    "traits": [],
+                    "backstory": "",
+                    "initial_relationship_to_you": "",
+                    "kickoff_prose": "",
+                },
+            )
+        append_event(
+            conn,
+            kind="chat_created",
+            payload={
+                "id": "chat_ab",
+                "host_bot_id": "bot_a",
+                "guest_bot_id": "bot_b",
+                "initial_time": "2026-04-26T20:00:00+00:00",
+                "narrative_anchor": "Day 1",
+                "weather": "",
+            },
+        )
+        project(conn)
+
+
+def test_record_for_present_no_guest_writes_single_memory_with_witness_1_1_0(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    _seed_minimal(db)
+    with open_db(db) as conn:
+        result = record_turn_memory_for_present(
+            conn,
+            chat_id="chat_bot_a",
+            host_bot_id="bot_a",
+            guest_bot_id=None,
+            narrative_text="BotA glances out the window.",
+            scene_id=None,
+            chat_clock_at="2026-04-26T20:00:00+00:00",
+        )
+
+        # Returned dict has only the host key.
+        assert set(result.keys()) == {"bot_a"}
+        eid_h, mid_h = result["bot_a"]
+        assert eid_h > 0
+        assert mid_h is not None and mid_h > 0
+
+        rows = conn.execute(
+            "SELECT owner_id, witness_you, witness_host, witness_guest "
+            "FROM memories"
+        ).fetchall()
+        assert len(rows) == 1
+        owner_id, w_you, w_host, w_guest = rows[0]
+        assert owner_id == "bot_a"
+        assert w_you == 1
+        assert w_host == 1
+        assert w_guest == 0
+
+        # Exactly one memory_written event was appended.
+        cur = conn.execute(
+            "SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
+        )
+        assert cur.fetchone()[0] == 1
+
+
+def test_record_for_present_with_guest_writes_two_memories_with_witness_1_1_1(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    _seed_two_bots(db)
+    with open_db(db) as conn:
+        result = record_turn_memory_for_present(
+            conn,
+            chat_id="chat_ab",
+            host_bot_id="bot_a",
+            guest_bot_id="bot_b",
+            narrative_text="BotA and BotB share a glance.",
+            scene_id=None,
+            chat_clock_at="2026-04-26T20:00:00+00:00",
+        )
+
+        # Returned dict has both keys.
+        assert set(result.keys()) == {"bot_a", "bot_b"}
+        eid_h, mid_h = result["bot_a"]
+        eid_g, mid_g = result["bot_b"]
+        assert eid_h > 0 and eid_g > 0
+        assert mid_h is not None and mid_h > 0
+        assert mid_g is not None and mid_g > 0
+        # Distinct event ids and memory ids.
+        assert eid_h != eid_g
+        assert mid_h != mid_g
+
+        rows = conn.execute(
+            "SELECT owner_id, witness_you, witness_host, witness_guest "
+            "FROM memories ORDER BY owner_id"
+        ).fetchall()
+        assert len(rows) == 2
+        owners = {r[0] for r in rows}
+        assert owners == {"bot_a", "bot_b"}
+        # All rows should have witness mask [1, 1, 1].
+        for _owner, w_you, w_host, w_guest in rows:
+            assert w_you == 1
+            assert w_host == 1
+            assert w_guest == 1
+
+        # Two memory_written events were appended.
+        cur = conn.execute(
+            "SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
+        )
+        assert cur.fetchone()[0] == 2
+
+
+def test_record_for_present_dict_keys_match(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    _seed_two_bots(db)
+    with open_db(db) as conn:
+        # No guest: keys == {host_bot_id}.
+        result_no_guest = record_turn_memory_for_present(
+            conn,
+            chat_id="chat_ab",
+            host_bot_id="bot_a",
+            guest_bot_id=None,
+            narrative_text="Just BotA's POV.",
+        )
+        assert set(result_no_guest.keys()) == {"bot_a"}
+
+        # With guest: keys == {host_bot_id, guest_bot_id}.
+        result_with_guest = record_turn_memory_for_present(
+            conn,
+            chat_id="chat_ab",
+            host_bot_id="bot_a",
+            guest_bot_id="bot_b",
+            narrative_text="Both bots witness this.",
+        )
+        assert set(result_with_guest.keys()) == {"bot_a", "bot_b"}

tests/test_multi_state_update.py

@@ -0,0 +1,147 @@
+"""Multi-entity state-update coordinator (T40).
+
+Wraps the single-pair :func:`compute_state_update` to run state updates
+for ALL directed pairs of present entities. With 3 present entities
+(you, host, guest) that's 6 directed pairs; with 2 (you, host) it's 2.
+
+Calls run sequentially to respect Featherless's 2-connection cap.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from chat.llm.mock import MockLLMClient
+from chat.services.multi_state_update import compute_state_updates_for_present
+from chat.services.state_update import StateUpdate
+
+
+def _canned_update(affinity: int, trust: int, facts: list[str] | None = None) -> str:
+    return json.dumps(
+        {
+            "affinity_delta": affinity,
+            "trust_delta": trust,
+            "knowledge_facts": facts or [],
+        }
+    )
+
+
+@pytest.mark.asyncio
+async def test_two_entities_returns_two_updates():
+    """you + bot_a -> 2 directed pairs (you->bot_a, bot_a->you)."""
+    canned = [
+        _canned_update(2, 1, ["likes coffee"]),  # you -> bot_a
+        _canned_update(1, 0, ["greets warmly"]),  # bot_a -> you
+    ]
+    mock = MockLLMClient(canned=canned)
+
+    results = await compute_state_updates_for_present(
+        mock,
+        classifier_model="x",
+        present_ids=["you", "bot_a"],
+        present_names={"you": "Me", "bot_a": "BotA"},
+        personas={"you": "", "bot_a": "thoughtful"},
+        prior_edges={
+            ("you", "bot_a"): {"affinity": 50, "trust": 50, "summary": ""},
+            ("bot_a", "you"): {"affinity": 50, "trust": 50, "summary": ""},
+        },
+        recent_dialogue=[
+            {"speaker": "you", "text": "hi"},
+            {"speaker": "BotA", "text": "Hello!"},
+        ],
+    )
+
+    assert len(results) == 2
+    assert results[0][0] == "you"
+    assert results[0][1] == "bot_a"
+    assert isinstance(results[0][2], StateUpdate)
+    assert results[0][2].affinity_delta == 2
+    assert results[0][2].trust_delta == 1
+    assert results[0][2].knowledge_facts == ["likes coffee"]
+
+    assert results[1][0] == "bot_a"
+    assert results[1][1] == "you"
+    assert isinstance(results[1][2], StateUpdate)
+    assert results[1][2].affinity_delta == 1
+    assert results[1][2].trust_delta == 0
+    assert results[1][2].knowledge_facts == ["greets warmly"]
+
+
+@pytest.mark.asyncio
+async def test_three_entities_returns_six_updates():
+    """you + bot_a + bot_b -> 6 directed pairs (no self-pairs)."""
+    canned = [_canned_update(i, 0) for i in range(6)]
+    mock = MockLLMClient(canned=canned)
+
+    results = await compute_state_updates_for_present(
+        mock,
+        classifier_model="x",
+        present_ids=["you", "bot_a", "bot_b"],
+        present_names={"you": "Me", "bot_a": "BotA", "bot_b": "BotB"},
+        personas={"you": "", "bot_a": "thoughtful", "bot_b": "cheerful"},
+        prior_edges={},  # all default to 50/50/""
+        recent_dialogue=[{"speaker": "you", "text": "hello all"}],
+    )
+
+    assert len(results) == 6
+
+    pairs = [(src, tgt) for src, tgt, _ in results]
+    # No self-pairs.
+    assert all(src != tgt for src, tgt in pairs)
+    # All 6 directed combinations present.
+    expected = {
+        ("you", "bot_a"),
+        ("you", "bot_b"),
+        ("bot_a", "you"),
+        ("bot_a", "bot_b"),
+        ("bot_b", "you"),
+        ("bot_b", "bot_a"),
+    }
+    assert set(pairs) == expected
+    # Every entry is a StateUpdate.
+    assert all(isinstance(u, StateUpdate) for _, _, u in results)
+
+
+@pytest.mark.asyncio
+async def test_failure_in_one_pair_does_not_kill_batch():
+    """First pair fails all 3 classify retries -> default; second parses OK."""
+    canned = [
+        # Pair 1 (you -> bot_a): 3 malformed responses -> default StateUpdate.
+        "bad",
+        "still bad",
+        "nope",
+        # Pair 2 (bot_a -> you): valid JSON.
+        _canned_update(3, 2, ["was warm"]),
+    ]
+    mock = MockLLMClient(canned=canned)
+
+    results = await compute_state_updates_for_present(
+        mock,
+        classifier_model="x",
+        present_ids=["you", "bot_a"],
+        present_names={"you": "Me", "bot_a": "BotA"},
+        personas={"you": "", "bot_a": "thoughtful"},
+        prior_edges={
+            ("you", "bot_a"): {"affinity": 60, "trust": 40, "summary": "some prior"},
+            ("bot_a", "you"): {"affinity": 50, "trust": 50, "summary": ""},
+        },
+        recent_dialogue=[{"speaker": "you", "text": "hi"}],
+    )
+
+    assert len(results) == 2
+
+    # First pair: default (zero-delta) StateUpdate.
+    src1, tgt1, update1 = results[0]
+    assert (src1, tgt1) == ("you", "bot_a")
+    assert update1.affinity_delta == 0
+    assert update1.trust_delta == 0
+    assert update1.knowledge_facts == []
+
+    # Second pair: parsed valid JSON.
+    src2, tgt2, update2 = results[1]
+    assert (src2, tgt2) == ("bot_a", "you")
+    assert update2.affinity_delta == 3
+    assert update2.trust_delta == 2
+    assert update2.knowledge_facts == ["was warm"]

tests/test_per_pov_summary.py

@@ -258,3 +258,425 @@ async def test_apply_scene_close_summary_updates_memories_and_edge(tmp_path):
 
         # Knowledge fact appended via edge_update.
         assert any("deadline" in fact for fact in edge["knowledge"])
+
+
+# ---------------------------------------------------------------------------
+# T45: per-POV summaries on close for each present witness.
+# ---------------------------------------------------------------------------
+
+
+def _bot_payload(bot_id: str, name: str, persona: str = "thoughtful") -> dict:
+    return {
+        "id": bot_id,
+        "name": name,
+        "persona": persona,
+        "voice_samples": [],
+        "traits": [],
+        "backstory": "",
+        "initial_relationship_to_you": "",
+        "kickoff_prose": "",
+    }
+
+
+def _seed_single_bot_scene(conn) -> None:
+    """Seed the canonical Phase 1 single-bot scene used by the regression test."""
+    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="memory_written",
+        payload={
+            "owner_id": "bot_a",
+            "chat_id": "chat_bot_a",
+            "scene_id": 1,
+            "pov_summary": "Original raw narrative (host)",
+            "witness_you": 1,
+            "witness_host": 1,
+            "witness_guest": 0,
+            "significance": 1,
+        },
+    )
+    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_two_bot_scene(conn, *, with_group_node: bool = False) -> None:
+    """Seed a host+guest scene with bot_a (host) and bot_b (guest), plus a
+    memory row per bot owner so each per-POV update has something to rewrite,
+    and seeded directed edges from each bot to ``you`` so each edge_summary
+    update has a row to operate on. Optionally seeds the group_node row too.
+    """
+    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": "engineer"},
+    )
+    append_event(
+        conn,
+        kind="chat_created",
+        payload={
+            "id": "chat_bot_a",
+            "host_bot_id": "bot_a",
+            "guest_bot_id": "bot_b",
+            "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", "bot_b"],
+        },
+    )
+    # Seed edges in both bot -> you directions so the edge_summary updates
+    # have rows to target.
+    append_event(
+        conn,
+        kind="edge_update",
+        payload={
+            "source_id": "bot_a",
+            "target_id": "you",
+            "chat_id": "chat_bot_a",
+        },
+    )
+    append_event(
+        conn,
+        kind="edge_update",
+        payload={
+            "source_id": "bot_b",
+            "target_id": "you",
+            "chat_id": "chat_bot_a",
+        },
+    )
+    # One memory per witness, scene 1.
+    append_event(
+        conn,
+        kind="memory_written",
+        payload={
+            "owner_id": "bot_a",
+            "chat_id": "chat_bot_a",
+            "scene_id": 1,
+            "pov_summary": "Original raw narrative (host)",
+            "witness_you": 1,
+            "witness_host": 1,
+            "witness_guest": 1,
+            "significance": 1,
+        },
+    )
+    append_event(
+        conn,
+        kind="memory_written",
+        payload={
+            "owner_id": "bot_b",
+            "chat_id": "chat_bot_a",
+            "scene_id": 1,
+            "pov_summary": "Original raw narrative (guest)",
+            "witness_you": 1,
+            "witness_host": 1,
+            "witness_guest": 1,
+            "significance": 1,
+        },
+    )
+    append_event(
+        conn,
+        kind="user_turn",
+        payload={
+            "chat_id": "chat_bot_a",
+            "prose": "Three of us in the office.",
+            "segments": [],
+        },
+    )
+    append_event(
+        conn,
+        kind="assistant_turn",
+        payload={
+            "chat_id": "chat_bot_a",
+            "speaker_id": "bot_a",
+            "text": "Glad you're both here.",
+            "truncated": False,
+            "user_turn_id": 1,
+        },
+    )
+    if with_group_node:
+        append_event(
+            conn,
+            kind="group_node_initialized",
+            payload={
+                "chat_id": "chat_bot_a",
+                "members": ["you", "bot_a", "bot_b"],
+                "summary": "",
+                "dynamic": "",
+                "threads": [],
+            },
+        )
+
+
+@pytest.mark.asyncio
+async def test_close_with_no_guest_matches_phase1(tmp_path):
+    """Regression: when guest_bot_id is None, the close summary path runs
+    summarize_scene exactly once and rewrites the host's memory + host->you
+    edge in place — same as Phase 1 behavior."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    canned = json.dumps(
+        {
+            "summary": "BotA helped you talk through the deadline anxiety.",
+            "knowledge_facts": ["Deadline next Friday."],
+            "relationship_summary": "BotA leaned in supportively.",
+        }
+    )
+    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])
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=1,
+            host_bot_id="bot_a",
+        )
+
+        # Exactly one classifier call -> exactly one canned entry consumed,
+        # leaving the second untouched.
+        assert len(client._canned) == 1
+
+        # Host memory rewritten with the per-POV summary content.
+        new_pov = conn.execute(
+            "SELECT pov_summary FROM memories "
+            "WHERE owner_id = 'bot_a' AND scene_id = 1"
+        ).fetchone()[0]
+        assert "BotA helped" in new_pov
+
+        # host->you edge summary rewritten with the relationship_summary.
+        from chat.state.edges import get_edge
+
+        edge = get_edge(conn, "bot_a", "you")
+        assert "supportively" in edge["summary"]
+
+
+@pytest.mark.asyncio
+async def test_close_with_guest_calls_summarize_twice(tmp_path):
+    """When a guest is present, summarize_scene runs once per witness
+    (host + guest) and each bot's memory rewrite uses its own POV summary."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    host_canned = json.dumps(
+        {
+            "summary": "BotA noticed BotB warming up to you.",
+            "knowledge_facts": ["You sketched on the whiteboard."],
+            "relationship_summary": "BotA felt steady around you.",
+        }
+    )
+    guest_canned = json.dumps(
+        {
+            "summary": "BotB found the office quieter than expected.",
+            "knowledge_facts": ["You prefer black coffee."],
+            "relationship_summary": "BotB warmed up to you a little.",
+        }
+    )
+    with open_db(db) as conn:
+        _seed_two_bot_scene(conn)
+        project(conn)
+
+        client = MockLLMClient(canned=[host_canned, guest_canned])
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=1,
+            host_bot_id="bot_a",
+        )
+
+        # Both canned entries consumed -> classifier ran twice.
+        assert client._canned == []
+
+        # Host memory carries the host's per-POV summary; guest memory
+        # carries the guest's.
+        host_pov = conn.execute(
+            "SELECT pov_summary FROM memories "
+            "WHERE owner_id = 'bot_a' AND scene_id = 1"
+        ).fetchone()[0]
+        guest_pov = conn.execute(
+            "SELECT pov_summary FROM memories "
+            "WHERE owner_id = 'bot_b' AND scene_id = 1"
+        ).fetchone()[0]
+        assert "BotA noticed" in host_pov
+        assert "BotB found" in guest_pov
+        assert host_pov != guest_pov
+
+
+@pytest.mark.asyncio
+async def test_close_with_guest_updates_both_edges(tmp_path):
+    """Both bot->you edges receive their own relationship_summary on close."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    host_canned = json.dumps(
+        {
+            "summary": "BotA noticed BotB warming up.",
+            "knowledge_facts": [],
+            "relationship_summary": "BotA felt steady around you.",
+        }
+    )
+    guest_canned = json.dumps(
+        {
+            "summary": "BotB warmed to the office.",
+            "knowledge_facts": [],
+            "relationship_summary": "BotB warmed up to you a little.",
+        }
+    )
+    with open_db(db) as conn:
+        _seed_two_bot_scene(conn)
+        project(conn)
+
+        client = MockLLMClient(canned=[host_canned, guest_canned])
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=1,
+            host_bot_id="bot_a",
+        )
+
+        from chat.state.edges import get_edge
+
+        edge_h2y = get_edge(conn, "bot_a", "you")
+        edge_g2y = get_edge(conn, "bot_b", "you")
+        assert "steady" in edge_h2y["summary"]
+        assert "warmed up" in edge_g2y["summary"]
+        # Per-POV; the two edges did not collapse onto the same text.
+        assert edge_h2y["summary"] != edge_g2y["summary"]
+
+
+@pytest.mark.asyncio
+async def test_close_with_group_node_updates_group_summary(tmp_path):
+    """When a group_node row exists, scene close emits group_node_updated
+    with a non-empty summary that mentions both bots' names (v2 naive
+    concat of per-POV summaries)."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    import chat.state.group_node  # noqa: F401  -- register handlers
+
+    host_canned = json.dumps(
+        {
+            "summary": "BotA appreciated the calm.",
+            "knowledge_facts": [],
+            "relationship_summary": "BotA felt steady.",
+        }
+    )
+    guest_canned = json.dumps(
+        {
+            "summary": "BotB found the room friendly.",
+            "knowledge_facts": [],
+            "relationship_summary": "BotB warmed up.",
+        }
+    )
+    with open_db(db) as conn:
+        _seed_two_bot_scene(conn, with_group_node=True)
+        project(conn)
+
+        client = MockLLMClient(canned=[host_canned, guest_canned])
+        await apply_scene_close_summary(
+            conn,
+            client,
+            classifier_model="x",
+            chat_id="chat_bot_a",
+            scene_id=1,
+            host_bot_id="bot_a",
+        )
+
+        from chat.state.group_node import get_group_node
+
+        gn = get_group_node(conn, "chat_bot_a")
+        assert gn is not None
+        assert gn["summary"]  # non-empty
+        # Naive concat surfaces both bot names in the group summary.
+        assert "BotA" in gn["summary"]
+        assert "BotB" in gn["summary"]
+        # Phase 2 v2 keeps dynamic empty (Phase 3 polishes).
+        assert gn["dynamic"] == ""

tests/test_prompt.py

@@ -253,3 +253,244 @@ def test_must_exceeds_budget_hard_raises_value_error(tmp_path):
                 budget_soft=5,
                 budget_hard=10,
             )
+
+
+# ---------------------------------------------------------------------------
+# Task 43: multi-entity prompt assembly (guest_id support)
+# ---------------------------------------------------------------------------
+
+
+def _seed_with_guest(conn) -> None:
+    """Seed a 3-entity scene: you (Sam) + host (Aria, bot_a) + guest (Iris, bot_b).
+
+    Group node row is initialized with summary + dynamic, edges in all
+    relevant directions are seeded, and activities are recorded for all
+    three entities.
+    """
+    append_event(conn, kind="bot_authored", payload={
+        "id": "bot_a",
+        "name": "Aria",
+        "persona": "reserved coworker who notices things",
+        "voice_samples": ["I — sorry, I didn't mean to.", "Right. Of course."],
+        "traits": ["introverted", "observant"],
+        "backstory": "An archivist who joined the firm last spring.",
+        "initial_relationship_to_you": "coworker; mild crush; never voiced",
+        "kickoff_prose": "you stay late at the office",
+    })
+    append_event(conn, kind="bot_authored", payload={
+        "id": "bot_b",
+        "name": "Iris",
+        "persona": "wry transplant from the Boston office",
+        "voice_samples": ["Oh, please.", "Don't make me say it twice."],
+        "traits": ["sardonic", "loyal"],
+        "backstory": "Met Aria at a conference two years back.",
+        "initial_relationship_to_you": "stranger; curious",
+        "kickoff_prose": "",
+    })
+    append_event(conn, kind="you_authored", payload={
+        "name": "Sam",
+        "pronouns": "they/them",
+        "persona": "tired analyst",
+    })
+    append_event(conn, kind="chat_created", payload={
+        "id": "chat_bot_a",
+        "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",
+    })
+    append_event(conn, kind="container_created", payload={
+        "chat_id": "chat_bot_a",
+        "name": "office bullpen",
+        "type": "workplace",
+        "properties": {"public": False, "moving": False, "audible_range": "room"},
+    })
+    # Edges: host -> you, guest -> you, host -> guest, guest -> host.
+    append_event(conn, kind="edge_update", payload={
+        "source_id": "bot_a",
+        "target_id": "you",
+        "affinity_delta": 12,
+        "trust_delta": 5,
+        "knowledge_facts": ["they work on the same floor"],
+    })
+    append_event(conn, kind="edge_update", payload={
+        "source_id": "bot_a",
+        "target_id": "bot_b",
+        "affinity_delta": 20,
+        "trust_delta": 15,
+        "knowledge_facts": ["studied physics together"],
+    })
+    append_event(conn, kind="edge_update", payload={
+        "source_id": "bot_b",
+        "target_id": "you",
+        "affinity_delta": 4,
+        "trust_delta": 0,
+        "knowledge_facts": ["Aria's coworker"],
+    })
+    append_event(conn, kind="edge_update", payload={
+        "source_id": "bot_b",
+        "target_id": "bot_a",
+        "affinity_delta": 18,
+        "trust_delta": 12,
+        "knowledge_facts": ["former roommate"],
+    })
+    # Activity for all three entities — note distinct verbs so we can
+    # check whose activity got dropped under tight budget.
+    append_event(conn, kind="activity_change", payload={
+        "entity_id": "you",
+        "container_id": 1,
+        "posture": "sitting at your desk",
+        "action": {"verb": "finishing emails"},
+        "attention": "the screen",
+        "holding": ["coffee mug"],
+    })
+    append_event(conn, kind="activity_change", payload={
+        "entity_id": "bot_a",
+        "container_id": 1,
+        "posture": "sitting at her desk",
+        "action": {"verb": "pretending to work"},
+        "attention": "you, in glances",
+    })
+    append_event(conn, kind="activity_change", payload={
+        "entity_id": "bot_b",
+        "container_id": 1,
+        "posture": "leaning against the doorframe",
+        "action": {"verb": "smirking-distinctively"},
+        "attention": "Aria",
+    })
+    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", "bot_b"],
+    })
+    append_event(conn, kind="group_node_initialized", payload={
+        "chat_id": "chat_bot_a",
+        "members": ["you", "bot_a", "bot_b"],
+        "summary": "Three coworkers catching up after hours UNIQUE-GROUP-SUMMARY.",
+        "dynamic": "warm-but-prickly UNIQUE-GROUP-DYNAMIC",
+    })
+    project(conn)
+
+
+def test_assemble_with_no_guest_matches_phase1(tmp_path):
+    """Regression: 2-entity scenario without guest_id behaves exactly as Phase 1."""
+    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
+    # Phase 1 must blocks present.
+    assert "Aria" in body
+    assert "PERSONA" in body
+    assert "Sam" in body
+    assert "ACTIVITIES" in body
+    assert "62/100" in body  # speaker → addressee edge intact
+    # No guest content leaks in.
+    assert "Group dynamic" not in body
+    assert "Iris" not in body
+
+
+def test_assemble_with_guest_includes_group_node_summary(tmp_path):
+    """When guest is present (auto-detected via chat.guest_bot_id) and a
+    group_node row exists, its summary + dynamic are rendered."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_with_guest(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 "Group dynamic" in body
+    assert "UNIQUE-GROUP-SUMMARY" in body
+    assert "UNIQUE-GROUP-DYNAMIC" in body
+    # Guest activity also present (SHOULD-tier, fits at default budget).
+    assert "smirking-distinctively" in body
+    # Speaker's other edges include the host -> guest direction.
+    assert "Iris" in body
+
+
+def test_assemble_when_speaker_is_guest_orients_edges_correctly(tmp_path):
+    """When the guest is the speaker, identity is the guest, the
+    addressee edge is guest → you, and other edges include guest → host."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_with_guest(conn)
+        msgs = assemble_narrative_prompt(
+            conn,
+            chat_id="chat_bot_a",
+            speaker_bot_id="bot_b",  # guest as speaker
+            recent_dialogue=[],
+            retrieved_memory_summaries=[],
+        )
+    body = msgs[0].content
+    # Speaker identity is the guest's persona.
+    assert "You are Iris." in body
+    assert "wry transplant from the Boston office" in body
+    # Edge to addressee is guest → you (Sam) with the seeded values
+    # (default 50 + 4 affinity = 54).
+    assert "YOUR EDGE TO Sam" in body
+    assert "54/100" in body
+    # Other edges include guest → host (Aria) with seeded value
+    # (default 50 + 18 = 68).
+    assert "OTHER EDGES" in body
+    assert "Aria" in body
+    assert "68/100" in body
+
+
+def test_assemble_with_tight_budget_drops_guest_activity_first(tmp_path):
+    """Under tight budget MUST blocks survive but SHOULD-tier guest
+    activity is dropped first."""
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    with open_db(db) as conn:
+        _seed_with_guest(conn)
+        # Short dialogue so MUST core (speaker identity + edge + last 4
+        # turns + closing) sits comfortably under the hard budget while
+        # SHOULD-tier additions (guest activity, group node, other edges)
+        # would push over.
+        dialogue = [
+            {"speaker": "you", "text": "line-16 hi there"},
+            {"speaker": "bot_a", "text": "line-17 hey"},
+            {"speaker": "you", "text": "line-18 quiet night"},
+            {"speaker": "bot_a", "text": "line-19 indeed"},
+        ]
+        msgs = assemble_narrative_prompt(
+            conn,
+            chat_id="chat_bot_a",
+            speaker_bot_id="bot_a",
+            recent_dialogue=dialogue,
+            retrieved_memory_summaries=[],
+            # MUST core ~310 tokens; SHOULD additions (guest activity +
+            # group node + other edges) push it well over 380. budget_hard
+            # is set just above MUST core so SHOULD-tier blocks must be
+            # trimmed away.
+            budget_soft=250,
+            budget_hard=340,
+        )
+    body = msgs[0].content
+    # MUST: speaker identity, edge to addressee, last 4 dialogue turns.
+    assert "Aria" in body
+    assert "YOUR EDGE TO Sam" in body
+    for i in range(16, 20):
+        assert f"line-{i:02d}" in body
+    # Guest activity (SHOULD-tier) must be dropped under tight budget.
+    assert "smirking-distinctively" not in body
+    # Token budget honoured.
+    import tiktoken
+    enc = tiktoken.get_encoding("cl100k_base")
+    assert len(enc.encode(body)) <= 340

tests/test_relationship_seed.py

@@ -0,0 +1,109 @@
+"""Tests for the relationship-seed service (T38).
+
+Per Requirements §5.2, when two bots first co-appear in a chat, the user
+is prompted with "Have they met before? If yes, write a short prose
+seed." The prose is parsed via classifier into structured directed-edge
+content for the ``botA -> botB`` and ``botB -> botA`` edges.
+
+These tests cover:
+
+* The happy path: a canned classifier response parses cleanly into a
+  populated :class:`RelationshipSeed` with both directions filled.
+* Empty prose short-circuits before any classifier call (mock has no
+  canned responses; an accidental call would raise ``IndexError``).
+* Whitespace-only prose has the same short-circuit behavior.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from chat.llm.mock import MockLLMClient
+from chat.services.relationship_seed import (
+    RelationshipSeed,
+    seed_inter_bot_edges,
+)
+
+
+@pytest.mark.asyncio
+async def test_seed_parses_canned_prose():
+    canned = json.dumps(
+        {
+            "a_to_b_summary": "old college friend who now distrusts him slightly",
+            "a_to_b_knowledge_facts": [
+                "studied physics together",
+                "lost touch after a falling out",
+            ],
+            "a_to_b_affinity_delta": 2,
+            "a_to_b_trust_delta": -1,
+            "b_to_a_summary": "former roommate; warm memories, mild resentment",
+            "b_to_a_knowledge_facts": ["lived together junior year"],
+            "b_to_a_affinity_delta": 3,
+            "b_to_a_trust_delta": 0,
+        }
+    )
+    mock = MockLLMClient(canned=[canned])
+    result = await seed_inter_bot_edges(
+        mock,
+        classifier_model="x",
+        bot_a_id="bot_a",
+        bot_a_name="Alice",
+        bot_b_id="bot_b",
+        bot_b_name="Bob",
+        relationship_prose=(
+            "Alice and Bob met in college. They studied physics together and "
+            "lived as roommates junior year, but drifted apart after a fight."
+        ),
+    )
+    assert isinstance(result, RelationshipSeed)
+    assert (
+        result.a_to_b_summary
+        == "old college friend who now distrusts him slightly"
+    )
+    assert result.a_to_b_knowledge_facts == [
+        "studied physics together",
+        "lost touch after a falling out",
+    ]
+    assert result.a_to_b_affinity_delta == 2
+    assert result.a_to_b_trust_delta == -1
+    assert (
+        result.b_to_a_summary
+        == "former roommate; warm memories, mild resentment"
+    )
+    assert result.b_to_a_knowledge_facts == ["lived together junior year"]
+    assert result.b_to_a_affinity_delta == 3
+    assert result.b_to_a_trust_delta == 0
+
+
+@pytest.mark.asyncio
+async def test_seed_empty_prose_returns_empty():
+    """Empty prose short-circuits — classifier must not be called."""
+    mock = MockLLMClient(canned=[])
+    result = await seed_inter_bot_edges(
+        mock,
+        classifier_model="x",
+        bot_a_id="bot_a",
+        bot_a_name="Alice",
+        bot_b_id="bot_b",
+        bot_b_name="Bob",
+        relationship_prose="",
+    )
+    assert result == RelationshipSeed()
+
+
+@pytest.mark.asyncio
+async def test_seed_whitespace_only_prose_returns_empty():
+    """Whitespace-only prose is treated the same as empty."""
+    mock = MockLLMClient(canned=[])
+    result = await seed_inter_bot_edges(
+        mock,
+        classifier_model="x",
+        bot_a_id="bot_a",
+        bot_a_name="Alice",
+        bot_b_id="bot_b",
+        bot_b_name="Bob",
+        relationship_prose="   \n  ",
+    )
+    assert result == RelationshipSeed()

tests/test_world.py

@@ -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_7(tmp_path):
+def test_schema_version_after_migration_is_8(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]) == 7
+        assert int(row[0]) == 8