From 64b561aa286921451280c9d1f82e58c153bd5cf3 Mon Sep 17 00:00:00 2001 From: yiliang114 <1204183885@qq.com> Date: Mon, 7 Sep 2026 11:07:51 +0900 Subject: [PATCH 1/2] feat(mesh): record run closes and derive the thread state from them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closing is two writes because a closing tool is called mid-turn and cannot mark its own still-executing runtime finished. The tool records what the run is closing as and moves it to finishing, which ends the turn; the runtime callback records the terminal state, and only there is the thread's status recomputed. A crash between the two leaves a finishing run with a closeKind, which is a complete instruction for restart reconciliation — a status written before the runtime actually stopped would be a lie the next reader cannot detect. A wait is refused when nothing could wake it, and the dependency is walked over parentThreadId rather than rootThreadId so a sibling sub-thread does not count as this thread's delegation. Any close discharges peers' waits on the same thread. A clean exit that never called a closing tool is recorded as unclosed rather than as implicit success. finishRun now delegates here so a run has exactly one way to end, and postMessage discharges outstanding obligations when it books work before applying the aggregate status. Without those two producers the resolver's rules had no writer: an obsolete failure kept the thread blocked after later work succeeded, and a post that booked nothing left it in in_progress with no live run and no explanation. --- ...6-09-06-multi-agent-board-collaboration.md | 17 +- ...26-09-07-mesh-implementation-acceptance.md | 2 + .../src/agents/mesh/run-lifecycle.test.ts | 344 ++++++++++++++ .../core/src/agents/mesh/run-lifecycle.ts | 449 ++++++++++++++++++ .../core/src/agents/mesh/thread-actions.ts | 55 ++- 5 files changed, 837 insertions(+), 30 deletions(-) create mode 100644 packages/core/src/agents/mesh/run-lifecycle.test.ts create mode 100644 packages/core/src/agents/mesh/run-lifecycle.ts diff --git a/docs/plans/2026-09-06-multi-agent-board-collaboration.md b/docs/plans/2026-09-06-multi-agent-board-collaboration.md index c14d23a4753..c417513c520 100644 --- a/docs/plans/2026-09-06-multi-agent-board-collaboration.md +++ b/docs/plans/2026-09-06-multi-agent-board-collaboration.md @@ -522,14 +522,15 @@ action on the child caused it. Pre-existing on this branch (five production files plus two tests; nothing starts an agent yet): -| File | Responsibility | -| ----------------------------------------- | -------------------------------------------------- | -| `core/src/agents/mesh/types.ts` | Entities and limits | -| `core/src/agents/mesh/mesh-store.ts` | Paths, validation, locking, CRUD | -| `core/src/agents/mesh/mentions.ts` | `@name` → agent ids | -| `core/src/agents/mesh/dispatch-policy.ts` | `decideDispatch` — pure | -| `core/src/agents/mesh/thread-actions.ts` | `postMessage` — append and book under one lock | -| `core/src/agents/mesh/thread-status.ts` | Aggregate status over every run's close obligation | +| File | Responsibility | +| ----------------------------------------- | ----------------------------------------------------- | +| `core/src/agents/mesh/types.ts` | Entities and limits | +| `core/src/agents/mesh/mesh-store.ts` | Paths, validation, locking, CRUD | +| `core/src/agents/mesh/mentions.ts` | `@name` → agent ids | +| `core/src/agents/mesh/dispatch-policy.ts` | `decideDispatch` — pure | +| `core/src/agents/mesh/thread-actions.ts` | `postMessage` — append and book under one lock | +| `core/src/agents/mesh/thread-status.ts` | Aggregate status over every run's close obligation | +| `core/src/agents/mesh/run-lifecycle.ts` | Run close, terminal state, status application, outbox | ### 5.1 Local review correction — committed and verified diff --git a/docs/plans/2026-09-07-mesh-implementation-acceptance.md b/docs/plans/2026-09-07-mesh-implementation-acceptance.md index 23aa07f8e7f..237bcb13712 100644 --- a/docs/plans/2026-09-07-mesh-implementation-acceptance.md +++ b/docs/plans/2026-09-07-mesh-implementation-acceptance.md @@ -58,6 +58,8 @@ Evidence: the assembled prompt text for cases first-entry / delta / gap / retry, **Aggregate status landed.** `thread-status.ts` derives the status from every run's close obligation rather than letting the last run to finish stamp it, and the three round-2 findings are each pinned by a test: a same-thread wait is discharged by a later close (I1), any later successful booking discharges an earlier failure or unclosed return (I2), and a quiescent thread whose last admission booked nothing becomes `blocked` (I6). Also covered: a live run outranks another agent's review, a blocker outranks a review, a wait is `in_progress` only while a child can wake it, `done` is sticky against a late post, and a failed run reports as a failure even when it recorded a close kind. Observed locally: `thread-status.test.ts` → 1 file, 13 tests passed; targeted ESLint clean. The producers that write `closeKind` are the thread tools in 5b, so nothing calls this resolver yet. +**Run close and status application landed.** `run-lifecycle.ts` splits closing into two writes: the tool records `closeKind` and moves the run to `finishing`, ending the agent's turn, and the runtime callback records the terminal state — the only place the aggregate status is recomputed. A `waiting` close is refused when nothing could wake it, and a live _descendant_ counts while a mere sibling under the same root does not. Any close discharges peers' waits on the same thread. A clean exit with no closing tool is recorded as `unclosed`, never as implicit success. `finishRun` now delegates to this one path, and `postMessage` discharges outstanding obligations when it books work and then applies the aggregate status, so the I2 and I6 fixes have producers rather than only a resolver. Observed locally: `run-lifecycle.test.ts`, `thread-actions.test.ts`, `thread-status.test.ts`, `mesh-store.test.ts` → 4 files, 57 tests passed; targeted ESLint clean. The six thread tools that call `closeRun` are still to come, so gates (a), (b), (c) and (e) remain unexecuted. + ### Step 6 — Minimal in-process dispatcher, no recovery Lands: pick the lowest `queueSequence` queued run per agent; branch on registry state `completed+resident → continue`, `completed → revive`, `paused → resume`, `unbound → launch`; `startRun` / `finishRun`; consume the parent-report outbox; leave `capacity_wait` queued. diff --git a/packages/core/src/agents/mesh/run-lifecycle.test.ts b/packages/core/src/agents/mesh/run-lifecycle.test.ts new file mode 100644 index 00000000000..3870d3c2c56 --- /dev/null +++ b/packages/core/src/agents/mesh/run-lifecycle.test.ts @@ -0,0 +1,344 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import * as fs from 'node:fs/promises'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import { Storage } from '../../config/storage.js'; +import { + createThread, + readThread, + updateMeshAgents, + writeThread, +} from './mesh-store.js'; +import { + closeRun, + finishRunInTransaction, + hasLiveDescendant, + MeshCloseRejectedError, +} from './run-lifecycle.js'; +import { withMeshStoreTransaction } from './mesh-store.js'; +import { postMessage } from './thread-actions.js'; +import { + HUMAN_AUTHOR_ID, + type MeshAgent, + type Thread, + type ThreadRun, +} from './types.js'; + +const PROJECT_ROOT = '/mesh-lifecycle-test'; +const ALICE: MeshAgent = { id: 'ag_alice', name: 'alice', createdAt: 1 }; +const BOB: MeshAgent = { id: 'ag_bob', name: 'bob', createdAt: 1 }; + +function run(overrides: Partial = {}): ThreadRun { + return { + id: 'rn_alice', + agentId: ALICE.id, + status: 'running', + triggerMessageIds: [], + acceptedMessageIds: [], + consumedMessageIds: [], + usageByRound: [], + // Well clear of the workspace counter: these fixtures are hand-written and + // must not collide with a sequence the store allocates during the test. + queueSequence: 100, + queuedAt: 1_000, + attempts: 1, + ...overrides, + }; +} + +async function seed(overrides: Partial = {}): Promise { + const created = await createThread(PROJECT_ROOT, { title: 'Investigate' }); + const thread: Thread = { + ...created, + status: 'in_progress', + runs: [run()], + ...overrides, + }; + await writeThread(PROJECT_ROOT, thread); + return thread; +} + +function finish( + threadId: string, + runId: string, + outcome: Parameters[1]['outcome'], +) { + return withMeshStoreTransaction(PROJECT_ROOT, (transaction) => + finishRunInTransaction(transaction, { threadId, runId, outcome }), + ); +} + +describe('mesh run lifecycle', () => { + let runtimeDir: string; + + beforeEach(async () => { + runtimeDir = await fs.mkdtemp(path.join(os.tmpdir(), 'mesh-lifecycle-')); + Storage.setRuntimeBaseDir(runtimeDir); + await updateMeshAgents(PROJECT_ROOT, () => [ALICE, BOB]); + }); + + afterEach(async () => { + Storage.setRuntimeBaseDir(null); + await fs.rm(runtimeDir, { recursive: true, force: true }); + }); + + it('posts the question, records the close, and ends the turn without finishing the run', async () => { + const thread = await seed(); + + const result = await closeRun(PROJECT_ROOT, { + threadId: thread.id, + runId: 'rn_alice', + agentId: ALICE.id, + request: { kind: 'blocked', question: 'which retry path?' }, + }); + + expect(result.message?.text).toBe('which retry path?'); + expect(result.message?.authorKind).toBe('agent'); + expect(result.message?.sourceRunId).toBe('rn_alice'); + expect(result.message?.authorNameSnapshot).toBe('alice'); + // The runtime is still executing, so the run may not be marked terminal. + expect(result.thread.runs[0]?.status).toBe('finishing'); + expect(result.thread.runs[0]?.closeKind).toBe('blocked'); + expect(result.thread.runs[0]?.finalMessageId).toBe(result.message?.id); + expect(result.thread.status).toBe('in_progress'); + expect(result.thread.outbox).toHaveLength(1); + expect(result.thread.outbox[0]?.payload['event']).toBe('blocker_raised'); + }); + + it('refuses a wait that nothing could ever wake', async () => { + const thread = await seed(); + + await expect( + closeRun(PROJECT_ROOT, { + threadId: thread.id, + runId: 'rn_alice', + agentId: ALICE.id, + request: { kind: 'waiting' }, + }), + ).rejects.toThrow(MeshCloseRejectedError); + }); + + it('allows a wait once a sub-thread is open, and not for a mere sibling', async () => { + const parent = await seed(); + const child = await createThread(PROJECT_ROOT, { + title: 'read the code', + parentThreadId: parent.id, + }); + + const waited = await closeRun(PROJECT_ROOT, { + threadId: parent.id, + runId: 'rn_alice', + agentId: ALICE.id, + request: { kind: 'waiting' }, + }); + expect(waited.thread.runs[0]?.closeKind).toBe('waiting'); + + // A sibling under the same root is not this thread's dependency. + const sibling = await createThread(PROJECT_ROOT, { + title: 'unrelated', + parentThreadId: parent.id, + }); + const threads = [ + { ...parent }, + { ...child, status: 'done' as const }, + { ...sibling, status: 'done' as const }, + ]; + expect(hasLiveDescendant(threads, parent.id)).toBe(false); + expect(hasLiveDescendant([{ ...parent }, { ...child }], parent.id)).toBe( + true, + ); + }); + + it('refuses a close for a run the caller does not own', async () => { + const thread = await seed(); + + await expect( + closeRun(PROJECT_ROOT, { + threadId: thread.id, + runId: 'rn_alice', + agentId: BOB.id, + request: { kind: 'review', summary: 'done' }, + }), + ).rejects.toThrow(/not a running run of agent "ag_bob"/); + }); + + it('discharges a peer wait so a review is not reported as blocked', async () => { + const thread = await seed({ + runs: [ + run({ id: 'rn_wait', status: 'completed', closeKind: 'waiting' }), + run({ + id: 'rn_bob', + agentId: BOB.id, + status: 'running', + queueSequence: 101, + }), + ], + }); + + const closed = await closeRun(PROJECT_ROOT, { + threadId: thread.id, + runId: 'rn_bob', + agentId: BOB.id, + request: { kind: 'review', summary: 'the flake is the retry path' }, + }); + expect( + closed.thread.runs.find((entry) => entry.id === 'rn_wait') + ?.closeAcknowledgedAtSequence, + ).toBe(1); + + const finished = await finish(thread.id, 'rn_bob', { status: 'completed' }); + expect(finished.status).toBe('in_review'); + }); + + it('records a clean exit with no closing tool as unclosed and blocks', async () => { + const thread = await seed(); + + const finished = await finish(thread.id, 'rn_alice', { + status: 'completed', + }); + + expect(finished.runs[0]?.closeKind).toBe('unclosed'); + expect(finished.status).toBe('blocked'); + expect( + finished.outbox.some( + (event) => event.payload['event'] === 'thread_blocked', + ), + ).toBe(true); + }); + + it('reports a child in review to its parent exactly once', async () => { + const parent = await createThread(PROJECT_ROOT, { title: 'parent' }); + const created = await createThread(PROJECT_ROOT, { + title: 'child', + parentThreadId: parent.id, + }); + await writeThread(PROJECT_ROOT, { + ...created, + status: 'in_progress', + runs: [run()], + }); + + await closeRun(PROJECT_ROOT, { + threadId: created.id, + runId: 'rn_alice', + agentId: ALICE.id, + request: { kind: 'review', summary: 'root cause found' }, + }); + const finished = await finish(created.id, 'rn_alice', { + status: 'completed', + }); + + expect(finished.status).toBe('in_review'); + const reports = finished.outbox.filter( + (event) => event.kind === 'parent_report', + ); + expect(reports).toHaveLength(1); + expect(reports[0]?.payload['parentThreadId']).toBe(parent.id); + + // Re-running the terminal write must not enqueue a second report. + const again = await finish(created.id, 'rn_alice', { status: 'completed' }); + expect(again.outbox.filter((e) => e.kind === 'parent_report')).toHaveLength( + 1, + ); + }); + + it('carries a typed failure stage onto the run and blocks the thread', async () => { + const thread = await seed(); + + const finished = await finish(thread.id, 'rn_alice', { + status: 'failed', + error: 'definition missing', + failureStage: 'launch', + }); + + expect(finished.runs[0]?.failureStage).toBe('launch'); + expect(finished.status).toBe('blocked'); + }); + + it('refuses any close on a thread a person already marked done', async () => { + const thread = await seed({ status: 'done' }); + + await expect( + closeRun(PROJECT_ROOT, { + threadId: thread.id, + runId: 'rn_alice', + agentId: ALICE.id, + request: { kind: 'review', summary: 'late' }, + }), + ).rejects.toThrow(/is done/); + }); + + it('clears an obsolete failure when a later post books real work', async () => { + const thread = await seed({ assigneeAgentId: ALICE.id }); + const failed = await finish(thread.id, 'rn_alice', { + status: 'failed', + error: 'launch failed', + }); + expect(failed.status).toBe('blocked'); + + const posted = await postMessage(PROJECT_ROOT, thread.id, { + from: HUMAN_AUTHOR_ID, + text: 'try again please', + }); + + expect(posted.dispatched).toHaveLength(1); + expect(posted.thread.status).toBe('in_progress'); + expect( + posted.thread.runs.find((entry) => entry.id === 'rn_alice') + ?.closeAcknowledgedAtSequence, + ).toBe(1); + }); + + it('blocks a quiescent thread whose post books nothing at all', async () => { + const created = await createThread(PROJECT_ROOT, { title: 'unassigned' }); + + const posted = await postMessage(PROJECT_ROOT, created.id, { + from: HUMAN_AUTHOR_ID, + text: 'anyone?', + }); + + expect(posted.dispatched).toHaveLength(0); + expect(posted.thread.status).toBe('blocked'); + expect( + posted.thread.outbox.some( + (event) => event.payload['event'] === 'thread_blocked', + ), + ).toBe(true); + }); + + it('leaves a thread in_progress while another run is still live', async () => { + const thread = await seed({ + runs: [ + run(), + run({ + id: 'rn_bob', + agentId: BOB.id, + status: 'queued', + queueSequence: 101, + }), + ], + }); + + await closeRun(PROJECT_ROOT, { + threadId: thread.id, + runId: 'rn_alice', + agentId: ALICE.id, + request: { kind: 'review', summary: 'my part is done' }, + }); + const finished = await finish(thread.id, 'rn_alice', { + status: 'completed', + }); + + expect(finished.status).toBe('in_progress'); + expect(await readThread(PROJECT_ROOT, thread.id)).toMatchObject({ + status: 'in_progress', + }); + }); +}); diff --git a/packages/core/src/agents/mesh/run-lifecycle.ts b/packages/core/src/agents/mesh/run-lifecycle.ts new file mode 100644 index 00000000000..5d1fd389c0e --- /dev/null +++ b/packages/core/src/agents/mesh/run-lifecycle.ts @@ -0,0 +1,449 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * @fileoverview How a mesh run ends, and what the thread does about it. + * + * A closing tool cannot mark its own still-executing runtime finished: the + * model is mid-turn when it calls one. So closing is two writes. The tool + * records *what* the run is closing as and moves it to `finishing`, which ends + * the agent's turn; the runtime callback then records the terminal state, and + * only there is the thread's status recomputed. Splitting it this way is what + * makes a crash between the two recoverable — a `finishing` run with a + * `closeKind` is a complete instruction for restart reconciliation, whereas a + * status written optimistically before the runtime actually stopped is a lie + * the next reader cannot detect. + * + * The status itself is never written by the closing run. See `thread-status.ts` + * for why, and for the three acknowledgement rules this module drives. + */ + +import { + generateEventId, + generateMessageId, + withMeshStoreTransaction, + type MeshStoreTransaction, +} from './mesh-store.js'; +import { + acknowledgeCloseObligations, + resolveThreadStatus, +} from './thread-status.js'; +import { + HUMAN_AUTHOR_ID, + type MeshAgent, + type Thread, + type ThreadEvent, + type ThreadMessage, + type ThreadRun, +} from './types.js'; + +/** How an agent says its run is done. `unclosed` is recorded, never chosen. */ +export type RunCloseRequest = + | { kind: 'waiting' } + | { kind: 'blocked'; question: string } + | { kind: 'review'; summary: string }; + +/** Raised when a close is refused, so a tool can tell the model what to do. */ +export class MeshCloseRejectedError extends Error { + constructor( + readonly code: 'no_live_dependency' | 'run_not_bound' | 'thread_done', + message: string, + ) { + super(message); + this.name = 'MeshCloseRejectedError'; + } +} + +export interface CloseRunInput { + threadId: string; + runId: string; + /** From the ambient frame, never from the model. */ + agentId: string; + request: RunCloseRequest; + now?: number; +} + +export interface CloseRunResult { + thread: Thread; + /** Present for `blocked` and `review`, which post before they close. */ + message?: ThreadMessage; +} + +/** + * A descendant of `threadId` that is not `done`. + * + * Walked over `parentThreadId` rather than `rootThreadId` so a sibling + * sub-thread of the same root does not count as this thread's dependency — + * waiting on work that was never delegated here is exactly the stranded wait + * the status resolver has to catch. + */ +export function hasLiveDescendant( + threads: readonly Thread[], + threadId: string, +): boolean { + const byParent = new Map(); + for (const thread of threads) { + if (!thread.parentThreadId) continue; + const siblings = byParent.get(thread.parentThreadId) ?? []; + siblings.push(thread); + byParent.set(thread.parentThreadId, siblings); + } + const seen = new Set([threadId]); + const queue = [threadId]; + while (queue.length > 0) { + const current = queue.shift()!; + for (const child of byParent.get(current) ?? []) { + if (seen.has(child.id)) continue; + seen.add(child.id); + if (child.status !== 'done') return true; + queue.push(child.id); + } + } + return false; +} + +function appendMessage( + thread: Thread, + fields: { + from: string; + authorNameSnapshot: string; + text: string; + sourceRunId?: string; + triggerKind?: string; + authorKind: ThreadMessage['authorKind']; + }, + now: number, +): { thread: Thread; message: ThreadMessage } { + const message: ThreadMessage = { + id: generateMessageId(), + sequence: thread.nextMessageSequence, + authorKind: fields.authorKind, + from: fields.from, + authorNameSnapshot: fields.authorNameSnapshot, + ...(fields.sourceRunId ? { sourceRunId: fields.sourceRunId } : {}), + ...(fields.triggerKind ? { triggerKind: fields.triggerKind } : {}), + text: fields.text, + mentions: [], + outcomes: [], + at: now, + }; + return { + thread: { + ...thread, + messages: [...thread.messages, message], + nextMessageSequence: thread.nextMessageSequence + 1, + }, + message, + }; +} + +function enqueue( + thread: Thread, + event: Omit, + now: number, +): Thread { + const stored: ThreadEvent = { + ...event, + id: generateEventId(), + status: 'pending', + attempts: 0, + createdAt: now, + }; + return { ...thread, outbox: [...thread.outbox, stored] }; +} + +/** + * Records a run's close and ends its turn. + * + * The run is verified against the caller's ambient identity before anything is + * written: a close that names a run the agent does not own, or a run that is + * not executing, is a wiring or replay error, not a workflow event. + */ +export async function closeRunInTransaction( + transaction: MeshStoreTransaction, + input: CloseRunInput, +): Promise { + const now = input.now ?? Date.now(); + const thread = await transaction.readThread(input.threadId); + if (!thread) throw new Error(`No thread with id "${input.threadId}".`); + if (thread.status === 'done') { + throw new MeshCloseRejectedError( + 'thread_done', + `Thread "${input.threadId}" is done; it accepts no further work.`, + ); + } + + const run = thread.runs.find((entry) => entry.id === input.runId); + if (!run || run.agentId !== input.agentId || run.status !== 'running') { + throw new MeshCloseRejectedError( + 'run_not_bound', + `Run "${input.runId}" is not a running run of agent "${input.agentId}" on thread "${input.threadId}".`, + ); + } + + if (input.request.kind === 'waiting') { + const otherLive = thread.runs.some( + (entry) => + entry.id !== run.id && + (entry.status === 'queued' || + entry.status === 'running' || + entry.status === 'finishing'), + ); + const { threads } = await transaction.listThreads(); + if (!otherLive && !hasLiveDescendant(threads, thread.id)) { + throw new MeshCloseRejectedError( + 'no_live_dependency', + 'Nothing else is running on this thread and no sub-thread is open, so waiting would strand it. Block with a question, submit for review, or keep working.', + ); + } + } + + const agents = await transaction.readAgents(); + const self = agents.find((agent) => agent.id === input.agentId); + const authorName = self?.name ?? input.agentId; + + let next = thread; + let message: ThreadMessage | undefined; + if (input.request.kind !== 'waiting') { + const appended = appendMessage( + next, + { + authorKind: 'agent', + from: input.agentId, + authorNameSnapshot: authorName, + sourceRunId: run.id, + triggerKind: `thread_${input.request.kind}`, + text: + input.request.kind === 'blocked' + ? input.request.question + : input.request.summary, + }, + now, + ); + next = appended.thread; + message = appended.message; + } + + // Any close discharges peers' waits on this thread: whatever they were + // waiting to see has now happened, and leaving the obligation outstanding + // would report the thread blocked when it is merely finished. + next = acknowledgeCloseObligations( + next, + next.nextMessageSequence - 1, + (obligation) => + obligation.kind === 'waiting' && obligation.runId !== run.id, + ); + + next = { + ...next, + runs: next.runs.map((entry) => + entry.id === run.id + ? { + ...entry, + status: 'finishing', + closeKind: input.request.kind, + ...(message ? { finalMessageId: message.id } : {}), + } + : entry, + ), + }; + + if (input.request.kind === 'blocked') { + next = enqueue( + next, + { + kind: 'notification', + causedByRunId: run.id, + payload: { + event: 'blocker_raised', + threadId: thread.id, + agentId: input.agentId, + messageId: message?.id, + }, + }, + now, + ); + } + + return { + thread: await transaction.writeThread(next), + ...(message ? { message } : {}), + }; +} + +export async function closeRun( + projectRoot: string, + input: CloseRunInput, +): Promise { + return withMeshStoreTransaction(projectRoot, (transaction) => + closeRunInTransaction(transaction, input), + ); +} + +/** + * Applies the aggregate status and emits what the new status owes. + * + * Called after any write that can make a thread quiescent. The parent report + * is emitted here rather than at close time because `in_review` is a property + * of the whole thread: an agent submitting its part while another still works + * must not wake the parent. + */ +export async function applyAggregateStatus( + transaction: MeshStoreTransaction, + thread: Thread, + now = Date.now(), +): Promise { + const { threads } = await transaction.listThreads(); + const resolution = resolveThreadStatus({ + thread, + hasLiveChildDependency: hasLiveDescendant(threads, thread.id), + }); + if (resolution.status === thread.status) return thread; + + let next: Thread = { ...thread, status: resolution.status }; + + const alreadyReported = (kind: string) => + next.outbox.some( + (event) => event.payload['event'] === kind && event.status === 'pending', + ); + + if (resolution.status === 'in_review') { + if (next.parentThreadId && !alreadyReported('child_in_review')) { + next = enqueue( + next, + { + kind: 'parent_report', + payload: { + event: 'child_in_review', + threadId: next.id, + parentThreadId: next.parentThreadId, + summaryMessageId: next.messages[next.messages.length - 1]?.id, + }, + }, + now, + ); + } + if (!alreadyReported('thread_in_review')) { + next = enqueue( + next, + { + kind: 'notification', + payload: { event: 'thread_in_review', threadId: next.id }, + }, + now, + ); + } + } + + if (resolution.status === 'blocked' && !alreadyReported('thread_blocked')) { + next = enqueue( + next, + { + kind: 'notification', + payload: { + event: 'thread_blocked', + threadId: next.id, + reason: resolution.reason, + }, + }, + now, + ); + } + + return next; +} + +/** + * Records a run's terminal state and recomputes the thread from it. + * + * A run that already reached a terminal state is left alone so a late + * completion cannot overwrite a cancellation. + */ +export async function finishRunInTransaction( + transaction: MeshStoreTransaction, + input: { + threadId: string; + runId: string; + outcome: { + status: 'completed' | 'failed' | 'cancelled'; + error?: string; + failureStage?: string; + }; + now?: number; + }, +): Promise { + const now = input.now ?? Date.now(); + const thread = await transaction.readThread(input.threadId); + if (!thread) throw new Error(`No thread with id "${input.threadId}".`); + + let next: Thread = { + ...thread, + runs: thread.runs.map((run) => + run.id === input.runId && + (run.status === 'queued' || + run.status === 'running' || + run.status === 'finishing' || + run.status === 'cancelling') + ? { + ...run, + status: input.outcome.status, + endedAt: now, + // A run that stopped without calling a closing tool is recorded as + // `unclosed`, never as an implicit success. + closeKind: + run.closeKind ?? + (input.outcome.status === 'completed' ? 'unclosed' : undefined), + ...(input.outcome.error ? { error: input.outcome.error } : {}), + ...(input.outcome.failureStage + ? { failureStage: input.outcome.failureStage } + : {}), + } + : run, + ), + }; + + next = await applyAggregateStatus(transaction, next, now); + return transaction.writeThread(next); +} + +/** + * Records that a person's or an agent's post booked work, discharging the + * obligations an earlier failure or unclosed return left behind. + * + * Separate from the close path because the trigger is different: this is + * "something new is running now", not "a run finished". + */ +export function acknowledgeAfterBooking( + thread: Thread, + atSequence: number, +): Thread { + return acknowledgeCloseObligations(thread, atSequence); +} + +/** Agents currently holding a live run on this thread, for UI and dispatch. */ +export function liveRunsFor( + thread: Thread, + agents: readonly MeshAgent[], +): Array<{ run: ThreadRun; agent?: MeshAgent }> { + return thread.runs + .filter( + (run) => + run.status === 'queued' || + run.status === 'running' || + run.status === 'finishing' || + run.status === 'cancelling', + ) + .map((run) => ({ + run, + ...(agents.find((agent) => agent.id === run.agentId) + ? { agent: agents.find((agent) => agent.id === run.agentId)! } + : {}), + })); +} + +/** Author id used when the system, not a person or agent, appends a post. */ +export const SYSTEM_AUTHOR_ID = 'system'; +export { HUMAN_AUTHOR_ID }; diff --git a/packages/core/src/agents/mesh/thread-actions.ts b/packages/core/src/agents/mesh/thread-actions.ts index b550a582f1f..920e2e5e31a 100644 --- a/packages/core/src/agents/mesh/thread-actions.ts +++ b/packages/core/src/agents/mesh/thread-actions.ts @@ -11,6 +11,11 @@ import { type MeshStoreTransaction, } from './mesh-store.js'; import { parseMentions } from './mentions.js'; +import { + applyAggregateStatus, + finishRunInTransaction, +} from './run-lifecycle.js'; +import { acknowledgeCloseObligations } from './thread-status.js'; import { decideDispatch, resolveTargets, @@ -298,6 +303,20 @@ export async function postMessageInTransaction( candidate.id === message.id ? storedMessage : candidate, ), }; + // A post that actually books work says the thread has moved on, so an + // earlier failure or unclosed return stops pinning it to `blocked`. Round-2 + // finding I2: acknowledgement used to be human-only, which left one launch + // failure blocking the thread even after another agent finished the job. + if ( + dispatched.length > 0 || + outcomes.some((o) => o.decision.kind === 'coalesce') + ) { + next = acknowledgeCloseObligations(next, storedMessage.sequence); + } + // The status is an aggregate over every run, never last-writer-wins, and it + // is recomputed here so an admission that books nothing cannot leave the + // thread sitting in `in_progress` with no live run and no explanation. + next = await applyAggregateStatus(transaction, next, now); const thread = await transaction.writeThread(next); return { thread, @@ -346,34 +365,26 @@ export async function startRun( }); } +/** + * Records a run's terminal state. + * + * Delegates to the lifecycle module so a run has exactly one way to end and + * the thread's aggregate status is recomputed from the same place every time. + */ export async function finishRun( projectRoot: string, threadId: string, runId: string, - outcome: { status: 'completed' | 'failed' | 'cancelled'; error?: string }, + outcome: { + status: 'completed' | 'failed' | 'cancelled'; + error?: string; + failureStage?: string; + }, now = Date.now(), ): Promise { - return withMeshStoreTransaction(projectRoot, async (transaction) => { - const thread = await transaction.readThread(threadId); - if (!thread) throw new Error(`No thread with id "${threadId}".`); - return transaction.writeThread({ - ...thread, - runs: thread.runs.map((run) => - run.id === runId && - (run.status === 'queued' || - run.status === 'running' || - run.status === 'finishing' || - run.status === 'cancelling') - ? { - ...run, - status: outcome.status, - endedAt: now, - ...(outcome.error ? { error: outcome.error } : {}), - } - : run, - ), - }); - }); + return withMeshStoreTransaction(projectRoot, (transaction) => + finishRunInTransaction(transaction, { threadId, runId, outcome, now }), + ); } export async function upsertRunUsage( From cbc043995c350463f9a36e892953718c4ebf3e59 Mon Sep 17 00:00:00 2001 From: yiliang114 <1204183885@qq.com> Date: Mon, 7 Sep 2026 11:09:41 +0900 Subject: [PATCH 2/2] refactor(mesh): carry system-trigger provenance on posts, drop dead exports A structured assignment or parent report is system-authored but must keep the run or human action that caused it, so it is charged as unattended work without being suppressed as an ordinary self-authored post. PostMessageInput now carries authorKind, sourceRunId and triggerKind; all three are derived by the server and none is accepted from a model. Removes three exports from run-lifecycle that had no reader: the booking acknowledger duplicated what postMessage already calls directly, and the live run listing and re-exported author id were never read. --- .../core/src/agents/mesh/run-lifecycle.ts | 48 +------------------ .../core/src/agents/mesh/thread-actions.ts | 24 ++++++++-- 2 files changed, 22 insertions(+), 50 deletions(-) diff --git a/packages/core/src/agents/mesh/run-lifecycle.ts b/packages/core/src/agents/mesh/run-lifecycle.ts index 5d1fd389c0e..b7d7baf4570 100644 --- a/packages/core/src/agents/mesh/run-lifecycle.ts +++ b/packages/core/src/agents/mesh/run-lifecycle.ts @@ -31,14 +31,7 @@ import { acknowledgeCloseObligations, resolveThreadStatus, } from './thread-status.js'; -import { - HUMAN_AUTHOR_ID, - type MeshAgent, - type Thread, - type ThreadEvent, - type ThreadMessage, - type ThreadRun, -} from './types.js'; +import type { Thread, ThreadEvent, ThreadMessage } from './types.js'; /** How an agent says its run is done. `unclosed` is recorded, never chosen. */ export type RunCloseRequest = @@ -408,42 +401,3 @@ export async function finishRunInTransaction( next = await applyAggregateStatus(transaction, next, now); return transaction.writeThread(next); } - -/** - * Records that a person's or an agent's post booked work, discharging the - * obligations an earlier failure or unclosed return left behind. - * - * Separate from the close path because the trigger is different: this is - * "something new is running now", not "a run finished". - */ -export function acknowledgeAfterBooking( - thread: Thread, - atSequence: number, -): Thread { - return acknowledgeCloseObligations(thread, atSequence); -} - -/** Agents currently holding a live run on this thread, for UI and dispatch. */ -export function liveRunsFor( - thread: Thread, - agents: readonly MeshAgent[], -): Array<{ run: ThreadRun; agent?: MeshAgent }> { - return thread.runs - .filter( - (run) => - run.status === 'queued' || - run.status === 'running' || - run.status === 'finishing' || - run.status === 'cancelling', - ) - .map((run) => ({ - run, - ...(agents.find((agent) => agent.id === run.agentId) - ? { agent: agents.find((agent) => agent.id === run.agentId)! } - : {}), - })); -} - -/** Author id used when the system, not a person or agent, appends a post. */ -export const SYSTEM_AUTHOR_ID = 'system'; -export { HUMAN_AUTHOR_ID }; diff --git a/packages/core/src/agents/mesh/thread-actions.ts b/packages/core/src/agents/mesh/thread-actions.ts index 920e2e5e31a..9ae35aacb56 100644 --- a/packages/core/src/agents/mesh/thread-actions.ts +++ b/packages/core/src/agents/mesh/thread-actions.ts @@ -36,8 +36,22 @@ export interface PostMessageInput { from: string; text: string; originEventId?: string; + /** + * `system` for a structured trigger — an assignment, or a parent dependency + * report. Derived from `from` when absent. A system trigger still records the + * run or human action that caused it, so it is charged as unattended work + * without being suppressed as an ordinary self-authored post. + */ + authorKind?: ThreadMessage['authorKind']; + /** The run that caused this post. Server-derived; never model-supplied. */ + sourceRunId?: string; + /** What kind of trigger this was, e.g. `assignment`. */ + triggerKind?: string; } +/** Author id recorded for a post neither a person nor an agent wrote. */ +export const SYSTEM_AUTHOR_ID = 'system'; + export interface TargetOutcome { agentId?: string; agentName?: string; @@ -181,16 +195,20 @@ export async function postMessageInTransaction( const message: ThreadMessage = { id: generateMessageId(), sequence: current.nextMessageSequence, - authorKind: input.from === HUMAN_AUTHOR_ID ? 'human' : 'agent', + authorKind: + input.authorKind ?? + (input.from === HUMAN_AUTHOR_ID ? 'human' : 'agent'), from: input.from, authorNameSnapshot: - input.from === HUMAN_AUTHOR_ID - ? HUMAN_AUTHOR_ID + input.from === HUMAN_AUTHOR_ID || input.from === SYSTEM_AUTHOR_ID + ? input.from : (agents.find((agent) => agent.id === input.from)?.name ?? input.from), text: input.text, mentions: parsed.ids, outcomes: [], at: now, + ...(input.sourceRunId ? { sourceRunId: input.sourceRunId } : {}), + ...(input.triggerKind ? { triggerKind: input.triggerKind } : {}), ...(input.originEventId ? { originEventId: input.originEventId } : {}), }; const outcomes: TargetOutcome[] = parsed.unknown.map((agentName) => ({