From 667cc0ab656be0c299710ff983824d47525028bd Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Thu, 13 Aug 2026 18:28:41 -0400 Subject: [PATCH 1/2] task_create homes by ref: the task's home conversation is her explicit choice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live 2026-08-13 (T-354): a wake batch spanning several alert threads created 'Investigate process-video-task failure burst' homed by the batch-level guess (last addressed message) — Noah's 'pull it together blacksmith' tease — so the worker's correct report delivered into the wrong incident's thread and she voiced it there. The deferred task_create source-binding audit item, come due. Same medicine as reply/react (R4): task_create gains a required ref (^r\d+$) and homes to conversationOf(that target). The heuristic can no longer route anything — service.ts's homeMsg now only seats the reply stream and the §14.2 fallback. Refless calls bounce with the standard 'is not a ref' card. Regression: two-conversation boot wake, task ref'd at the first conversation, asserts home != the batch's last address (the exact T-354 shape). Co-Authored-By: Claude Fable 5 --- src/service.ts | 9 +++++---- src/turn-runner/toolset.ts | 25 ++++++++++++++++++------ test/resident.test.ts | 39 ++++++++++++++++++++++++++++++++++---- test/service.test.ts | 12 +++++++----- test/toolset.test.ts | 18 +++++++++++------- 5 files changed, 77 insertions(+), 26 deletions(-) diff --git a/src/service.ts b/src/service.ts index e11f96c..e220cc0 100644 --- a/src/service.ts +++ b/src/service.ts @@ -571,10 +571,11 @@ export class Service { // not spoken TO her — a dead wake over thread chatter fails into the log, never the room // (SPEC §18: "a thread-follow turn's failure is ledger/log-only"). const direct = pending.filter(isDirectAddress); - // Tasks born in this wake home to the conversation that most recently engaged her (the - // last addressed message, else the last overheard one) — its thread gets the checklist - // and progress posts. Posting is never homed: reply/react take explicit coordinates - // (SPEC §11) because a batch can span conversations and a guessed destination misroutes. + // The wake's primary conversation (last addressed, else last overheard) — where the + // native reply stream rides and where a §14.2 death fallback lands. NOTHING routes by + // this guess: replies/reacts address by ref (SPEC §11), and tasks home to a required + // ref too (2026-08-13 live: the batch-level guess homed an incident task to an + // adjacent thread and its report answered the wrong incident). 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 diff --git a/src/turn-runner/toolset.ts b/src/turn-runner/toolset.ts index 64dc01d..18e7784 100644 --- a/src/turn-runner/toolset.ts +++ b/src/turn-runner/toolset.ts @@ -182,24 +182,37 @@ function taskCreateTool(ctx: ToolsetContext): ToolFactory { spec: { name: "task_create", description: - "Record a new delegated task; a worker runs it and reports back to you. Input: { title, spec, tier? }. tier is how hard the worker thinks: 'low' for routine mechanical work (tailing a ticket, fetching status), 'medium' for normal work, 'high' (default) for problems that need real thought. Write the spec as a full handoff — the worker starts with none of this conversation.", + "Record a new delegated task; a worker runs it and reports back to you. Input: { title, spec, ref, tier? }. ref is the [rN] tag of the conversation (or a message in it) this task is FOR — the worker's report comes home to that conversation, so pick the room that asked for the work, not whoever spoke last. tier is how hard the worker thinks: 'low' for routine mechanical work (tailing a ticket, fetching status), 'medium' for normal work, 'high' (default) for problems that need real thought. Write the spec as a full handoff — the worker starts with none of this conversation.", inputSchema: { type: "object", additionalProperties: false, - required: ["title", "spec"], - properties: { title: { type: "string" }, spec: { type: "string" }, tier: { type: "string", enum: ["low", "medium", "high"] } }, + required: ["title", "spec", "ref"], + properties: { + title: { type: "string" }, + spec: { type: "string" }, + ref: { type: "string", pattern: "^r\\d+$" }, + tier: { type: "string", enum: ["low", "medium", "high"] }, + }, }, }, impl: async (args) => { - const a = args as { title: string; spec: string; tier?: "low" | "medium" | "high" }; - if (!ctx.anchor || !ctx.principal || !ctx.originEventId) return { success: false, output: "missing turn context for task_create" }; + const a = args as { title: string; spec: string; ref?: string; tier?: "low" | "medium" | "high" }; + if (!ctx.principal || !ctx.originEventId) return { success: false, output: "missing turn context for task_create" }; + // The task's home is HER call, bound to a rendered conversation — never a batch-level + // guess (live 2026-08-13: a task about an alert burst homed to the last thread that + // happened to address her, and its report answered an adjacent incident). + const target = a.ref ? ctx.refs?.get(a.ref) : undefined; + if (!target) { + return { success: false, output: `"${a.ref ?? ""}" is not a ref — home the task with the [rN] tag of the conversation its report belongs in` }; + } + const home = conversationOf(target); const task = createTask(ctx.db, ctx.clock, { id: nextTaskId(ctx.db), identityId: ctx.identity.id, title: a.title, spec: a.spec, sponsorId: ctx.principal.id, - homeAnchor: ctx.anchor, + homeAnchor: { venueId: home.venueId, threadRootId: home.threadRootId }, originEventId: ctx.originEventId, tier: a.tier, sponsorIsOperator: ctx.principal.isOperator, diff --git a/test/resident.test.ts b/test/resident.test.ts index 294db80..f5ce954 100644 --- a/test/resident.test.ts +++ b/test/resident.test.ts @@ -287,7 +287,7 @@ describe("resident delivery", () => { // outcome-report wake included. Act exactly once, and let the spawned execution finish // its task cleanly, or the test loops (task_create per wake / yield-redispatch forever). let acted = false; - const { adapter, service, db } = harness(async (_turn, tools) => { + const { adapter, service, db } = harness(async (_turn, tools, _act, prompt) => { const complete = tools.get("task_complete"); if (complete) { await complete.run({ report: "done" }); @@ -296,7 +296,7 @@ describe("resident delivery", () => { const taskCreate = tools.get("task_create"); if (!taskCreate || acted) return; acted = true; - await taskCreate.run({ title: "file the export bug", spec: "repro + ticket" }); + await taskCreate.run({ title: "file the export bug", spec: "repro + ticket", ref: refIn(prompt, "file this") }); throw new Error("died after acting"); }); await service.start(); @@ -373,10 +373,10 @@ describe("resident delivery", () => { test("a task born in a wake homes to the conversation that addressed her", async () => { let sessions = 0; - const { adapter, service, db } = harness(async (_n, t) => { + const { adapter, service, db } = harness(async (_n, t, _act, prompt) => { // 1: the wake that delegates; 2: the worker; 3+: the report wake (does nothing) const which = ++sessions; - if (which === 1) await t.get("task_create")!.run({ title: "dig", spec: "dig in" }); + if (which === 1) await t.get("task_create")!.run({ title: "dig", spec: "dig in", ref: refIn(prompt, /<#C1>/) }); if (which === 2) await t.get("task_complete")!.run({ report: "done" }); }); await service.start(); @@ -423,6 +423,37 @@ describe("resident delivery", () => { await service.stop(); }); + // Same live defect, task edition (2026-08-13, T-354): a wake batch spanning two conversations, + // and the task homed to whichever one the harness guessed (the batch's last address) — so the + // worker's report answered an adjacent incident. task_create homes by HER ref or not at all. + test("§11: a task homes to the ref'd conversation, not the batch's last address — a refless task_create is rejected", async () => { + const db = openLedger(":memory:"); + const seed = db.query( + `INSERT INTO events (id, dedup_key, kind, identity_id, venue_id, thread_root_id, principal_id, payload, received_at) + VALUES (?, ?, 'addressed_message', 'eng', ?, ?, 'U1', ?, '2026-07-01T00:00:00Z')`, + ); + seed.run("e1", "k1", "C1", "1.0", JSON.stringify({ text: "<@BOT1> alert burst, investigate", ts: "1.1", addressMode: "mention" })); + seed.run("e2", "k2", "C2", null, JSON.stringify({ text: "<@BOT1> pull it together blacksmith", ts: "2.0", addressMode: "mention" })); + + const rejected: string[] = []; + const { service } = harness(async (_turn, tools, _mark, prompt) => { + const taskCreate = tools.get("task_create"); + if (!taskCreate) return; // the ear / the worker (which never reaches its report here) + const bare = await taskCreate.run({ title: "dig", spec: "s" }); + expect(bare.success).toBe(false); + rejected.push(bare.output); + await taskCreate.run({ title: "dig", spec: "s", ref: refIn(prompt, "alert burst") }); + }, db); + await service.start(); + await service.idle(); // flushes the boot wake carrying both conversations + + expect(rejected[0]).toContain("is not a ref"); + const row = db.query("SELECT home_venue_id, home_thread_root_id FROM tasks").get() as { home_venue_id: string; home_thread_root_id: string | null } | null; + expect(row?.home_venue_id).toBe("C1"); // the incident's thread... + expect(row?.home_thread_root_id).toBe("1.0"); // ...not C2, the batch's last-addressed guess + 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 diff --git a/test/service.test.ts b/test/service.test.ts index eebdb7e..b5993e3 100644 --- a/test/service.test.ts +++ b/test/service.test.ts @@ -273,8 +273,8 @@ describe("Service dispatch driver (SPEC §6.2, §17.3, §17.4)", () => { const { db, adapter, service } = makeService({ // Kind-aware script (the ear shifted session ordering; indices were a trap): the ear holds, // the worker completes, the first wake delegates, later wakes choose silence. - sessionFactory: (tools) => - new FakeAgentRuntimeSession(tools, async (_turn, t) => { + sessionFactory: (tools) => { + const sess: FakeAgentRuntimeSession = new FakeAgentRuntimeSession(tools, async (_turn, t) => { if (t.get("verdict")) return; // the ear: nothing needs her const complete = t.get("task_complete"); if (complete) { @@ -283,10 +283,12 @@ describe("Service dispatch driver (SPEC §6.2, §17.3, §17.4)", () => { } if (!delegated) { delegated = true; - await t.get("task_create")!.run({ title: "dig in", spec: "why slow" }); + await t.get("task_create")!.run({ title: "dig in", spec: "why slow", ref: firstRef(sess) }); } // later wakes (the worker's report) — she chooses silence - }), + }); + return sess; + }, }); await service.start(); @@ -403,7 +405,7 @@ describe("Service workers report to the mind (2026-07-13)", () => { } if (!delegated) { delegated = true; - await t.get("task_create")!.run({ title: "dig", spec: "dig into the export bug", tier: "low" }); + await t.get("task_create")!.run({ title: "dig", spec: "dig into the export bug", tier: "low", ref: firstRef(sess) }); await t.get("reply")!.run({ text: "on it", ref: firstRef(sess) }); return; } diff --git a/test/toolset.test.ts b/test/toolset.test.ts index 5834aaf..49856d7 100644 --- a/test/toolset.test.ts +++ b/test/toolset.test.ts @@ -39,7 +39,11 @@ function identity(overrides: Partial = {}): IdentityConfig { function baseCtx(db: ReturnType, clock: Clock, overrides: Partial = {}): ToolsetContext { const posts: { anchor: any; text: string }[] = []; + // A standing rendered ref for the wake's home conversation — what task_create homes to. + const refs = makeRefTable(); + refs.mint({ venueId: "C1", threadRootId: null, via: "rendered" }); // r1 return { + refs, db, clock, identity: identity(), @@ -72,7 +76,7 @@ describe("task_create (SPEC §5.3, §11)", () => { const ctx = baseCtx(db, clock); const tools = buildToolset(ctx); - const result = await tool(tools, "task_create").run({ title: "dig in", spec: "why is it slow" }); + const result = await tool(tools, "task_create").run({ title: "dig in", spec: "why is it slow", ref: "r1" }); expect(result.success).toBe(true); const parsed = JSON.parse(result.output); expect(parsed.taskId).toBe("T-1"); @@ -100,7 +104,7 @@ describe("task_create (SPEC §5.3, §11)", () => { const tools = buildToolset(baseCtx(db, clock)); const create = tool(tools, "task_create"); expect(JSON.stringify(create.spec.inputSchema)).not.toContain("recurrence"); - const result = await create.run({ title: "t", spec: "s", recurrence: "every day" }); + const result = await create.run({ title: "t", spec: "s", ref: "r1", recurrence: "every day" }); expect(result.success).toBe(true); // the stray arg is ignored, never stored const row = db.query("SELECT recurrence FROM tasks WHERE id = 'T-1'").get() as { recurrence: string | null }; expect(row.recurrence).toBeNull(); @@ -111,7 +115,7 @@ describe("task_create (SPEC §5.3, §11)", () => { describe("task_steer / task_cancel / task_confirm", () => { async function activeTask(db: ReturnType, clock: Clock, ctx: ToolsetContext) { seedEvent(db, "e1", clock); - await tool(buildToolset(ctx), "task_create").run({ title: "t", spec: "s" }); + await tool(buildToolset(ctx), "task_create").run({ title: "t", spec: "s", ref: "r1" }); transition(db, clock, "T-1", "active", { type: "dispatch", executionId: "x1" }); } @@ -203,7 +207,7 @@ describe("task_query returns the identity's ledger view", () => { async function activeCreate(db: ReturnType, clock: Clock, ctx: ToolsetContext) { seedEvent(db, "e1", clock); - await tool(buildToolset(ctx), "task_create").run({ title: "t", spec: "s" }); + await tool(buildToolset(ctx), "task_create").run({ title: "t", spec: "s", ref: "r1" }); } }); @@ -317,7 +321,7 @@ describe("execution_step outcome tools (SPEC §6.3, §17.4)", () => { async function activeExecutionCtx(db: ReturnType, clock: Clock) { const createCtx = baseCtx(db, clock); seedEvent(db, "e1", clock); - await tool(buildToolset(createCtx), "task_create").run({ title: "t", spec: "s" }); + await tool(buildToolset(createCtx), "task_create").run({ title: "t", spec: "s", ref: "r1" }); transition(db, clock, "T-1", "active", { type: "dispatch", executionId: "x1" }); return baseCtx(db, clock, { turnKind: "execution_step", taskId: "T-1", anchor: { venueId: "C1", threadRootId: null } }); } @@ -402,7 +406,7 @@ describe("external tool: grant + scope + action-class confirmation flow", () => const clock = fakeClock(); seedEvent(db, "e1", clock); const createCtx = baseCtx(db, clock); - await tool(buildToolset(createCtx), "task_create").run({ title: "t", spec: "s" }); + await tool(buildToolset(createCtx), "task_create").run({ title: "t", spec: "s", ref: "r1" }); transition(db, clock, "T-1", "active", { type: "dispatch", executionId: "x1" }); const execCtx = baseCtx(db, clock, { @@ -565,7 +569,7 @@ describe("audit_query (SPEC §15: granted per identity, scoped to that identity) const tools = buildToolset(ctx); expect(tools.some((t) => t.spec.name === "audit_query")).toBe(true); - await tool(tools, "task_create").run({ title: "t", spec: "s" }); + await tool(tools, "task_create").run({ title: "t", spec: "s", ref: "r1" }); const result = await tool(tools, "audit_query").run({ kind: "task_created" }); const records = JSON.parse(result.output); expect(records).toHaveLength(1); From bb8ed46ae239823ae5b01b5fc4b505028e5f49ba Mon Sep 17 00:00:00 2001 From: Noah Lindner Date: Thu, 13 Aug 2026 19:48:31 -0400 Subject: [PATCH 2/2] =?UTF-8?q?kill=20every=20remaining=20guessed=20destin?= =?UTF-8?q?ation:=20refs=20seat=20streams,=20cards,=20provenance,=20and=20?= =?UTF-8?q?the=20=C2=A714.2=20apology?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The T-354 class, hunted to extinction. An adversarial audit (19 agents) found the batch-level guess (addressed.at(-1) ?? pending.at(-1)) surviving in five more places after the task-homing fix, plus two silent-drop bugs of the same family: - task_create sponsor/origin still came from the batch pick: tasks homed to one thread but sponsored by a speaker in another. Both now bind to the ref's own provenance (RefTarget carries eventId/principalId, minted by the renderer; provenanceOfRef resolves the rest). Machine-authored lines fall back to the newest human IN that conversation (lastSpeakerIn), never across the batch. - task_confirm recorded the wake-level principal as approver. Now requires the MESSAGE ref of the approve/deny line (conversation refs and via:'search' refs rejected) — the durable resolution names who actually said yes. - task_steer/task_cancel bind their source event to the asking message's ref. - checklist was the one posting tool with no ref; its cards could only land on the guessed home. Required ref, seated on conversationOf(ref), holder keyed per conversation; setCards now refuses streams that can never open, so the fallback actually fires instead of reporting cards nobody will see. - The single pre-seated ReplyStream became lazy per-conversation streams — seat and recipient follow her first ref-addressed post, so a multi-room wake streams every reply natively instead of degrading the 'wrong' ones. - §14.2: one wake-scoped answered boolean let any answer anywhere silence every other owed asker, and direct.at(-1) sent the one apology to a batch-tail guess. Per-conversation answered set; one fallback per owed conversation. - A refless ear hold/wake returned 'noted' while recording nothing (the 2026-08-10 discarded-judgment shape). Bounces with a correctable error. - React acts file at the ref target's own thread (a tail-line react used to file at the surface and render in the wrong conversation). Resident ToolsetContext now carries anchor: null — no batch-level anchor exists to be reused. principal remains for broker gating only. Reviewed by a second adversarial workflow (21 agents): 9 confirmed findings all fixed, including a wiring loss (originEventId dropped, task_steer/task_cancel dead on live wakes while the suite stayed green) now pinned by a regression that drives both tools through Service.runWake()'s own toolset. Co-Authored-By: Claude Fable 5 --- src/adapter/reply-stream.ts | 9 +- src/ledger/conversations.ts | 107 ++++++++++++++-- src/service.ts | 165 +++++++++++++++---------- src/turn-runner/execution-loop.ts | 2 +- src/turn-runner/toolset.ts | 198 +++++++++++++++++++++++------- test/ear.test.ts | 52 +++++--- test/resident.test.ts | 177 ++++++++++++++++++++++++-- test/toolset.test.ts | 20 ++- 8 files changed, 582 insertions(+), 148 deletions(-) diff --git a/src/adapter/reply-stream.ts b/src/adapter/reply-stream.ts index e6f986b..1f84c31 100644 --- a/src/adapter/reply-stream.ts +++ b/src/adapter/reply-stream.ts @@ -84,10 +84,15 @@ export class ReplyStream { }); } - // Replace the checklist. Returns false when the surface has no native cards (the caller falls - // back to its own checklist rendering). Buffered until the message exists; live afterwards. + // Replace the checklist. Returns false when the surface has no native cards OR this stream + // can never carry them (already failed, or missing open()'s preconditions with no message + // yet) — the caller falls back to its own checklist rendering instead of trusting cards that + // would silently never render (audit 2026-08-13: a seat with no thread/recipient reported + // success while showing nothing). Buffered until the message exists; live afterwards. setCards(items: ChecklistItem[]): boolean { if (!this.opts.adapter.appendTaskUpdate) return false; + if (this.failed) return false; + if (!this.msg && (!this.opts.threadTs || !this.opts.recipient || !this.opts.adapter.startStream)) return false; this.cards = items; const m = this.msg; if (m) void this.enqueue(() => this.flushCards(m.messageId)); diff --git a/src/ledger/conversations.ts b/src/ledger/conversations.ts index d054bcc..1ce0828 100644 --- a/src/ledger/conversations.ts +++ b/src/ledger/conversations.ts @@ -472,6 +472,11 @@ export interface RefTarget { threadRootId: string | null; ts?: string; // present on message refs — the message's own surface ts via: "rendered" | "search"; + // Provenance, when the renderer knew it at mint time: the exact event behind the line and who + // spoke it. Durable writes (a task's sponsor/origin, a confirmation's approver) bind to these + // — never to any batch-level "whoever addressed her last" pick (T-354's second half). + eventId?: string; + principalId?: string | null; } export interface RefTable { @@ -498,12 +503,82 @@ export function conversationOf(t: RefTarget): ConversationKey { return { venueId: t.venueId, threadRootId: t.threadRootId ?? t.ts ?? null }; } +// The event (and speaker) a ref stands on, for durable writes. Renderer-minted refs carry it; +// for the rest (search refs, older mints) it resolves from the ledger: the exact event when the +// ref names a message, else the newest event IN the ref's conversation — always within the +// conversation the model chose, never across the batch. Null when the conversation has no +// recorded events (nothing to bind to — callers bounce rather than guess). +export function provenanceOfRef(db: Database, identityId: string, t: RefTarget): { eventId: string; principalId: string | null } | null { + if (t.eventId) return { eventId: t.eventId, principalId: t.principalId ?? null }; + if (t.ts) { + const exact = db + .query( + `SELECT id, principal_id FROM events + WHERE identity_id = ? AND venue_id = ? AND json_extract(payload, '$.ts') = ? + ORDER BY rowid DESC LIMIT 1`, + ) + .get(identityId, t.venueId, t.ts) as { id: string; principal_id: string | null } | null; + if (exact) return { eventId: exact.id, principalId: exact.principal_id }; + } + const key = conversationOf(t); + const row = ( + key.threadRootId + ? db + .query( + `SELECT id, principal_id FROM events + WHERE identity_id = ? AND venue_id = ? + AND kind IN ('addressed_message','observed_message','external_signal') + AND (thread_root_id = ? OR json_extract(payload, '$.ts') = ?) + ORDER BY rowid DESC LIMIT 1`, + ) + .get(identityId, key.venueId, key.threadRootId, key.threadRootId) + : db + .query( + `SELECT id, principal_id FROM events + WHERE identity_id = ? AND venue_id = ? + AND kind IN ('addressed_message','observed_message','external_signal') + AND thread_root_id IS NULL + ORDER BY rowid DESC LIMIT 1`, + ) + .get(identityId, key.venueId) + ) as { id: string; principal_id: string | null } | null; + return row ? { eventId: row.id, principalId: row.principal_id } : null; +} + +// The newest HUMAN speaker in a conversation — the sponsor fallback when a ref's own line is +// machine-authored (a worker report has no principal). Scoped to the conversation the model +// chose; never a batch-level pick. +export function lastSpeakerIn(db: Database, identityId: string, key: ConversationKey): string | null { + const row = ( + key.threadRootId + ? db + .query( + `SELECT principal_id FROM events + WHERE identity_id = ? AND venue_id = ? AND principal_id IS NOT NULL + AND (thread_root_id = ? OR json_extract(payload, '$.ts') = ?) + ORDER BY rowid DESC LIMIT 1`, + ) + .get(identityId, key.venueId, key.threadRootId, key.threadRootId) + : db + .query( + `SELECT principal_id FROM events + WHERE identity_id = ? AND venue_id = ? AND principal_id IS NOT NULL + AND thread_root_id IS NULL + ORDER BY rowid DESC LIMIT 1`, + ) + .get(identityId, key.venueId) + ) as { principal_id: string } | null; + return row?.principal_id ?? null; +} + // --- the one renderer ------------------------------------------------------------------------ interface TailLine { sortTs: number; surfaceTs: string | null; // their messages carry one (a ref target); her acts render bare line: string; // formatted WITHOUT a ref prefix — the renderer prepends the minted ref + eventId?: string; // provenance for the minted ref (their messages only) + principalId?: string | null; } function who(p: { principalId: string | null; principalName?: string }): string { @@ -530,7 +605,7 @@ function tailOf(db: Database, identityId: string, key: ConversationKey, beforeRo key.threadRootId ? db .query( - `SELECT principal_id, json_extract(payload, '$.text') AS text, json_extract(payload, '$.principalName') AS name, + `SELECT id, principal_id, json_extract(payload, '$.text') AS text, json_extract(payload, '$.principalName') AS name, json_extract(payload, '$.ts') AS ts FROM events WHERE identity_id = ? AND venue_id = ? AND rowid <= ? @@ -541,7 +616,7 @@ function tailOf(db: Database, identityId: string, key: ConversationKey, beforeRo .all(identityId, key.venueId, beforeRowid, key.threadRootId, key.threadRootId, TAIL_LIMIT) : db .query( - `SELECT principal_id, json_extract(payload, '$.text') AS text, json_extract(payload, '$.principalName') AS name, + `SELECT id, principal_id, json_extract(payload, '$.text') AS text, json_extract(payload, '$.principalName') AS name, json_extract(payload, '$.ts') AS ts FROM events WHERE identity_id = ? AND venue_id = ? AND rowid <= ? @@ -551,6 +626,7 @@ function tailOf(db: Database, identityId: string, key: ConversationKey, beforeRo ) .all(identityId, key.venueId, beforeRowid, TAIL_LIMIT) ) as { + id: string; principal_id: string | null; text: string | null; name: string | null; @@ -559,6 +635,8 @@ function tailOf(db: Database, identityId: string, key: ConversationKey, beforeRo const theirs: TailLine[] = events.reverse().map((r) => ({ sortTs: r.ts ? Number(r.ts) : 0, surfaceTs: r.ts, + eventId: r.id, + principalId: r.principal_id, line: `${who({ principalId: r.principal_id, ...(r.name ? { principalName: r.name } : {}) })}: ${(r.text ?? "").slice(0, 300)}`, })); const acts = db @@ -612,18 +690,33 @@ export function renderConversation(db: Database, identityId: string, key: Conver if (opts.judgment?.wakeWhy) { headerBits.push(`first read: ${opts.judgment.wakeWhy}`); } - const cref = opts.refs?.mint({ venueId: key.venueId, threadRootId: key.threadRootId, via: "rendered" }); + // A conversation ref carries the provenance of its newest delivered line (who asked, which + // event) so durable writes through a conversation-level ref still bind inside the right room. + const lastNew = opts.newMessages.at(-1); + const cref = opts.refs?.mint({ + venueId: key.venueId, + threadRootId: key.threadRootId, + via: "rendered", + ...(lastNew ? { eventId: lastNew.id, principalId: lastNew.principalId } : {}), + }); const address = cref ? `${cref} ${where}` : where; const header = headerBits.length || cref ? `[${address}${headerBits.length ? `: ${headerBits.join(" | ")}` : ""}]\n` : ""; - const tag = (surfaceTs: string | null): string => { + const tag = (surfaceTs: string | null, eventId?: string, principalId?: string | null): string => { if (!opts.refs || !surfaceTs) return ""; - return `[${opts.refs.mint({ venueId: key.venueId, threadRootId: key.threadRootId, ts: surfaceTs, via: "rendered" })}] `; + return `[${opts.refs.mint({ + venueId: key.venueId, + threadRootId: key.threadRootId, + ts: surfaceTs, + via: "rendered", + ...(eventId ? { eventId } : {}), + ...(principalId !== undefined ? { principalId } : {}), + })}] `; }; const tail = tailOf(db, identityId, key, opts.beforeRowid, selfLabel); const tailBlock = tail.length - ? `earlier in ${where} (already heard — so you can tell who is talking to whom):\n${tail.map((t) => ` ${tag(t.surfaceTs)}${t.line}`).join("\n")}\n` + ? `earlier in ${where} (already heard — so you can tell who is talking to whom):\n${tail.map((t) => ` ${tag(t.surfaceTs, t.eventId, t.principalId)}${t.line}`).join("\n")}\n` : ""; - const newLines = opts.newMessages.map((m) => `${tag(m.ts)}${mark(m)}${inboxLine(m)}`).join("\n"); + const newLines = opts.newMessages.map((m) => `${tag(m.ts, m.id, m.principalId)}${mark(m)}${inboxLine(m)}`).join("\n"); return `${header}${tailBlock}${newLines}`; } diff --git a/src/service.ts b/src/service.ts index e220cc0..fa81df1 100644 --- a/src/service.ts +++ b/src/service.ts @@ -438,6 +438,12 @@ export class Service { if (a.ref && !target) { return { success: false, output: `"${a.ref}" is not a ref — copy the [rN] tag (like r3) from the start of the line you are judging; timestamps and channel ids are labels, not addresses` }; } + // A hold/wake without a ref has nowhere durable to live — bounced, never nodded + // through (audit 2026-08-13: a refless hold returned "noted" while recording nothing, + // the 2026-08-10 discarded-judgment failure wearing a polite face). + if ((a.decision === "hold" || a.decision === "wake") && !target) { + return { success: false, output: `${a.decision} needs ref — the [rN] tag of a line in the conversation being judged, so the judgment lands on its row` }; + } const venueId = target?.venueId; // hold/wake judge the conversation the message LIVES in (a top-level line is surface // traffic); open_ask ROOTS the debt at the ask itself, where its answer will land. @@ -571,25 +577,31 @@ export class Service { // not spoken TO her — a dead wake over thread chatter fails into the log, never the room // (SPEC §18: "a thread-follow turn's failure is ledger/log-only"). const direct = pending.filter(isDirectAddress); - // The wake's primary conversation (last addressed, else last overheard) — where the - // native reply stream rides and where a §14.2 death fallback lands. NOTHING routes by - // this guess: replies/reacts address by ref (SPEC §11), and tasks home to a required - // ref too (2026-08-13 live: the batch-level guess homed an incident task to an - // adjacent thread and its report answered the wrong incident). - 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, - }); + // Broker gating (guest checks) keys on the wake's most recent human addresser — policy, + // not routing: no destination and no durable row derives from this pick. Everything that + // lands somewhere (replies, reacts, cards, tasks, confirmations, the §14.2 fallback) + // routes by ref or by the exact events owed (audit 2026-08-13: every seat this pick used + // to feed misrouted live at least once). + const gatingMsg = addressed.at(-1) ?? pending.at(-1)!; + // One native streamed message PER CONVERSATION she speaks into (reply-stream.ts): the + // stream for a conversation is created lazily at her first ref-addressed post there, so + // the seat is always model-chosen. Checklist cards buffer inside their conversation's + // stream until her first words materialize it — a plan box alone never posts and never + // notifies (2026-07-20 live defect). The recipient is that conversation's own last human + // speaker in the batch (a worker-report conversation has none; its stream fails open to + // plain posts). + const streams = new Map(); + const streamFor = (a: Anchor): ReplyStream => { + const k = convoKey(a.venueId, a.threadRootId); + let s = streams.get(k); + if (!s) { + const recipient = + [...pending].reverse().find((m) => m.principalId && convoKey(m.venueId ?? "", m.threadRootId ?? m.ts) === k)?.principalId ?? null; + s = new ReplyStream({ adapter: this.d.adapter, venueId: a.venueId, threadTs: a.threadRootId, recipient, log: this.log }); + streams.set(k, s); + } + return s; + }; const effects: unknown[] = []; let failureCause = ""; // §5.5 stale-reply withholding: nobody addressed this wake directly, so a reply races the @@ -636,8 +648,7 @@ export class Service { if (!act.inserted) continue; // an earlier attempt of this wake already sent it let result: { messageId: string }; try { - const streamedId = - b.anchor.venueId === anchorObj.venueId && b.anchor.threadRootId === anchorObj.threadRootId ? await stream.post(b.text) : null; + const streamedId = await streamFor(b.anchor).post(b.text); result = streamedId ? { messageId: streamedId } : await this.postMessage(b.anchor, b.text); } catch (e) { deleteAct(this.d.db, wakeId, act.actKey); @@ -657,27 +668,32 @@ export class Service { effects.push({ kind: "posted", anchor: b.anchor, text: b.text }); } }; - // §14.2 gate: flipped when a reply or react lands on a directly addressed message — a - // wake that answered someone before dying leaves nobody hanging, so no fallback. Every - // flip must co-occur with a pushed effect (the same tool call records one): the retry - // loop's effects-nonempty guard is what keeps a later attempt from seeing answered=true - // off a prior attempt's partial work. - let answered = false; + // §14.2 gate, PER CONVERSATION: a post or react into a conversation marks IT answered — + // a wake that answered one asker before dying still owes the others their fallback (audit + // 2026-08-13: one wake-scoped boolean let any answer anywhere silence every other owed + // conversation, and the apology itself went to a batch-tail guess). Every insertion + // co-occurs with a pushed effect (the same tool call records one): the retry loop's + // effects-nonempty guard is what keeps a later attempt from seeing prior partial work as + // its own. + const answeredConvos = new Set(); // Built PER ATTEMPT (below): tool factories carry per-turn state (the reply tool's // step-back bounce). A retry is a fresh session that never saw a prior attempt's tool // results, so it must re-decide against re-armed tools — a bounce a dead attempt consumed - // must not wave the next attempt through. Shared wake state (effects, stream, answered, - // the checklist holder) lives out here and survives rebuilds. - const checklist = { messageId: null as string | null }; + // must not wave the next attempt through. Shared wake state (effects, streams, + // answeredConvos, the checklist holder) lives out here and survives rebuilds. + const checklist = new Map(); const makeTools = () => buildToolset({ db: this.d.db, clock: this.d.clock, identity, turnKind: "resident", catalog: this.catalog, - anchor: anchorObj, - principal: this.principalOf(homeMsg.principalId), - originEventId: homeMsg.id, + // No batch-level anchor exists to be reused: resident posting scope is venue-wide, and + // every tool that needs a place gets it from a ref. principal serves broker gating only + // — durable writes (task sponsor/origin, confirmation approver) bind to ref provenance. + anchor: null, + principal: this.principalOf(gatingMsg.principalId), + resolvePrincipal: (id) => this.principalOf(id), nudgeAfterMs: this.policy().tasks.nudgeAfterMs, outwardScopeId: wakeId, permalink: (v, ts) => this.d.adapter.permalink?.(v, ts), @@ -686,8 +702,7 @@ export class Service { if (!act.inserted) return { messageId: "already-sent-this-wake" }; // a retry attempt re-issuing the identical post is a no-op let result: { messageId: string }; try { - const streamedId = - a.venueId === anchorObj.venueId && a.threadRootId === anchorObj.threadRootId ? await stream.post(text) : null; + const streamedId = await streamFor(a).post(text); result = streamedId ? { messageId: streamedId } : await this.postMessage(a, text); } catch (e) { deleteAct(this.d.db, wakeId, act.actKey); // intent must not outlive a failed call @@ -702,20 +717,25 @@ export class Service { // message id) — her opening message must render in that thread, not on the surface. setActTs(this.d.db, wakeId, act.actKey, result.messageId, a.threadRootId ?? result.messageId); engage(this.d.db, this.d.clock, identityId, a.venueId, a.threadRootId ?? result.messageId); - if (direct.some((m) => a.venueId === (m.venueId ?? "") && a.threadRootId === (m.threadRootId ?? m.ts))) answered = true; + answeredConvos.add(convoKey(a.venueId, a.threadRootId)); // 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. - reactTo: async (v, ts, emoji) => { - const m = pending.find((p) => v === (p.venueId ?? "") && ts === p.ts); - const act = recordAct(this.d.db, this.d.clock, identityId, wakeId, { kind: "reacted", venueId: v, threadRootId: m?.threadRootId ?? null, ts, text: emoji }); + renderChecklist: async (items, seat) => streamFor(seat).setCards(items), + // Reactions reach any delivered message by venue + ts (the values in her lines), and + // carry the same bookkeeping a reply does — the §14.2 answered mark and the optimistic + // attention close — for the conversation the ref target itself names: a react on a tail + // line files into ITS thread, never a batch-derived one (audit 2026-08-13). + reactTo: async (v, ts, emoji, threadRootId) => { + // The act files at the target's OWN thread — null for a top-level line, whose acts + // render on the venue-surface conversation (filing at its ts would hide them there). + // The root key (thread it roots, for a top-level message) serves the §14.2 answered + // mark and the attention close, which speak in conversation keys. + const residence = threadRootId ?? ts; + const act = recordAct(this.d.db, this.d.clock, identityId, wakeId, { kind: "reacted", venueId: v, threadRootId, ts, text: emoji }); if (!act.inserted) return; // already reacted in an earlier attempt of this wake try { await this.d.adapter.addReaction(v, ts, emoji); @@ -723,9 +743,8 @@ export class Service { deleteAct(this.d.db, wakeId, act.actKey); // a failed call is not "already reacted" throw e; } - if (!m) return; - if (isDirectAddress(m)) answered = true; - closeAttentionItemsForThread(this.d.db, this.d.clock, identityId, v, m.threadRootId ?? m.ts ?? ts, "reacted in thread"); + answeredConvos.add(convoKey(v, residence)); + closeAttentionItemsForThread(this.d.db, this.d.clock, identityId, v, residence, "reacted in thread"); }, checklist, effects, @@ -837,29 +856,43 @@ export class Service { if (attempt < turns.maxRetries) await new Promise((r) => setTimeout(r, turns.backoffMs * 2 ** attempt)); } // §14.2's one carve-out: someone directly addressed her and the model died before it - // could answer. Honest, in the runtime's words when they read human. - if (status !== "succeeded" && direct.length > 0 && !answered) { - const last = direct.at(-1)!; + // could answer. Honest, in the runtime's words when they read human. One fallback per + // DISTINCT owed conversation, each anchored to its own coordinates — derived from the + // exact addressed events, never a batch-tail pick (audit 2026-08-13: `direct.at(-1)` + // apologized to one room and left the other asker hanging; any answer anywhere used to + // silence all of them). A conversation counts answered at either of a direct's two + // anchors: its thread, or — for a top-level mention/DM — the venue surface. + if (status !== "succeeded" && direct.length > 0) { + const owed = new Map(); + for (const m of direct) { + const anchor: Anchor = { venueId: m.venueId ?? "", threadRootId: m.threadRootId ?? m.ts }; + const k = convoKey(anchor.venueId, anchor.threadRootId); + if (!owed.has(k)) owed.set(k, { anchor, aliases: [k, ...(m.threadRootId ? [] : [convoKey(anchor.venueId, null)])] }); + } const why = failureCause || (status === "timed_out" ? "it ran out of time" : "my agent runtime failed"); const fallbackText = `can't run right now — ${why}. try me again, or flag the operator if it keeps up.`; - const fallbackAnchor = { venueId: last.venueId ?? "", threadRootId: last.threadRootId ?? last.ts }; - // The sole harness-authored words the room ever hears go through the same acts door - // as everything outward: idempotent across restarts of the same wake, visible in her - // own tail, never a post the ledger doesn't know about. - const fallbackAct = recordAct(this.d.db, this.d.clock, identityId, wakeId, { kind: "posted", venueId: fallbackAnchor.venueId, threadRootId: fallbackAnchor.threadRootId, ts: null, text: fallbackText }); - if (fallbackAct.inserted) { - await this.postMessage(fallbackAnchor, fallbackText) - .then((r) => (r.messageId === "undelivered" ? deleteAct(this.d.db, wakeId, fallbackAct.actKey) : setActTs(this.d.db, wakeId, fallbackAct.actKey, r.messageId))) - .catch(() => deleteAct(this.d.db, wakeId, fallbackAct.actKey)); + for (const { anchor, aliases } of owed.values()) { + if (aliases.some((k) => answeredConvos.has(k))) continue; + // The sole harness-authored words the room ever hears go through the same acts door + // as everything outward: idempotent across restarts of the same wake, visible in her + // own tail, never a post the ledger doesn't know about. + const fallbackAct = recordAct(this.d.db, this.d.clock, identityId, wakeId, { kind: "posted", venueId: anchor.venueId, threadRootId: anchor.threadRootId, ts: null, text: fallbackText }); + if (fallbackAct.inserted) { + await this.postMessage(anchor, fallbackText) + .then((r) => (r.messageId === "undelivered" ? deleteAct(this.d.db, wakeId, fallbackAct.actKey) : setActTs(this.d.db, wakeId, fallbackAct.actKey, r.messageId))) + .catch(() => deleteAct(this.d.db, wakeId, fallbackAct.actKey)); + } } } } 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(() => {}); + // Close every conversation's 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. + for (const s of streams.values()) { + if (status === "succeeded") s.settleCards(); + else s.clearCards(); + await s.close().catch(() => {}); + } // Delivery commits HERE, after the wake — done even when the turn failed (re-delivering // the same batch to a broken thread just loops the failure, observed live pre-collapse), // but never before it: a process death mid-wake leaves every watermark unadvanced and @@ -1031,6 +1064,10 @@ export class Service { anchor: null, nudgeAfterMs: 0, postMessage: async () => ({ messageId: "digest-probe" }), + // A live resident wake always has a ref table, and several tools shape their + // schema on its presence (checklist/task_confirm require ref) — the digest must + // describe the schemas she will actually see. + refs: makeRefTable(), effects: [], }), this.registries, diff --git a/src/turn-runner/execution-loop.ts b/src/turn-runner/execution-loop.ts index 47b9a0f..3c498a8 100644 --- a/src/turn-runner/execution-loop.ts +++ b/src/turn-runner/execution-loop.ts @@ -82,7 +82,7 @@ export async function runExecution(params: ExecutionLoopParams): Promise Promise; - // Shared holder for the execution's live checklist message id — persists across the execution's - // turns so the `checklist` tool edits ONE message in place (Claude Tag's signature UX). - checklist?: { messageId: string | null }; + // Shared holder for live checklist message ids, keyed by convoKey — persists across a turn's + // attempts (and an execution's turns) so the `checklist` tool edits ONE message in place per + // conversation (Claude Tag's signature UX). + checklist?: Map; // React to a message by venue + surface ts (Slack reactions.add) — sometimes an emoji IS the - // right reply ("if u see this please emoji it"). Venue-scoped like any post. - reactTo?: (venueId: string, messageId: string, emoji: string) => Promise; - // Render the execution's checklist as NATIVE task cards on its streamed message. Returns false - // when no stream is live (caller falls back to the emoji-text message). - renderChecklist?: (items: { text: string; done: boolean }[]) => Promise; + // right reply ("if u see this please emoji it"). threadRootId is the ref target's own thread + // (null for a top-level message): the react's ledger residence comes from the line the model + // was shown, never re-derived from the batch. Venue-scoped like any post. + reactTo?: (venueId: string, messageId: string, emoji: string, threadRootId: string | null) => Promise; + // Render a checklist as NATIVE task cards on the stream seated at `seat`. Returns false when + // the surface has no native cards (caller falls back to the emoji-text message). + renderChecklist?: (items: { text: string; done: boolean }[], seat: Anchor) => Promise; + // Resolve a principal id to its standing (operator/guest) — for durable writes whose person + // comes from a ref's provenance rather than the wake-level principal. + resolvePrincipal?: (principalId: string) => Principal; // Build a surface permalink for a message (SPEC §8.7: search hits carry receipts). Absent when // the surface can't construct one; hits then cite venue + timestamp only. permalink?: (venueId: string, messageId: string) => string | undefined; @@ -197,7 +203,6 @@ function taskCreateTool(ctx: ToolsetContext): ToolFactory { }, impl: async (args) => { const a = args as { title: string; spec: string; ref?: string; tier?: "low" | "medium" | "high" }; - if (!ctx.principal || !ctx.originEventId) return { success: false, output: "missing turn context for task_create" }; // The task's home is HER call, bound to a rendered conversation — never a batch-level // guess (live 2026-08-13: a task about an alert burst homed to the last thread that // happened to address her, and its report answered an adjacent incident). @@ -206,16 +211,27 @@ function taskCreateTool(ctx: ToolsetContext): ToolFactory { return { success: false, output: `"${a.ref ?? ""}" is not a ref — home the task with the [rN] tag of the conversation its report belongs in` }; } const home = conversationOf(target); + // Sponsor and origin bind to the ref's own provenance too: the same audit found the T-354 + // fix left both on the batch-level pick, producing tasks homed to one thread but sponsored + // by a speaker in another. A machine-authored line (worker report) has no speaker — the + // newest human IN THAT CONVERSATION stands sponsor, never a batch-level principal. + const prov = provenanceOfRef(ctx.db, ctx.identity.id, target); + if (!prov) { + return { success: false, output: "nothing recorded in that conversation yet — home the task with the [rN] tag of the message that asked for it" }; + } + const sponsorId = prov.principalId ?? lastSpeakerIn(ctx.db, ctx.identity.id, home); + if (!sponsorId) return { success: false, output: "can't tell who this task is for — use the [rN] tag of the asking message" }; + const sponsor = ctx.resolvePrincipal?.(sponsorId) ?? (ctx.principal?.id === sponsorId ? ctx.principal : undefined); const task = createTask(ctx.db, ctx.clock, { id: nextTaskId(ctx.db), identityId: ctx.identity.id, title: a.title, spec: a.spec, - sponsorId: ctx.principal.id, + sponsorId, homeAnchor: { venueId: home.venueId, threadRootId: home.threadRootId }, - originEventId: ctx.originEventId, + originEventId: prov.eventId, tier: a.tier, - sponsorIsOperator: ctx.principal.isOperator, + sponsorIsOperator: sponsor?.isOperator ?? false, }); pushEffect(ctx, { kind: "task_created", taskId: task.id }); return { success: true, output: JSON.stringify({ taskId: task.id, status: task.status }) }; @@ -223,28 +239,50 @@ function taskCreateTool(ctx: ToolsetContext): ToolFactory { }; } +// The durable source event a steer/cancel records: from the ref's provenance in ref-bearing +// turns (the message that asked for this — the same rung as every other durable write), else +// the turn's own origin event. A string return is the correctable bounce. +function steerSourceEvent(ctx: ToolsetContext, ref: string | undefined, asking: string): string | { bounce: string } { + if (ctx.refs) { + const target = ref ? ctx.refs.get(ref) : undefined; + if (!target) return { bounce: `"${ref ?? ""}" is not a ref — pass the [rN] tag of the message ${asking}` }; + const prov = provenanceOfRef(ctx.db, ctx.identity.id, target); + if (!prov) return { bounce: "nothing recorded in that conversation yet — point at the message itself" }; + return prov.eventId; + } + if (!ctx.originEventId) return { bounce: "missing turn context" }; + return ctx.originEventId; +} + function taskSteerTool(ctx: ToolsetContext): ToolFactory { + const withRef = !!ctx.refs; return { spec: { name: "task_steer", - description: "Attach guidance, a pause, or a resume to an existing task. Input: { taskId, kind: 'guidance'|'pause'|'resume', text? }.", + description: `Attach guidance, a pause, or a resume to an existing task. Input: { taskId, kind: 'guidance'|'pause'|'resume', text?${withRef ? ", ref" : ""} }.${withRef ? " ref is the [rN] tag of the message asking for this." : ""}`, inputSchema: { type: "object", additionalProperties: false, - required: ["taskId", "kind"], - properties: { taskId: { type: "string" }, kind: { type: "string", enum: ["guidance", "pause", "resume"] }, text: { type: "string" } }, + required: withRef ? ["taskId", "kind", "ref"] : ["taskId", "kind"], + properties: { + taskId: { type: "string" }, + kind: { type: "string", enum: ["guidance", "pause", "resume"] }, + text: { type: "string" }, + ...(withRef ? { ref: { type: "string", pattern: "^r\\d+$" } } : {}), + }, }, }, impl: async (args) => { - const a = args as { taskId: string; kind: SteeringKind; text?: string }; - if (!ctx.originEventId) return { success: false, output: "missing turn context for task_steer" }; + const a = args as { taskId: string; kind: SteeringKind; text?: string; ref?: string }; + const source = steerSourceEvent(ctx, a.ref, "asking for this steer"); + if (typeof source !== "string") return { success: false, output: source.bounce }; // "cancel"/"confirm" have their own dedicated tools (task_cancel/task_confirm) with their // own eligibility rules — task_steer's declared schema excludes them, and the JS-level call // must enforce that too, not just trust codex to validate against inputSchema. if (a.kind !== "guidance" && a.kind !== "pause" && a.kind !== "resume") { return { success: false, output: `invalid_kind: task_steer only accepts guidance/pause/resume; use task_cancel or task_confirm for ${a.kind}` }; } - const result = steerTask(ctx.db, ctx.clock, { identityId: ctx.identity.id, taskId: a.taskId, kind: a.kind, payload: { text: a.text }, sourceEventId: ctx.originEventId }); + const result = steerTask(ctx.db, ctx.clock, { identityId: ctx.identity.id, taskId: a.taskId, kind: a.kind, payload: { text: a.text }, sourceEventId: source }); pushEffect(ctx, { kind: "task_steered", taskId: a.taskId, steerKind: a.kind, applied: result.applied }); return { success: result.applied, output: result.reply ?? JSON.stringify({ status: result.task.status }) }; }, @@ -252,17 +290,27 @@ function taskSteerTool(ctx: ToolsetContext): ToolFactory { } function taskCancelTool(ctx: ToolsetContext): ToolFactory { + const withRef = !!ctx.refs; return { spec: { name: "task_cancel", - description: - "Cancel a task. The report is a ledger record — it is NOT posted to the thread. If the room should hear that the work stopped, say it yourself with reply. Input: { taskId, report? }.", - inputSchema: { type: "object", additionalProperties: false, required: ["taskId"], properties: { taskId: { type: "string" }, report: { type: "string" } } }, + description: `Cancel a task. The report is a ledger record — it is NOT posted to the thread. If the room should hear that the work stopped, say it yourself with reply. Input: { taskId, report?${withRef ? ", ref" : ""} }.${withRef ? " ref is the [rN] tag of the message asking for the cancel." : ""}`, + inputSchema: { + type: "object", + additionalProperties: false, + required: withRef ? ["taskId", "ref"] : ["taskId"], + properties: { + taskId: { type: "string" }, + report: { type: "string" }, + ...(withRef ? { ref: { type: "string", pattern: "^r\\d+$" } } : {}), + }, + }, }, impl: async (args) => { - const a = args as { taskId: string; report?: string }; - if (!ctx.originEventId) return { success: false, output: "missing turn context for task_cancel" }; - const result = steerTask(ctx.db, ctx.clock, { identityId: ctx.identity.id, taskId: a.taskId, kind: "cancel", payload: { report: a.report }, sourceEventId: ctx.originEventId }); + const a = args as { taskId: string; report?: string; ref?: string }; + const source = steerSourceEvent(ctx, a.ref, "asking for the cancel"); + if (typeof source !== "string") return { success: false, output: source.bounce }; + const result = steerTask(ctx.db, ctx.clock, { identityId: ctx.identity.id, taskId: a.taskId, kind: "cancel", payload: { report: a.report }, sourceEventId: source }); pushEffect(ctx, { kind: "task_cancelled", taskId: a.taskId, applied: result.applied }); return { success: result.applied, output: result.reply ?? JSON.stringify({ status: result.task.status }) }; }, @@ -270,16 +318,53 @@ function taskCancelTool(ctx: ToolsetContext): ToolFactory { } function taskConfirmTool(ctx: ToolsetContext): ToolFactory { + // With a ref table (resident wakes), the approver is the SPEAKER of the ref'd message — the + // durable resolution records who actually said yes/no, never a wake-level principal pick. + // Ref-less contexts (no rendered lines to point at) keep the turn principal. + const withRef = !!ctx.refs; return { spec: { name: "task_confirm", - description: "Resolve a pending confirmation on a task from a member's approve/deny. Input: { taskId, approve }.", - inputSchema: { type: "object", additionalProperties: false, required: ["taskId", "approve"], properties: { taskId: { type: "string" }, approve: { type: "boolean" } } }, + description: withRef + ? "Resolve a pending confirmation on a task from a member's approve/deny. Input: { taskId, approve, ref } — ref is the [rN] tag of the message where they granted or denied it; their word is the authority, so point at it." + : "Resolve a pending confirmation on a task from a member's approve/deny. Input: { taskId, approve }.", + inputSchema: { + type: "object", + additionalProperties: false, + required: withRef ? ["taskId", "approve", "ref"] : ["taskId", "approve"], + properties: { + taskId: { type: "string" }, + approve: { type: "boolean" }, + ...(withRef ? { ref: { type: "string", pattern: "^r\\d+$" } } : {}), + }, + }, }, impl: async (args) => { - const a = args as { taskId: string; approve: boolean }; - if (!ctx.principal) return { success: false, output: "missing principal for task_confirm" }; - const result = resolveConfirmation(ctx.db, ctx.clock, { identityId: ctx.identity.id, taskId: a.taskId, principalId: ctx.principal.id, approve: a.approve }); + const a = args as { taskId: string; approve: boolean; ref?: string }; + let approverId: string; + if (withRef) { + const target = a.ref ? ctx.refs?.get(a.ref) : undefined; + // A go-ahead belongs to the person who SAID it: only a message ref names a speaker. A + // conversation ref would resolve to whoever spoke last in the room — the exact + // batch-tail guess this tool exists to prevent (audit 2026-08-13, verified live-shape). + if (!target?.ts) { + return { success: false, output: `"${a.ref ?? ""}" is not a message ref — pass the [rN] tag of the member's own approve/deny line, not the conversation's` }; + } + // Unread targets are rejected outright (no one-shot bounce like reply's): recording who + // authorized a consequential action from a line this turn never read is never right. + if (target.via === "search") { + return { success: false, output: "that line isn't from this conversation as you just read it — point at the [rN] tag of the approve/deny message in the rendered card" }; + } + const prov = provenanceOfRef(ctx.db, ctx.identity.id, target); + if (!prov?.principalId) { + return { success: false, output: "that line has no speaker to attribute the decision to — use the [rN] tag of the member's own message" }; + } + approverId = prov.principalId; + } else { + if (!ctx.principal) return { success: false, output: "missing principal for task_confirm" }; + approverId = ctx.principal.id; + } + const result = resolveConfirmation(ctx.db, ctx.clock, { identityId: ctx.identity.id, taskId: a.taskId, principalId: approverId, approve: a.approve }); pushEffect(ctx, { kind: "confirmation_resolved", taskId: a.taskId, approve: a.approve, applied: result.applied }); return { success: result.applied, output: result.reply ?? JSON.stringify({ status: result.task.status }) }; }, @@ -397,7 +482,10 @@ function reactTool(ctx: ToolsetContext): ToolFactory { const violation = checkPostingScope(ctx, { venueId: target.venueId, threadRootId: null }); if (violation) return { success: false, output: `posting_scope_violation: ${violation}` }; try { - await ctx.reactTo(target.venueId, target.ts, emoji); + // The target's own thread rides along: the react's ledger residence is the line she was + // shown, never re-derived from the wake's batch (audit 2026-08-13: a react on a tail + // line filed at the surface and rendered in the wrong conversation on later wakes). + await ctx.reactTo(target.venueId, target.ts, emoji, target.threadRootId); } catch (e) { return { success: false, output: `reaction failed: ${e instanceof Error ? e.message : String(e)}` }; } @@ -520,44 +608,64 @@ function renderChecklist(items: { text: string; done: boolean }[]): string { return items.map((i) => `${i.done ? "✅" : "⬜️"} ${i.text}`).join("\n"); } function checklistTool(ctx: ToolsetContext): ToolFactory { + // Resident wakes seat the checklist by ref — the model says which conversation the work is + // for, same rung as reply/react/task_create (audit 2026-08-13: this was the one posting tool + // whose destination was still the harness's batch-level guess). Ref-less contexts (an + // execution) seat on their anchor: a task's home is already ref-bound at creation. + const withRef = !!ctx.refs; return { spec: { name: "checklist", description: - "Post/update a live progress checklist for this piece of work — it edits ONE message in place. Most replies don't need one: reach for it only when the work is genuinely long and multi-step, with 2-4 high-level goals (what you're finding out, not which tools you'll run). Call it FIRST with the stages (all done:false), then flip each done as you finish. Input: { items: [{ text, done }] }.", + `Post/update a live progress checklist for this piece of work — it edits ONE message in place${withRef ? ", in the conversation whose [rN] ref you pass" : ""}. Most replies don't need one: reach for it only when the work is genuinely long and multi-step, with 2-4 high-level goals (what you're finding out, not which tools you'll run). Call it FIRST with the stages (all done:false), then flip each done as you finish. Input: { items: [{ text, done }]${withRef ? ", ref" : ""} }.${withRef ? " It renders alongside your reply there — a checklist without any words in that conversation shows nothing." : ""}`, inputSchema: { type: "object", additionalProperties: false, - required: ["items"], + required: withRef ? ["items", "ref"] : ["items"], properties: { items: { type: "array", items: { type: "object", additionalProperties: false, required: ["text", "done"], properties: { text: { type: "string" }, done: { type: "boolean" } } }, }, + ...(withRef ? { ref: { type: "string", pattern: "^r\\d+$" } } : {}), }, }, }, impl: async (args) => { - const a = args as { items: { text: string; done: boolean }[] }; - if (!ctx.anchor) return { success: false, output: "no anchor for this turn" }; - const ref = ctx.checklist; - if (!ref) return { success: false, output: "checklist is not available in this turn" }; - // Preferred rendering: native task cards on the execution's streamed message (the harness - // provides renderChecklist when a stream is live). Falls back to one edited-in-place emoji - // message only when no stream exists (e.g. a recovered task with no thread to stream into). - const native = ctx.renderChecklist ? await ctx.renderChecklist(a.items) : false; + const a = args as { items: { text: string; done: boolean }[]; ref?: string }; + let seat: Anchor; + if (withRef) { + const target = a.ref ? ctx.refs?.get(a.ref) : undefined; + if (!target) { + return { success: false, output: `"${a.ref ?? ""}" is not a ref — seat the checklist with the [rN] tag of the conversation its work is for` }; + } + const key = conversationOf(target); + seat = { venueId: key.venueId, threadRootId: key.threadRootId }; + } else { + if (!ctx.anchor) return { success: false, output: "no anchor for this turn" }; + seat = ctx.anchor; + } + const violation = checkPostingScope(ctx, seat); + if (violation) return { success: false, output: `posting_scope_violation: ${violation}` }; + const holder = ctx.checklist; + if (!holder) return { success: false, output: "checklist is not available in this turn" }; + // Preferred rendering: native task cards on the seat conversation's streamed message. + // Falls back to one edited-in-place emoji message only when the surface has no cards. + const native = ctx.renderChecklist ? await ctx.renderChecklist(a.items, seat) : false; if (!native) { const text = renderChecklist(a.items); - if (ref.messageId && ctx.updateMessage) { - await ctx.updateMessage(ctx.anchor.venueId, ref.messageId, text); + const seatKey = convoKey(seat.venueId, seat.threadRootId); + const existing = holder.get(seatKey); + if (existing && ctx.updateMessage) { + await ctx.updateMessage(seat.venueId, existing, text); } else { - const result = await ctx.postMessage(ctx.anchor, text); // first call, or no edit support → (re)post + const result = await ctx.postMessage(seat, text); // first call, or no edit support → (re)post // A delivery sentinel is not a message id — latching it would aim every later edit at // the literal string "undelivered" (review finding, 2026-08-11). if (result.messageId === "undelivered" || result.messageId === "already-sent-this-wake") { return { success: false, output: "the checklist message didn't land — try again" }; } - ref.messageId = result.messageId; + holder.set(seatKey, result.messageId); } } pushEffect(ctx, { kind: "checklist", items: a.items.length, done: a.items.filter((i) => i.done).length }); diff --git a/test/ear.test.ts b/test/ear.test.ts index 9eaaccf..41dfba2 100644 --- a/test/ear.test.ts +++ b/test/ear.test.ts @@ -82,11 +82,35 @@ function msg(overrides: Partial = {}): RawMessage { } describe("the ear gates waking, never delivery", () => { + // Audit 2026-08-13: a refless hold used to return "noted" while recording NOTHING — the + // 2026-08-10 discarded-judgment failure wearing a polite face. hold/wake bounce without a + // ref; the re-issue with one lands durably (its why rides the next delivery). + test("a refless hold/wake bounces with a correctable error — judgment is never silently dropped", async () => { + const verdictResults: { success: boolean; output: string }[] = []; + const { db, adapter, service } = harness(async (_turn, tools, _act, prompt) => { + const verdict = tools.get("verdict"); + if (!verdict) return; // the mind: nothing needed + verdictResults.push(await verdict.run({ decision: "hold", why: "teammates have it" })); + verdictResults.push(await verdict.run({ decision: "hold", why: "teammates have it", ref: refIn(prompt, "lunch") })); + }); + await service.start(); + adapter.emit(msg({ text: "who's in for lunch", ts: "3.1" })); + await service.idle(); + + expect(verdictResults[0]!.success).toBe(false); + expect(verdictResults[0]!.output).toContain("needs ref"); + expect(verdictResults[1]!.success).toBe(true); + // The recorded hold is durable judgment on the conversation row, not a discarded verdict. + const row = db.query("SELECT holds FROM conversations WHERE venue_id = 'C1'").get() as { holds: number } | null; + expect(row?.holds).toBe(1); + await service.stop(); + }); + test("a hold verdict wakes nobody, posts nothing — and the held lines ride the NEXT wake verbatim", async () => { - const { adapter, service, earSessions, mindSessions } = harness(async (_turn, tools) => { + const { adapter, service, earSessions, mindSessions } = harness(async (_turn, tools, _act, prompt) => { const verdict = tools.get("verdict"); if (verdict) { - await verdict.run({ decision: "hold", why: "teammates comparing lunch orders" }); + await verdict.run({ decision: "hold", why: "teammates comparing lunch orders", ref: refIn(prompt, /<#C1>/) }); return; } // the mind: no action needed for this row @@ -307,7 +331,7 @@ describe("thread-follow is the ear's to judge (SPEC §11)", () => { earCalls++; // pass 2 sees the teammates' aside: hold. pass 3 sees the reply that is plainly hers: wake. if (earCalls === 3) await verdict.run({ decision: "wake", why: "kate is asking her to go ahead", ref: refIn(prompt, "go ahead") }); - else await verdict.run({ decision: "hold", why: "teammates talking to each other" }); + else await verdict.run({ decision: "hold", why: "teammates talking to each other", ref: refIn(prompt, /<#C1>/) }); return; } mindCalls++; @@ -341,7 +365,7 @@ describe("thread-follow is the ear's to judge (SPEC §11)", () => { if (verdict) { earCalls++; if (earCalls === 2) await verdict.run({ decision: "wake", why: "this thread needs her", ref: refIn(prompt, /<#C1>/) }); - else await verdict.run({ decision: "hold", why: "nothing yet" }); + else await verdict.run({ decision: "hold", why: "nothing yet", ref: refIn(prompt, /<#C1>/) }); return; } throw new Error("mind runtime exploded"); @@ -370,7 +394,7 @@ describe("step_back (standing engagement state)", () => { earCalls++; // pass 2 carries the "stop" reply: plainly hers, wake her for it if (earCalls === 2) await verdict.run({ decision: "wake", why: "they are telling her to stop", ref: refIn(prompt, /<#C1>/) }); - else await verdict.run({ decision: "hold", why: "the humans have this one" }); + else await verdict.run({ decision: "hold", why: "the humans have this one", ref: refIn(prompt, /<#C1>/) }); return; } mindCalls++; @@ -409,7 +433,7 @@ describe("step_back (standing engagement state)", () => { await verdict.run({ decision: "open_ask", why: "kate asked her to weigh in", ref: refIn(prompt, "weigh in") }); await verdict.run({ decision: "wake", why: "kate asked her to weigh in", ref: refIn(prompt, "weigh in") }); } else { - await verdict.run({ decision: "hold", why: "nothing new" }); + await verdict.run({ decision: "hold", why: "nothing new", ref: refIn(prompt, /<#C1>/) }); } return; } @@ -425,10 +449,10 @@ describe("step_back (standing engagement state)", () => { describe("what the prompts carry", () => { test("the mind's prompt marks direct addresses [to you]; ride-along chatter is unmarked", async () => { - const h = harness(async (_turn, tools) => { + const h = harness(async (_turn, tools, _act, prompt) => { const verdict = tools.get("verdict"); if (verdict) { - await verdict.run({ decision: "hold", why: "just chatter" }); + await verdict.run({ decision: "hold", why: "just chatter", ref: refIn(prompt, /<#C1>/) }); return; } }); @@ -447,9 +471,9 @@ describe("what the prompts carry", () => { // The ear design's "plus the live threads that delta touches". Live 2026-07-30: a pass // whose whole batch was one mid-thread line ("LMK if you wanna get in on browserstack") // had no way to see the offer was aimed at a teammate, and recorded the ask as hers. - const h = harness(async (_turn, tools) => { + const h = harness(async (_turn, tools, _act, prompt) => { const verdict = tools.get("verdict"); - if (verdict) await verdict.run({ decision: "hold", why: "teammates talking to each other" }); + if (verdict) await verdict.run({ decision: "hold", why: "teammates talking to each other", ref: refIn(prompt, /<#C1>/) }); }); await h.service.start(); h.adapter.emit(msg({ text: "Ready for QA: the safari fix", ts: "80.0", principalId: "U_PEDRO", principalName: "pedro" })); @@ -468,9 +492,9 @@ describe("what the prompts carry", () => { }); test("the ear knows which id is hers — the standing doc names her principal", async () => { - const h = harness(async (_turn, tools) => { + const h = harness(async (_turn, tools, _act, prompt) => { const verdict = tools.get("verdict"); - if (verdict) await verdict.run({ decision: "hold", why: "nothing needed" }); + if (verdict) await verdict.run({ decision: "hold", why: "nothing needed", ref: refIn(prompt, /<#C1>/) }); }); await h.service.start(); h.adapter.emit(msg({ text: "chatter", ts: "81.1" })); @@ -503,10 +527,10 @@ describe("what the prompts carry", () => { describe("delivery invariants hold under the ear", () => { test("nothing dangles: after any mix of held and promoted traffic, the inbox drains to empty on the next wake", async () => { - const { db, service, adapter } = harness(async (_turn, tools) => { + const { db, service, adapter } = harness(async (_turn, tools, _act, prompt) => { const verdict = tools.get("verdict"); if (verdict) { - await verdict.run({ decision: "hold", why: "just chatter" }); + await verdict.run({ decision: "hold", why: "just chatter", ref: refIn(prompt, /<#C1>/) }); return; } }); diff --git a/test/resident.test.ts b/test/resident.test.ts index f5ce954..d3ddf62 100644 --- a/test/resident.test.ts +++ b/test/resident.test.ts @@ -417,9 +417,13 @@ describe("resident delivery", () => { await service.idle(); // flushes the boot wake carrying both conversations expect(rejected[0]).toContain("is not a ref"); - expect(adapter.posts).toHaveLength(1); - expect(adapter.posts[0]!.venueId).toBe("C1"); // where the answer belongs... - expect(adapter.posts[0]!.threadRootTs).toBe("1.0"); // ...in ITS thread, not the batch's last + // The reply rides a native stream seated in ITS OWN conversation — streams are created + // per conversation at her first ref-addressed post, not pre-seated on a batch-tail guess. + expect(adapter.posts).toHaveLength(0); + expect(adapter.streams).toHaveLength(1); + expect(adapter.streams[0]!.venueId).toBe("C1"); // where the answer belongs... + expect(adapter.streams[0]!.threadTs).toBe("1.0"); // ...in ITS thread, not the batch's last + expect(adapter.streams[0]!.text).toBe("the export fix landed"); await service.stop(); }); @@ -434,6 +438,7 @@ describe("resident delivery", () => { ); seed.run("e1", "k1", "C1", "1.0", JSON.stringify({ text: "<@BOT1> alert burst, investigate", ts: "1.1", addressMode: "mention" })); seed.run("e2", "k2", "C2", null, JSON.stringify({ text: "<@BOT1> pull it together blacksmith", ts: "2.0", addressMode: "mention" })); + db.query("UPDATE events SET principal_id = 'U2' WHERE id = 'e2'").run(); // a different asker tails the batch const rejected: string[] = []; const { service } = harness(async (_turn, tools, _mark, prompt) => { @@ -448,9 +453,156 @@ describe("resident delivery", () => { await service.idle(); // flushes the boot wake carrying both conversations expect(rejected[0]).toContain("is not a ref"); - const row = db.query("SELECT home_venue_id, home_thread_root_id FROM tasks").get() as { home_venue_id: string; home_thread_root_id: string | null } | null; + const row = db + .query("SELECT home_venue_id, home_thread_root_id, sponsor_id, origin_event_id FROM tasks") + .get() as { home_venue_id: string; home_thread_root_id: string | null; sponsor_id: string; origin_event_id: string } | null; expect(row?.home_venue_id).toBe("C1"); // the incident's thread... expect(row?.home_thread_root_id).toBe("1.0"); // ...not C2, the batch's last-addressed guess + // Provenance binds to the ref too: sponsor and origin are the ref'd message's speaker and + // event — not U2/e2, the batch-tail pick that survived the first T-354 fix in these columns. + expect(row?.sponsor_id).toBe("U1"); + expect(row?.origin_event_id).toBe("e1"); + await service.stop(); + }); + + // Audit 2026-08-13, §14.2 batch-granularity: `direct.at(-1)` used to apologize to ONE + // conversation when several addressed her, and one wake-scoped answered boolean let any + // answer anywhere silence every other owed room. The fallback is per owed conversation. + test("§14.2: a dead wake owing two conversations apologizes in each; an answered one is skipped", async () => { + const db = openLedger(":memory:"); + const seed = db.query( + `INSERT INTO events (id, dedup_key, kind, identity_id, venue_id, thread_root_id, principal_id, payload, received_at) + VALUES (?, ?, 'addressed_message', 'eng', ?, ?, 'U1', ?, '2026-07-01T00:00:00Z')`, + ); + seed.run("e1", "k1", "C1", "1.0", JSON.stringify({ text: "<@BOT1> what broke?", ts: "1.1", addressMode: "mention" })); + seed.run("e2", "k2", "C2", "2.0", JSON.stringify({ text: "<@BOT1> status?", ts: "2.1", addressMode: "mention" })); + + const { adapter, service } = harness(async (_turn, tools, _mark, prompt) => { + if (!tools.get("reply")) return; // the ear + // She answers C1, then the runtime dies before C2 — C2 alone is owed the fallback. + await tools.get("reply")!.run({ text: "looking", ref: refIn(prompt, "what broke?") }); + throw new Error("runtime died mid-wake"); + }, db); + await service.start(); + await service.idle(); + + const fallbacks = adapter.posts.filter((p) => p.text.includes("can't run right now")); + expect(fallbacks).toHaveLength(1); + expect(fallbacks[0]!.venueId).toBe("C2"); // the unanswered asker... + expect(fallbacks[0]!.threadRootTs).toBe("2.0"); // ...in their own thread + await service.stop(); + }); + + test("§14.2: a dead wake that answered nobody apologizes once per owed conversation, each in its own thread", async () => { + const db = openLedger(":memory:"); + const seed = db.query( + `INSERT INTO events (id, dedup_key, kind, identity_id, venue_id, thread_root_id, principal_id, payload, received_at) + VALUES (?, ?, 'addressed_message', 'eng', ?, ?, 'U1', ?, '2026-07-01T00:00:00Z')`, + ); + seed.run("e1", "k1", "C1", "1.0", JSON.stringify({ text: "<@BOT1> what broke?", ts: "1.1", addressMode: "mention" })); + seed.run("e2", "k2", "C2", "2.0", JSON.stringify({ text: "<@BOT1> status?", ts: "2.1", addressMode: "mention" })); + + const { adapter, service } = harness(async (_turn, tools) => { + if (!tools.get("reply")) return; // the ear + throw new Error("runtime died before any answer"); + }, db); + await service.start(); + await service.idle(); + + const fallbacks = adapter.posts.filter((p) => p.text.includes("can't run right now")); + const where = fallbacks.map((p) => `${p.venueId}:${p.threadRootTs}`).sort(); + expect(where).toEqual(["C1:1.0", "C2:2.0"]); // one per owed conversation — nobody left hanging + await service.stop(); + }); + + // Review 2026-08-13: the wake stopped passing originEventId and task_steer/task_cancel died + // for EVERY live resident turn while the whole suite stayed green — the toolset tests + // hand-built their context. These run through Service.runWake()'s own toolset, so the wiring + // itself is what's under test. Steers bind their source event to the ref's provenance. + test("task_steer and task_cancel work through a real wake, sourced from the asking message's ref", async () => { + const { db, adapter, service } = harness(async (_turn, tools, _mark, prompt) => { + if (!tools.get("reply")) return; // the ear + const taskCreate = tools.get("task_create"); + if (!taskCreate) return; + if (!prompt.includes("check canary too")) { + await taskCreate.run({ title: "watch", spec: "watch it", ref: refIn(prompt, "watch the deploy") }); + return; + } + const steerRef = refIn(prompt, "check canary too"); + const steered = await tools.get("task_steer")!.run({ taskId: "T-1", kind: "guidance", text: "check canary too", ref: steerRef }); + expect(steered.success).toBe(true); + const cancelled = await tools.get("task_cancel")!.run({ taskId: "T-1", report: "asked to stop", ref: steerRef }); + expect(cancelled.success).toBe(true); + }); + await service.start(); + adapter.emit(msg({ text: "<@BOT1> watch the deploy", mentionsBotId: true, ts: "90.1", threadRootTs: "90.0" })); + await service.idle(); + adapter.emit(msg({ text: "<@BOT1> check canary too, actually just stop", mentionsBotId: true, ts: "90.2", threadRootTs: "90.0", principalId: "U3" })); + await service.idle(); + + const task = db.query("SELECT status FROM tasks WHERE id = 'T-1'").get() as { status: string } | null; + expect(task?.status).toBe("cancelled"); + const steer = db.query("SELECT source_event_id FROM steering WHERE kind = 'guidance'").get() as { source_event_id: string } | null; + const askEvent = db.query("SELECT id FROM events WHERE json_extract(payload, '$.ts') = '90.2'").get() as { id: string } | null; + expect(steer?.source_event_id).toBe(askEvent!.id); // provenance = the message that asked + await service.stop(); + }); + + // Audit 2026-08-13: a react's ledger residence used to be re-derived from the wake's pending + // batch — a react on a TAIL line (delivered in an earlier wake) filed at the surface and + // rendered in the wrong conversation later. Residence comes from the ref target itself. + test("a react on a tail line files its act into that line's thread, not the surface", async () => { + let wakes = 0; + const { db, adapter, service } = harness(async (_turn, tools, _mark, prompt) => { + if (!tools.get("reply")) return; // the ear + wakes++; + if (wakes === 1) return; // first wake delivers the root ask; she holds her tongue + await tools.get("react")!.run({ emoji: "eyes", ref: refIn(prompt, "root ask") }); // the TAIL line + }); + await service.start(); + adapter.emit(msg({ text: "<@BOT1> root ask", mentionsBotId: true, ts: "77.1", threadRootTs: "77.0" })); + await service.idle(); + adapter.emit(msg({ text: "<@BOT1> did you see it?", mentionsBotId: true, ts: "77.9", threadRootTs: "77.0" })); + await service.idle(); + + const act = db.query("SELECT venue_id, thread_root_id, ts FROM acts WHERE kind = 'reacted'").get() as { venue_id: string; thread_root_id: string | null; ts: string } | null; + expect(act?.ts).toBe("77.1"); // the tail line she reacted to... + expect(act?.thread_root_id).toBe("77.0"); // ...filed in ITS thread — never the surface + expect(adapter.reactions.at(-1)).toMatchObject({ venueId: "C1", messageId: "77.1", emoji: "eyes" }); + await service.stop(); + }); + + // Audit 2026-08-13: checklist was the one posting tool with no ref — its cards could only + // land on the wake's guessed home. Now the model seats it, and each conversation she speaks + // into gets its own native stream: cards ride the seat's stream, not the batch tail's. + test("a checklist seats on its ref'd conversation's stream in a two-conversation wake", async () => { + const db = openLedger(":memory:"); + const seed = db.query( + `INSERT INTO events (id, dedup_key, kind, identity_id, venue_id, thread_root_id, principal_id, payload, received_at) + VALUES (?, ?, 'addressed_message', 'eng', ?, ?, 'U1', ?, '2026-07-01T00:00:00Z')`, + ); + seed.run("e1", "k1", "C1", "1.0", JSON.stringify({ text: "<@BOT1> quick one", ts: "1.1", addressMode: "mention" })); + seed.run("e2", "k2", "C2", "2.0", JSON.stringify({ text: "<@BOT1> the long migration", ts: "2.1", addressMode: "mention" })); + + const { adapter, service } = harness(async (_turn, tools, _mark, prompt) => { + if (!tools.get("reply")) return; // the ear + const longRef = refIn(prompt, "long migration"); + await tools.get("reply")!.run({ text: "62 done", ref: refIn(prompt, "quick one") }); + await tools.get("checklist")!.run({ items: [{ text: "migrate tables", done: false }], ref: longRef }); + await tools.get("reply")!.run({ text: "starting the migration", ref: longRef }); + }, db); + await service.start(); + await service.idle(); + + // Each conversation streams its own reply — no plain posts, no shared seat. + expect(adapter.posts).toHaveLength(0); + expect(adapter.streams).toHaveLength(2); + const byVenue = new Map(adapter.streams.map((s) => [s.venueId, s])); + expect(byVenue.get("C1")?.text).toBe("62 done"); + expect(byVenue.get("C2")?.text).toBe("starting the migration"); + // The cards ride the C2 stream — the conversation SHE said the work is for. + const cardMessages = new Set(adapter.taskCards.map((c) => c.messageId)); + expect(cardMessages).toEqual(new Set([byVenue.get("C2")!.messageId])); await service.stop(); }); @@ -461,9 +613,12 @@ describe("resident delivery", () => { test("checklist cards buffer until the reply materializes the stream — a plan box alone never posts", async () => { const { adapter, service } = harness(async (_turn, tools, _mark, prompt) => { 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", ref: refIn(prompt, "organize") }); - await tools.get("checklist")!.run({ items: [{ text: "collect reports", done: true }, { text: "send the list", done: false }] }); + // The checklist seats by ref like every posting tool — the model says which conversation + // the work is for; the cards ride that conversation's stream. + const ref = refIn(prompt, "organize"); + await tools.get("checklist")!.run({ items: [{ text: "collect reports", done: false }, { text: "send the list", done: false }], ref }); + await tools.get("reply")!.run({ text: "3 follow-ups, list below", ref }); + await tools.get("checklist")!.run({ items: [{ text: "collect reports", done: true }, { text: "send the list", done: false }], ref }); }); await service.start(); adapter.emit(msg({ text: "<@BOT1> organize today's reports", mentionsBotId: true, ts: "5.0" })); @@ -484,14 +639,16 @@ describe("resident delivery", () => { }); 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) => { + const outcomes: { success: boolean }[] = []; + const { adapter, service } = harness(async (_turn, tools, _mark, prompt) => { if (tools.get("verdict")) return; - await tools.get("checklist")!.run({ items: [{ text: "a plan with no words", done: false }] }); + outcomes.push(await tools.get("checklist")!.run({ items: [{ text: "a plan with no words", done: false }], ref: refIn(prompt, "hm") })); }); await service.start(); - adapter.emit(msg({ text: "<@BOT1> hm", mentionsBotId: true, ts: "6.0" })); + adapter.emit(msg({ text: "<@BOT1> hm", mentionsBotId: true, ts: "6.0", threadRootTs: "6.0" })); await service.idle(); + expect(outcomes[0]!.success).toBe(true); // the call RAN — this test must never pass at the ref gate expect(adapter.posts).toHaveLength(0); expect(adapter.streams).toHaveLength(0); expect(adapter.taskCards).toHaveLength(0); diff --git a/test/toolset.test.ts b/test/toolset.test.ts index 49856d7..bcf61f5 100644 --- a/test/toolset.test.ts +++ b/test/toolset.test.ts @@ -40,8 +40,10 @@ function identity(overrides: Partial = {}): IdentityConfig { function baseCtx(db: ReturnType, clock: Clock, overrides: Partial = {}): ToolsetContext { const posts: { anchor: any; text: string }[] = []; // A standing rendered ref for the wake's home conversation — what task_create homes to. + // Minted the way the renderer does: carrying the provenance (event + speaker) of the line, + // which is where durable writes (sponsor/origin, confirmation approver) now bind. const refs = makeRefTable(); - refs.mint({ venueId: "C1", threadRootId: null, via: "rendered" }); // r1 + refs.mint({ venueId: "C1", threadRootId: null, via: "rendered", eventId: "e1", principalId: "U1" }); // r1 return { refs, db, @@ -128,7 +130,7 @@ describe("task_steer / task_cancel / task_confirm", () => { const steerCtx = { ...ctx, originEventId: "e2" }; const tools = buildToolset(steerCtx); - const result = await tool(tools, "task_steer").run({ taskId: "T-1", kind: "guidance", text: "check redis too" }); + const result = await tool(tools, "task_steer").run({ taskId: "T-1", kind: "guidance", text: "check redis too", ref: "r1" }); expect(result.success).toBe(true); expect(getTask(db, "T-1")?.spec).toContain("check redis too"); }); @@ -142,7 +144,7 @@ describe("task_steer / task_cancel / task_confirm", () => { const steerCtx = { ...ctx, originEventId: "e2" }; const tools = buildToolset(steerCtx); - const result = await tool(tools, "task_steer").run({ taskId: "T-1", kind: "cancel" }); + const result = await tool(tools, "task_steer").run({ taskId: "T-1", kind: "cancel", ref: "r1" }); expect(result.success).toBe(false); expect(result.output).toContain("invalid_kind"); expect(getTask(db, "T-1")?.status).toBe("active"); // unaffected @@ -155,7 +157,7 @@ describe("task_steer / task_cancel / task_confirm", () => { await activeTask(db, clock, ctx); seedEvent(db, "e2", clock); const cancelCtx = { ...ctx, originEventId: "e2", effects: [] as unknown[] }; - const result = await tool(buildToolset(cancelCtx), "task_cancel").run({ taskId: "T-1", report: "member asked to stop" }); + const result = await tool(buildToolset(cancelCtx), "task_cancel").run({ taskId: "T-1", report: "member asked to stop", ref: "r1" }); expect(result.success).toBe(true); expect(getTask(db, "T-1")?.status).toBe("cancelled"); @@ -173,9 +175,17 @@ describe("task_steer / task_cancel / task_confirm", () => { requestConfirmation(db, clock, { taskId: "T-1", actionRef: "send_email:x", description: "send it?", nudgeDeadline: "2026-07-03T00:00:00Z" }); const confirmCtx = baseCtx(db, clock, { principal: { id: "U2", isGuest: false, isOperator: false } }); - const result = await tool(buildToolset(confirmCtx), "task_confirm").run({ taskId: "T-1", approve: true }); + // The approver is the SPEAKER of the ref'd approval message — recorded from the ref's + // provenance, never from the wake-level principal (audit 2026-08-13). + seedEvent(db, "e9", clock); + const approvalRef = confirmCtx.refs!.mint({ venueId: "C1", threadRootId: null, ts: "9.9", via: "rendered", eventId: "e9", principalId: "U2" }); + const bare = await tool(buildToolset(confirmCtx), "task_confirm").run({ taskId: "T-1", approve: true }); + expect(bare.success).toBe(false); // a refless confirm has no speaker to attribute + expect(bare.output).toContain("is not a message ref"); + const result = await tool(buildToolset(confirmCtx), "task_confirm").run({ taskId: "T-1", approve: true, ref: approvalRef }); expect(result.success).toBe(true); expect(getTask(db, "T-1")?.status).toBe("open"); + expect(getTask(db, "T-1")?.pendingConfirmation?.resolution?.principalId).toBe("U2"); }); test("task_confirm is denied outright for a guest principal, before ever touching the ledger", async () => {