claude/bevelina status 020905

src/service.ts

@@ -30,6 +30,7 @@ import { buildToolset, BUILTIN_REGISTRIES } from "./turn-runner/toolset";
 import { buildToolbox, renderToolbox, type ToolRegistry } from "./tools/catalog";
 import { composeInstructions } from "./turn-runner/soul";
 import { deliverPost } from "./adapter/outbound";
+import { ReplyStream } from "./adapter/reply-stream";
 import { routeMessage } from "./adapter/router";
 import type { SurfaceAdapter } from "@bevyl-ai/agent-tools";
 import type { AgentRuntimeSession, DynamicTool, AgentEvent } from "./turn-runner/types";
@@ -530,6 +531,18 @@ export class Service {
       // (SPEC §11) because a batch can span conversations and a guessed destination misroutes.
       const homeMsg = addressed.at(-1) ?? pending.at(-1)!;
       const anchorObj: Anchor = { venueId: homeMsg.venueId ?? "", threadRootId: homeMsg.threadRootId ?? homeMsg.ts };
+      // The home thread's reply is ONE native streamed message (reply-stream.ts): checklist
+      // cards buffer inside the stream until her first words materialize it, so a plan box
+      // alone never posts and never notifies (2026-07-20 live defect: a bare card-only
+      // checklist landed as her whole reply while she worked). Replies addressed elsewhere
+      // still go out as plain posts — a stream belongs to exactly one thread.
+      const stream = new ReplyStream({
+        adapter: this.d.adapter,
+        venueId: anchorObj.venueId,
+        threadTs: anchorObj.threadRootId,
+        recipient: homeMsg.principalId,
+        log: this.log,
+      });
       const effects: unknown[] = [];
       let failureCause = "";
       // §14.2 gate: flipped when a reply or react lands on a directly addressed message — a
@@ -550,14 +563,17 @@ export class Service {
         nudgeAfterMs: this.policy().tasks.nudgeAfterMs,
         permalink: (v, ts) => this.d.adapter.permalink?.(v, ts),
         postMessage: async (a, text) => {
-          const result = await this.postMessage(a, text);
+          const streamedId =
+            a.venueId === anchorObj.venueId && a.threadRootId === anchorObj.threadRootId ? await stream.post(text) : null;
+          const result = streamedId ? { messageId: streamedId } : await this.postMessage(a, text);
           if (direct.some((m) => a.venueId === (m.venueId ?? "") && a.threadRootId === (m.threadRootId ?? m.ts))) answered = true;
           // Optimistic close (ear design): answering in a thread settles its recorded debts the
           // moment the post lands — she never re-answers her own work. The ear can reopen.
           closeAttentionItemsForThread(this.d.db, this.d.clock, identityId, a.venueId, a.threadRootId ?? null, "answered in thread");
           return result;
         },
         updateMessage: this.d.adapter.updateMessage ? (v, m, t) => this.d.adapter.updateMessage!(v, m, t) : undefined,
+        renderChecklist: async (items) => stream.setCards(items),
         // Reactions reach any delivered message by venue + ts (the values in her lines). When
         // one lands on a message in this batch, it carries the same bookkeeping a reply does:
         // the §14.2 answered flip and the optimistic attention close for that message's thread.
@@ -659,6 +675,12 @@ export class Service {
           ).catch(() => {});
         }
       } finally {
+        // Close the home stream: a succeeded wake settles any still-pending cards (Slack
+        // renders a pending card on a stopped stream as "Something went wrong"); a failed
+        // wake drops buffered cards instead — a checked-off plan over a failure is a lie.
+        if (status === "succeeded") stream.settleCards();
+        else stream.clearCards();
+        await stream.close().catch(() => {});
         // Delivery is done even when the turn wasn't — re-delivering the same batch to a broken
         // thread just loops the failure (observed live pre-collapse); the fallback above settled
         // the addressed duty, and everything stays searchable.
@@ -795,7 +817,9 @@ export class Service {
       const knowledge = identities.map((i) => {
         const { kept, dropped } = coreWithinBudget(queryMemory(this.d.db, i.id, { tier: "core" }), this.policy().memory.coreCharBudget);
         if (dropped.length) this.log.warn("core memory over budget — items truncated from the soul (§8.6 hygiene defect)", { identityId: i.id, dropped: dropped.length });
-        return { identity: i.id, facts: kept.map((m) => m.content) };
+        // The dropped count rides into the soul so SHE curates (§8.6: curation is the fix;
+        // post-Collapse there is no distiller — an ordinary wake with memory tools is it).
+        return { identity: i.id, facts: kept.map((m) => m.content), dropped: dropped.length };
       });
       // §9.5: standing venue instructions ride the soul — standing config in the standing channel.
       const standing = identities.map((i) => ({ identity: i.id, venues: i.venueInstructions }));

src/turn-runner/soul.ts

@@ -227,7 +227,7 @@ did not write down is gone.
 // and keeps that snapshot for its life (same freshness contract as the other context slots).
 export function composeInstructions(
   personas: string[],
-  knowledge: { identity: string; facts: string[] }[] = [],
+  knowledge: { identity: string; facts: string[]; dropped?: number }[] = [],
   standing: { identity: string; venues: Record<string, string> }[] = [],
   toolDigests: { identity: string; digest: string }[] = [],
 ): string {
@@ -236,7 +236,13 @@ export function composeInstructions(
   parts.push(...voices.map((v) => `## Persona\n\n${v}`));
   for (const k of knowledge) {
     if (k.facts.length === 0) continue;
-    parts.push(`## What you know (as ${k.identity})\n\nDurable facts you carry into every conversation. Each keeps the strength it was saved at; your memory tools update them.\n\n${k.facts.map((f) => `- ${f}`).join("\n")}`);
+    // §8.6: truncation is the safety net, curation is the fix — and post-Collapse the curator
+    // is HER, on an ordinary wake. Telling her what fell off is what makes curation happen;
+    // a silent drop recurs forever (observed live 2026-07-20: 3 items truncated every wake).
+    const overflow = k.dropped
+      ? `\n\n(${k.dropped} more didn't fit your memory budget and are NOT loaded — they're still searchable. When you have a quiet moment, tidy up: merge overlapping facts, retire stale ones to archive with memory_tier, until everything durable fits.)`
+      : "";
+    parts.push(`## What you know (as ${k.identity})\n\nDurable facts you carry into every conversation. Each keeps the strength it was saved at; your memory tools update them.\n\n${k.facts.map((f) => `- ${f}`).join("\n")}${overflow}`);
   }
   for (const td of toolDigests) {
     if (!td.digest) continue;

test/ear.test.ts

@@ -177,7 +177,7 @@ describe("attention items (what she owes)", () => {
 
     expect(mindSessions()[0]!.prompts[0]).toContain("[still owed]");
     expect(mindSessions()[0]!.prompts[0]).toContain("julia asked for a ticket");
-    expect(adapter.posts.map((p) => p.text)).toContain("filed it");
+    expect(adapter.streams.map((s) => s.text)).toContain("filed it"); // home reply streams (reply-stream.ts)
     expect(openItems(db, "eng")).toHaveLength(0); // the reply into the thread settled the debt
     await service.stop();
   });
@@ -339,7 +339,7 @@ describe("step_back (standing engagement state)", () => {
     h.adapter.emit(msg({ text: "<@BOT1> ok actually help", mentionsBotId: true, ts: "20.4", threadRootTs: "20.0" }));
     await h.service.idle();
     expect(h.mindSessions()).toHaveLength(3);
-    expect(h.adapter.posts.map((p) => p.text)).toContain("back");
+    expect(h.adapter.streams.map((s) => s.text)).toContain("back"); // home reply streams (reply-stream.ts)
     await h.service.stop();
   });
 

test/resident.test.ts

@@ -206,8 +206,9 @@ describe("resident delivery", () => {
     await service.idle();
 
     expect(minds()).toHaveLength(2);
-    expect(adapter.posts).toHaveLength(1);
-    expect(adapter.posts[0]!.text).toBe("back — answering now");
+    expect(adapter.posts).toHaveLength(0);
+    expect(adapter.streams).toHaveLength(1); // one wake, one streamed home reply — retries share it
+    expect(adapter.lastStreamText()).toBe("back — answering now");
     await service.stop();
   });
 
@@ -252,8 +253,8 @@ describe("resident delivery", () => {
     await service.idle();
 
     expect(minds()).toHaveLength(2); // the dead attempt, then its retry
-    expect(adapter.posts).toHaveLength(1);
-    expect(adapter.posts[0]!.text).toBe("here — filing it");
+    expect(adapter.posts).toHaveLength(0);
+    expect(adapter.lastStreamText()).toBe("here — filing it");
     await service.stop();
   });
 
@@ -269,8 +270,8 @@ describe("resident delivery", () => {
 
     // the reply landed; nobody is left hanging, so the harness stays silent and doesn't retry
     expect(minds()).toHaveLength(1);
-    expect(adapter.posts).toHaveLength(1);
-    expect(adapter.posts[0]!.text).toBe("on it — checking now");
+    expect(adapter.posts).toHaveLength(0);
+    expect(adapter.lastStreamText()).toBe("on it — checking now");
     await service.stop();
   });
 
@@ -349,4 +350,64 @@ describe("resident delivery", () => {
     expect(adapter.posts[0]!.threadRootTs).toBe("1.0"); // ...in ITS thread, not the batch's last
     await service.stop();
   });
+
+  // The reply-stream contract (reply-stream.ts): checklist cards alone must never create (and
+  // notify on) a message — they buffer until her first words materialize the stream, then ride
+  // the SAME message as native task cards. Live defect 2026-07-20: the resident wake never wired
+  // the stream, so a bare card-only plan box posted as her whole reply while she worked.
+  test("checklist cards buffer until the reply materializes the stream — a plan box alone never posts", async () => {
+    const { adapter, service } = harness(async (_turn, tools) => {
+      if (tools.get("verdict")) return; // the ear bookkeeps quietly
+      await tools.get("checklist")!.run({ items: [{ text: "collect reports", done: false }, { text: "send the list", done: false }] });
+      await tools.get("reply")!.run({ text: "3 follow-ups, list below", venueId: "C1", threadRootId: "5.0" });
+      await tools.get("checklist")!.run({ items: [{ text: "collect reports", done: true }, { text: "send the list", done: false }] });
+    });
+    await service.start();
+    adapter.emit(msg({ text: "<@BOT1> organize today's reports", mentionsBotId: true, ts: "5.0" }));
+    await service.idle();
+
+    expect(adapter.posts).toHaveLength(0); // no standalone emoji checklist, no plain reply
+    expect(adapter.streams).toHaveLength(1); // ONE message carries cards + words
+    const stream = adapter.streams[0]!;
+    expect(stream.text).toBe("3 follow-ups, list below");
+    expect(stream.stopped).toBe(true);
+    const cards = adapter.taskCards.filter((c) => c.messageId === stream.messageId);
+    expect(cards.length).toBeGreaterThan(0);
+    // The stream closed with every card settled — Slack renders a pending card on a stopped
+    // stream as "Something went wrong".
+    const lastByCardId = new Map(cards.map((c) => [c.id, c.status]));
+    expect([...lastByCardId.values()].every((s) => s === "complete")).toBe(true);
+    await service.stop();
+  });
+
+  test("a wake that only plans and never speaks posts NOTHING — buffered cards die with the wake", async () => {
+    const { adapter, service } = harness(async (_turn, tools) => {
+      if (tools.get("verdict")) return;
+      await tools.get("checklist")!.run({ items: [{ text: "a plan with no words", done: false }] });
+    });
+    await service.start();
+    adapter.emit(msg({ text: "<@BOT1> hm", mentionsBotId: true, ts: "6.0" }));
+    await service.idle();
+
+    expect(adapter.posts).toHaveLength(0);
+    expect(adapter.streams).toHaveLength(0);
+    expect(adapter.taskCards).toHaveLength(0);
+    await service.stop();
+  });
+
+  test("when the surface has no native streaming, the reply falls back to a plain post", async () => {
+    const { adapter, service } = harness(async (_turn, tools) => {
+      if (tools.get("verdict")) return;
+      await tools.get("reply")!.run({ text: "plain delivery still works", venueId: "C1", threadRootId: "7.0" });
+    });
+    adapter.failStreams = true;
+    await service.start();
+    adapter.emit(msg({ text: "<@BOT1> ping", mentionsBotId: true, ts: "7.0" }));
+    await service.idle();
+
+    expect(adapter.streams).toHaveLength(0);
+    expect(adapter.posts).toHaveLength(1);
+    expect(adapter.posts[0]!.text).toBe("plain delivery still works");
+    await service.stop();
+  });
 });

test/service.test.ts

@@ -122,10 +122,13 @@ describe("Service inbound (SPEC §5, §17.1)", () => {
     adapter.emit(mention({ text: "<@BOT1> what's our SLA?", ts: "42.1" }));
     await service.idle();
 
-    expect(adapter.posts).toHaveLength(1);
-    expect(adapter.posts[0]!.text).toBe("ack");
-    expect(adapter.posts[0]!.venueId).toBe("C1");
-    expect(adapter.posts[0]!.threadRootTs).toBe("42.1"); // reply defaults to the addressing thread
+    // The home-thread reply rides ONE native streamed message (reply-stream.ts) — no plain post.
+    expect(adapter.posts).toHaveLength(0);
+    expect(adapter.streams).toHaveLength(1);
+    expect(adapter.lastStreamText()).toBe("ack");
+    expect(adapter.streams[0]!.venueId).toBe("C1");
+    expect(adapter.streams[0]!.threadTs).toBe("42.1"); // reply defaults to the addressing thread
+    expect(adapter.streams[0]!.stopped).toBe(true);
     await service.stop();
   });
 
@@ -426,8 +429,10 @@ describe("Service workers report to the mind (2026-07-13)", () => {
     await service.idle();
 
     expect(getTask(db, "T-1")?.status).toBe("done");
-    expect(adapter.streams).toHaveLength(0); // nobody streams anymore
-    const texts = adapter.posts.map((p) => p.text);
+    // The worker itself never posts or streams; everything the room hears is HER replies —
+    // streamed when the wake has a human home (the mention), plain when it doesn't (the
+    // report wake's task-update signal carries no recipient to stream to).
+    const texts = [...adapter.streams.map((s) => s.text), ...adapter.posts.map((p) => p.text)];
     expect(texts).toContain("on it");
     expect(texts.some((t) => t.includes("N+1 query"))).toBe(true); // HER voice, not the worker's
     expect(nonEar()).toHaveLength(3);
@@ -443,7 +448,8 @@ describe("Service workers report to the mind (2026-07-13)", () => {
     await service.idle();
 
     expect(nonEar()).toHaveLength(2); // wake + worker, no report wake
-    expect(adapter.posts.map((p) => p.text)).toEqual(["on it"]);
+    expect(adapter.posts).toHaveLength(0);
+    expect(adapter.streams.map((s) => s.text)).toEqual(["on it"]);
     await service.stop();
   });
 

test/soul.test.ts

@@ -26,4 +26,19 @@ describe("soul / composeInstructions", () => {
     // no empty persona heading left dangling
     expect(out).not.toMatch(/Persona\n+\s*\n+\s*Persona/);
   });
+
+  // §8.6: over-budget core truncates from injection, and curation is the fix — post-Collapse the
+  // curator is her, so the soul must SAY what fell off or the defect recurs silently forever.
+  test("an over-budget knowledge section tells her how many items didn't fit and to curate", () => {
+    const out = composeInstructions([], [{ identity: "eng", facts: ["fact one"], dropped: 3 }]);
+    expect(out).toContain("fact one");
+    expect(out).toContain("3 more didn't fit your memory budget");
+    expect(out).toContain("memory_tier");
+  });
+
+  test("a within-budget knowledge section carries no overflow note", () => {
+    const out = composeInstructions([], [{ identity: "eng", facts: ["fact one"] }]);
+    expect(out).toContain("fact one");
+    expect(out).not.toContain("memory budget");
+  });
 });