diff --git a/src/service.ts b/src/service.ts index d35876e..ed64fa5 100644 --- a/src/service.ts +++ b/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,7 +563,9 @@ 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. @@ -558,6 +573,7 @@ export class Service { 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 })); diff --git a/src/turn-runner/soul.ts b/src/turn-runner/soul.ts index 7b13f54..9ebb8b8 100644 --- a/src/turn-runner/soul.ts +++ b/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 }[] = [], 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; diff --git a/test/ear.test.ts b/test/ear.test.ts index 155a983..3535c24 100644 --- a/test/ear.test.ts +++ b/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(); }); diff --git a/test/resident.test.ts b/test/resident.test.ts index 1353849..8350f63 100644 --- a/test/resident.test.ts +++ b/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(); + }); }); diff --git a/test/service.test.ts b/test/service.test.ts index 9533589..2e84a48 100644 --- a/test/service.test.ts +++ b/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(); }); diff --git a/test/soul.test.ts b/test/soul.test.ts index 819fb88..fbaea1d 100644 --- a/test/soul.test.ts +++ b/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"); + }); });