diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ae4f2b..d921998 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,19 @@ contain breaking changes**; patch releases are fixes only. ## Unreleased +- **[BREAKING] `@polymind-inc/agent-framework-a2a`** — a remote task's status message becomes a + response message only when the task is waiting for input (`input-required`). Previously an + awaited `run()` also materialized the status message of a `completed`, `failed`, `canceled` or + `rejected` task, and fell back to the last agent message in `task.history` for a terminal task + with no artifacts, while the streamed form of the same task did neither — so how the run was + consumed changed the answer. Both paths now follow one rule, matching .NET, whose + `AgentTaskStatusExtensions` returns content for `TaskState.InputRequired` alone and never reads + `task.history`. An agent that answers with a closing status message and no artifact now folds to + an empty response: read the task from `rawRepresentation` on the response, whose messages still + carry it, or take the state from the session. An `input-required` status message that carries no + parts likewise no longer names a message on the awaited path, which the streamed path already + declined to do. Artifact conversion and the streamed-artifact deduplication are unchanged. + - **`@polymind-inc/agent-framework-core`** — an approval granted for a call id that an earlier completed call had already used is no longer discarded. The approval layer correlated decisions against a transcript-wide set of answered call ids, so a provider that reused a call id produced diff --git a/packages/a2a/README.md b/packages/a2a/README.md index 5191b23..198dcb7 100644 --- a/packages/a2a/README.md +++ b/packages/a2a/README.md @@ -111,6 +111,24 @@ context providers of its own — the shared run-option fields it does not declar `middleware`, `responseFormat`, `options`) are ignored, as they are in the .NET, Python and Go implementations of this client. +### What becomes a response message + +A task's **artifacts are its answer**: each one becomes a single message, and an artifact already +delivered by a streamed `TaskArtifactUpdateEvent` is not repeated when the closing task snapshot +carries it again. + +A task's **status message becomes a message only when the task is waiting for input** +(`input-required`), because that message is the question addressed to you. In every other state it +describes the run rather than answering it — the progress notes of a `working` task, the closing +remark of a `completed` one, the reason a `failed`, `canceled` or `rejected` one stopped, the +challenge of an `auth-required` one — and none of those reach the transcript, which would otherwise +put "working on it" or "done" where the answer belongs. A task's `history` is never a source of +messages either: it is the conversation so far, not this turn's output. + +Nothing is lost by this: the task or event each update came from is on `rawRepresentation`, and the +task's state is on the session. The rule is the same whichever way you consume the run, so a task +that answers with a closing remark and no artifacts folds to an empty response either way. + ## Sessions A session holds the remote conversation's identity, and is plain JSON: @@ -180,8 +198,11 @@ original part is always on `rawRepresentation`. needs to continue one conversation across calls. - **No push notifications, task listing or cancellation.** The client covers send, stream, get and re-subscribe. Use the SDK client directly for the rest. -- **Progress messages are not transcript.** A status message is turned into content only when the - task is waiting for input; commentary attached to `working` is dropped. +- **Status messages are not transcript.** Only an `input-required` status message becomes content; + see [What becomes a response message](#what-becomes-a-response-message). An agent that answers + with a closing status message and no artifact therefore folds to an empty response — read + `rawRepresentation` for the task itself. This matches .NET, which materializes status content for + `input-required` alone; Python and Go each surface a wider set. - **Server hosting is not part of this package.** Exposing a framework agent *as* an A2A agent is a separate concern; use `@a2a-js/sdk/server` directly. diff --git a/packages/a2a/src/agent.test.ts b/packages/a2a/src/agent.test.ts index a69614b..647f2f4 100644 --- a/packages/a2a/src/agent.test.ts +++ b/packages/a2a/src/agent.test.ts @@ -1,6 +1,15 @@ import { UnsupportedOperationError } from '@a2a-js/sdk/errors'; -import type { AgentResponseUpdate, ContinuationToken } from '@polymind-inc/agent-framework-core'; -import { AgentSession, ConfigurationError, isAbortError } from '@polymind-inc/agent-framework-core'; +import type { + AgentResponse, + AgentResponseUpdate, + ContinuationToken, +} from '@polymind-inc/agent-framework-core'; +import { + AgentSession, + ConfigurationError, + isAbortError, + textOfContents, +} from '@polymind-inc/agent-framework-core'; import { assert, describe, expect, it } from 'vitest'; import { A2AAgent } from './agent.js'; import { A2AAgentError } from './errors.js'; @@ -339,6 +348,145 @@ describe('a streamed turn', () => { }); }); +/** + * Folds one remote task both ways. + * + * Awaiting a run reads a whole task snapshot; iterating it reads the status and artifact events + * that describe the same task. Both are handed the same wire JSON, so any difference in the folded + * transcript is this package's, not the agent's. + */ +async function foldBothWays( + snapshot: Record, + events: Array>, +): Promise<{ awaited: AgentResponse; streamed: AgentResponse }> { + const blocking = fakeClient({ sendMessage: task(snapshot) }); + const streaming = fakeClient({ sendMessageStream: events.map((event) => streamEvent(event)) }); + + const awaited = await new A2AAgent({ client: blocking.client, id: 'a1' }).run('q'); + const stream = new A2AAgent({ client: streaming.client, id: 'a1' }).run('q'); + for await (const _ of stream) { + // Drain: the folded transcript, not the individual updates, is what is being compared. + } + return { awaited, streamed: await stream.finalResponse() }; +} + +/** The whole folded transcript, as what a caller of either path actually reads. */ +function transcript(response: AgentResponse): Array<{ role: string; text: string }> { + return response.messages.map((item) => ({ role: item.role, text: textOfContents(item.contents) })); +} + +describe('one task consumed both ways', () => { + it('keeps a terminal status message out of the transcript on both paths', async () => { + const status = { + state: 'TASK_STATE_COMPLETED', + message: { messageId: 's1', role: 'ROLE_AGENT', parts: [{ text: 'Finished the lookup.' }] }, + }; + + const { awaited, streamed } = await foldBothWays({ id: 'task-1', contextId: 'ctx-1', status }, [ + { statusUpdate: { taskId: 'task-1', contextId: 'ctx-1', status } }, + ]); + + expect(transcript(awaited)).toEqual(transcript(streamed)); + // The closing commentary of a finished task is not its answer, so neither path speaks it. + expect(awaited.text).toBe(''); + expect(streamed.text).toBe(''); + }); + + it.each(['TASK_STATE_FAILED', 'TASK_STATE_CANCELED', 'TASK_STATE_REJECTED', 'TASK_STATE_AUTH_REQUIRED'])( + 'keeps a %s status message out of the transcript on both paths', + async (state) => { + const status = { + state, + message: { messageId: 's1', role: 'ROLE_AGENT', parts: [{ text: 'the reason' }] }, + }; + + const { awaited, streamed } = await foldBothWays({ id: 'task-1', contextId: 'ctx-1', status }, [ + { statusUpdate: { taskId: 'task-1', contextId: 'ctx-1', status } }, + ]); + + expect(transcript(awaited)).toEqual(transcript(streamed)); + expect(awaited.text).toBe(''); + expect(streamed.text).toBe(''); + }, + ); + + it('materializes an input-required status message identically on both paths', async () => { + const status = { + state: 'TASK_STATE_INPUT_REQUIRED', + message: { messageId: 'q1', role: 'ROLE_AGENT', parts: [{ text: 'Which invoice?' }] }, + }; + + const { awaited, streamed } = await foldBothWays({ id: 'task-1', contextId: 'ctx-1', status }, [ + { statusUpdate: { taskId: 'task-1', contextId: 'ctx-1', status } }, + ]); + + expect(transcript(awaited)).toEqual(transcript(streamed)); + // A question is addressed to the caller, so it is the one status message that is an answer. + expect(awaited.text).toBe('Which invoice?'); + expect(streamed.text).toBe('Which invoice?'); + }); + + it('names no message for an input-required status that carries no parts', async () => { + const status = { + state: 'TASK_STATE_INPUT_REQUIRED', + message: { messageId: 'q1', role: 'ROLE_AGENT', parts: [] }, + }; + + const { awaited, streamed } = await foldBothWays({ id: 'task-1', contextId: 'ctx-1', status }, [ + { statusUpdate: { taskId: 'task-1', contextId: 'ctx-1', status } }, + ]); + + // A question with nothing in it asks nothing. Identifying the message anyway would start an + // empty one during folding and split whatever surrounds it. + expect(transcript(awaited)).toEqual(transcript(streamed)); + expect(awaited.messages.map((item) => item.messageId)).toEqual( + streamed.messages.map((item) => item.messageId), + ); + }); + + it('does not synthesize an answer from the history of a terminal task without artifacts', async () => { + const history = [ + { messageId: 'u1', contextId: 'ctx-1', role: 'ROLE_USER', parts: [{ text: 'question' }] }, + { messageId: 'a1', contextId: 'ctx-1', role: 'ROLE_AGENT', parts: [{ text: 'answer from history' }] }, + ]; + const status = { state: 'TASK_STATE_COMPLETED' }; + + const { awaited, streamed } = await foldBothWays({ id: 'task-1', contextId: 'ctx-1', status, history }, [ + { statusUpdate: { taskId: 'task-1', contextId: 'ctx-1', status } }, + ]); + + // The history is the conversation so far, not new output: replaying it would answer with a + // message the caller already has, and only the awaited path could ever see it. + expect(transcript(awaited)).toEqual(transcript(streamed)); + expect(awaited.text).toBe(''); + expect(streamed.text).toBe(''); + }); + + it('emits one message per artifact and never twice for a streamed one', async () => { + const artifacts = [ + { artifactId: 'a1', parts: [{ text: 'Invoice 42 ' }] }, + { artifactId: 'a2', parts: [{ text: 'is paid.' }] }, + ]; + const status = { state: 'TASK_STATE_COMPLETED' }; + const snapshot = { id: 'task-1', contextId: 'ctx-1', status, artifacts }; + + const { awaited, streamed } = await foldBothWays(snapshot, [ + { artifactUpdate: { taskId: 'task-1', contextId: 'ctx-1', artifact: artifacts[0] } }, + { artifactUpdate: { taskId: 'task-1', contextId: 'ctx-1', artifact: artifacts[1] } }, + // Agents that close a stream by repeating the whole task must not double the answer. + { task: snapshot }, + ]); + + // One message per artifact, in order, and the repeated snapshot adds none. + const expected = [ + { role: 'assistant', text: 'Invoice 42 ' }, + { role: 'assistant', text: 'is paid.' }, + ]; + expect(transcript(awaited)).toEqual(expected); + expect(transcript(streamed)).toEqual(expected); + }); +}); + describe('linking turns together', () => { it('continues a task that is waiting for input', async () => { const asking = task({ diff --git a/packages/a2a/src/agent.ts b/packages/a2a/src/agent.ts index 2963017..15bc7e5 100644 --- a/packages/a2a/src/agent.ts +++ b/packages/a2a/src/agent.ts @@ -114,6 +114,19 @@ function hasReason(error: unknown, reason: string): boolean { * Fields of the shared run options it does not declare (`tools`, `middleware`, `responseFormat`, * `options`) are ignored, as they are in the other implementations of this protocol. * + * ## What becomes a response message + * + * A task's artifacts are its answer: one message each, and an artifact already delivered while + * streaming is not repeated by the closing task snapshot. A task's status message becomes a message + * only when the task is waiting for input (`input-required`), where that message is the question + * addressed to the caller; a status message in any other state describes the run rather than + * answering it and stays out of the transcript, as does the task's `history`, which is the + * conversation so far rather than this turn's output. The task or event behind every update is + * still on `rawRepresentation`, and the task state is on the session. + * + * The rule does not depend on how the run is consumed, so a task that answers with a closing status + * message and no artifacts folds to an empty response whether awaited or streamed. + * * ## Security considerations * * - **The remote agent is untrusted.** Everything it returns — text, structured data, file URLs — diff --git a/packages/a2a/src/convert.test.ts b/packages/a2a/src/convert.test.ts index 8ea78a5..5414a50 100644 --- a/packages/a2a/src/convert.test.ts +++ b/packages/a2a/src/convert.test.ts @@ -310,7 +310,35 @@ describe('payloads to response updates', () => { expect(question.continuationToken).toBeUndefined(); }); - it('surfaces a terminal status message, including a failure reason', () => { + it.each(['TASK_STATE_COMPLETED', 'TASK_STATE_FAILED', 'TASK_STATE_CANCELED', 'TASK_STATE_REJECTED'])( + 'keeps the status message of a %s task out of the transcript', + (state) => { + const taskValue = task({ + id: 'task-1', + contextId: 'ctx-1', + status: { + state, + message: { + messageId: 'status-1', + role: 'ROLE_AGENT', + parts: [{ text: 'invoice service failed' }], + }, + }, + }); + + const updates = updatesFromPayload({ $case: 'task', value: taskValue }, {}); + + expect(updates.flatMap((update) => update.contents)).toEqual([]); + // Dropped from the transcript, not from the response: the whole task is still reachable. + const [update] = updates; + assert.exists(update); + expect(update.rawRepresentation).toBe(taskValue); + }, + ); + + it('keeps the status message of an auth-required task out of the transcript', () => { + // Not terminal — the task waits for the caller to authenticate — but the challenge describes + // the run rather than answering it, so it is commentary like any other non-question status. const updates = updatesFromPayload( { $case: 'task', @@ -318,22 +346,18 @@ describe('payloads to response updates', () => { id: 'task-1', contextId: 'ctx-1', status: { - state: 'TASK_STATE_FAILED', - message: { - messageId: 'failure-1', - role: 'ROLE_AGENT', - parts: [{ text: 'invoice service failed' }], - }, + state: 'TASK_STATE_AUTH_REQUIRED', + message: { messageId: 'auth-1', role: 'ROLE_AGENT', parts: [{ text: 'sign in first' }] }, }, }), }, {}, ); - expect(updates.map((update) => update.text)).toContain('invoice service failed'); + expect(updates.flatMap((update) => update.contents)).toEqual([]); }); - it('falls back to the last agent history message when a task has no artifacts', () => { + it('does not fall back to history when a terminal task has no artifacts', () => { const updates = updatesFromPayload( { $case: 'task', @@ -350,7 +374,8 @@ describe('payloads to response updates', () => { {}, ); - expect(updates.map((update) => update.text)).toContain('answer from history'); + // The history is the conversation so far, not this turn's output. + expect(updates.flatMap((update) => update.contents)).toEqual([]); }); it('does not replay a history message while a task is still working', () => { @@ -365,7 +390,7 @@ describe('payloads to response updates', () => { }); // Two polls of the same unfinished task, as resuming produces: neither may present the - // history as fresh output, or every poll would repeat it. + // history as fresh output, or every poll would repeat the same answer. for (let poll = 0; poll < 2; poll += 1) { const updates = updatesFromPayload({ $case: 'task', value: working }, {}); @@ -376,7 +401,7 @@ describe('payloads to response updates', () => { } }); - it('does not fall back to history when the terminal task repeats only streamed artifacts', () => { + it('adds nothing when a terminal task repeats only streamed artifacts', () => { const observed: ObservedTaskState = {}; const streamed = streamEvent({ artifactUpdate: { @@ -403,11 +428,12 @@ describe('payloads to response updates', () => { observed, ); - // The artifact already delivered the answer; the history copy of it must not bring it back. + // The artifact already delivered the answer; neither the snapshot's copy of it nor the history + // may bring it back. expect(terminal.flatMap((update) => update.contents)).toEqual([]); }); - it('emits a message mirrored in both the status and the history exactly once', () => { + it('emits nothing for a message mirrored in both the status and the history', () => { const finalMessage = { messageId: 'final-1', role: 'ROLE_AGENT', parts: [{ text: 'the answer' }] }; const updates = updatesFromPayload( @@ -423,7 +449,9 @@ describe('payloads to response updates', () => { {}, ); - expect(updates.map((update) => update.text).filter((text) => text === 'the answer')).toHaveLength(1); + // Neither source of a terminal task's closing message contributes, so there is nothing to + // deduplicate between them. + expect(updates.flatMap((update) => update.contents)).toEqual([]); }); it('does not emit an artifact again when the terminal task repeats a streamed artifact', () => { diff --git a/packages/a2a/src/convert.ts b/packages/a2a/src/convert.ts index 25de7f8..f01b8ed 100644 --- a/packages/a2a/src/convert.ts +++ b/packages/a2a/src/convert.ts @@ -275,21 +275,35 @@ function artifactUpdate( }); } -function isTerminalState(state: TaskState | undefined): boolean { - return ( - state === TaskState.TASK_STATE_COMPLETED || - state === TaskState.TASK_STATE_FAILED || - state === TaskState.TASK_STATE_CANCELED || - state === TaskState.TASK_STATE_REJECTED - ); +/** + * What a task status contributes to the transcript, which is nothing unless the task is waiting + * for input. + * + * Only then is the status message part of the answer: it is the question addressed to the caller, + * so it has to be readable for the caller to answer it. A status message in any other state — the + * progress notes of a working task, the closing remark of a `completed` one, the reason a `failed`, + * `canceled` or `rejected` one stopped, or the challenge of an `auth-required` one — describes the + * run rather than answering it, and folding it in would put "working on it" or "done" where the + * answer belongs. None of it is lost: every update carries the protocol object it came from in + * `rawRepresentation`, and the task state is on the session. + * + * This is the single rule for both ways a run is consumed. A whole-task snapshot (an awaited run) + * and a status-update event (a streamed one) describe the same remote task, so they must produce + * the same transcript. An empty result therefore also has to be left unnamed by both: a message id + * whose content was dropped would start an empty message during folding and split the surrounding + * artifact in two. + */ +function statusContents(state: TaskState | undefined, statusMessage: A2AMessage | undefined): Content[] { + return state === TaskState.TASK_STATE_INPUT_REQUIRED && statusMessage !== undefined + ? fromA2AParts(statusMessage.parts) + : []; } /** * Turns a whole task into updates: one per artifact, plus the question it is waiting on. * - * A status message is only content when the task is waiting for input. In any other state it is - * progress commentary the agent may or may not send, and folding it into the transcript would put - * "working on it" in front of the answer. + * A task's own `history` is never a source of updates: it is the conversation so far, not this + * turn's output, and replaying it would answer with messages the caller already has. */ function updatesFromTask(task: Task, observed: ObservedTaskState): AgentResponseUpdate[] { const state = task.status?.state; @@ -302,34 +316,11 @@ function updatesFromTask(task: Task, observed: ObservedTaskState): AgentResponse ); const statusMessage = task.status?.message; - if (isTerminalState(state) && artifacts.length === 0) { - // A finished task that produced no artifacts at all may still have answered as a plain - // message, kept in its history. Only that case falls back to the history: an unfinished - // task's history would be replayed by every poll, a task whose artifacts were merely - // filtered as already streamed has delivered its answer, and a history message the status - // branch below is about to surface would arrive twice. - const historyAnswer = [...(task.history ?? [])] - .reverse() - .find((message) => message.role === Role.ROLE_AGENT); - if (historyAnswer !== undefined && historyAnswer.messageId !== statusMessage?.messageId) { - updates.push( - update({ - contents: fromA2AParts(historyAnswer.parts), - responseId: task.id, - messageId: historyAnswer.messageId, - additionalProperties: mergeMetadata(historyAnswer.metadata, task.metadata), - rawRepresentation: task, - }), - ); - } - } - if ( - (state === TaskState.TASK_STATE_INPUT_REQUIRED || isTerminalState(state)) && - statusMessage !== undefined - ) { + const question = statusContents(state, statusMessage); + if (question.length > 0 && statusMessage !== undefined) { updates.push( update({ - contents: fromA2AParts(statusMessage.parts), + contents: question, responseId: task.id, messageId: statusMessage.messageId, additionalProperties: mergeMetadata(statusMessage.metadata, task.metadata), @@ -370,10 +361,7 @@ function updatesFromTask(task: Task, observed: ObservedTaskState): AgentResponse function updatesFromStatusUpdate(event: TaskStatusUpdateEvent): AgentResponseUpdate[] { const state = event.status?.state; const statusMessage = event.status?.message; - const contents = - state === TaskState.TASK_STATE_INPUT_REQUIRED && statusMessage !== undefined - ? fromA2AParts(statusMessage.parts) - : []; + const contents = statusContents(state, statusMessage); const createdAt = omitEmpty(event.status?.timestamp); const finishReason = finishReasonFor(state); return [