diff --git a/docs/api-reference.md b/docs/api-reference.md index 8660ca4bb..124783ff3 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -47,7 +47,7 @@ Require `session_index`. Workflow identity comes from the session. |------|------------|---------|-------------| | `get_workflow` | `session_index` | Orchestrator technique bundle + workflow stubs | Orchestrator load: rules, variables, `initialActivity`, activity list. [Resolution](resource_resolution_model.md) | | `next_activity` | `session_index`, `activity_id`, manifests? | `activity_id`, `name`; trace in `_meta` | Advance to an activity (does not return its body). [Fidelity](workflow-fidelity.md) | -| `get_activity` | `session_index`, `context_tokens`, `agent_id?`, `bundle?` | Worker bundle + activity body, `_meta.dispatch` | Worker load for the current activity. `context_tokens` is required; `agent_id` scopes delivery to this worker context. [Bundling](resource_resolution_model.md#12-hybrid-technique-bundling) · [Reference delivery](resource_resolution_model.md#11-reference-delivery) | +| `get_activity` | `session_index`, `context_tokens`, `agent_id?`, `bundle?` | `batch` standing, worker bundle, activity definition; `_meta.dispatch` and `_meta.batch` | Worker load for the current activity. `context_tokens` is required; `agent_id` scopes delivery to this worker context. [Bundling](resource_resolution_model.md#12-hybrid-technique-bundling) · [Reference delivery](resource_resolution_model.md#11-reference-delivery) · [Batch bound](dispatch_model.md#batching-a-run-of-activities-407) | | `yield_checkpoint` | `session_index`, `checkpoint_id` | `yielded` or `replayed` | Pause for a user decision (or replay a prior answer). [Checkpoints](checkpoint_model.md) | | `resume_checkpoint` | `session_index` | Status | Worker continues after the checkpoint is resolved. | | `present_checkpoint` | `session_index` | Message, options, effects | Load the active checkpoint for the user. | diff --git a/docs/resource_resolution_model.md b/docs/resource_resolution_model.md index 897bea672..251cd19dd 100644 --- a/docs/resource_resolution_model.md +++ b/docs/resource_resolution_model.md @@ -184,7 +184,7 @@ A marker is only ever valid for the context that received the bytes it stands fo The server hashes each payload it delivers and records it in `session.json#deliveredContent`. It records in every mode, so a call that opts in with `bundle: "reference"` can still refer back to content that arrived under the default full mode. -Keys are namespaced by delivery channel — `bundle:*`, `technique:*`, `activity_rules:*`, `workflow_bundle:*`, `resource:*` — so a marker only ever points at content delivered through that same channel. +Keys are namespaced by delivery channel — `bundle:*`, `technique:*`, `activity_rules:*`, `activity:*`, `workflow_bundle:*`, `resource:*` — so a marker only ever points at content delivered through that same channel. The ledger is keyed on the **delivery scope**: the per-call `agent_id` when one is supplied, otherwise the session's recorded `agentId`. This matters because a dispatched worker authenticates against the orchestrator's `session_index`, and several workers can hold that index at once — the scope names the agent context a payload went to, rather than the session they share. @@ -192,7 +192,7 @@ The orchestrator mints an `agent_id` per dispatch and reuses it verbatim for as ### What collapses, call by call -- **`get_activity`** — the response carries `bundle_mode: reference` and a `bundle_note`. Any bundled technique whose composed content is byte-identical to an earlier delivery collapses to a marker, as do the `rules` and `activity_rules` blocks. Techniques new to the activity, or whose content changed, arrive in full. The activity body itself is always delivered. +- **`get_activity`** — the response carries `bundle_mode: reference` and a `bundle_note`. Any bundled technique whose composed content is byte-identical to an earlier delivery collapses to a marker, as do the `rules` and `activity_rules` blocks. Techniques new to the activity, or whose content changed, arrive in full. The activity definition collapses in parts, keyed under `activity::` — its step list, transitions, outcome and synthesised artifact contract each on their own hash, while the identity fields are never keyed. A worker confirms the returned activity id against the one it was dispatched for, so that field is present however much of the rest is a marker. - **`get_technique`** — a byte-identical refetch returns `delivery: unchanged` and a `content_hash` instead of the composed technique. Step-bound provenance annotations (`source:` / `destination:`) are part of that content. They are fixed for a given corpus and step, so refetching the same step collapses; fetching the same operation from a *different* step re-delivers in full rather than handing back a stale reference. - **`get_resource`** — a byte-identical refetch of the same `resource_id` returns `delivery: unchanged` and a `content_hash` instead of the body. The key is the caller's exact `resource_id`, anchor included, so `pr-description` and `pr-description#templates` occupy independent slots. - **`get_workflow`** — under `context_mode: "persistent"` the orchestrator ops bundle (everything above the `---` separator) is keyed under `workflow_bundle:`. On a resume where the agent already holds it, the whole bundle collapses to a single marker, while the workflow summary below the separator stays full. diff --git a/src/tools/workflow-tools.ts b/src/tools/workflow-tools.ts index 3b8d0886c..55f9ddb01 100644 --- a/src/tools/workflow-tools.ts +++ b/src/tools/workflow-tools.ts @@ -26,6 +26,7 @@ import { readdir } from 'node:fs/promises'; import { join as pathJoin } from 'node:path'; import { DEFAULT_MAX_EAGER_RESOURCE_CHARS, loadResourceDelivery } from '../utils/resource-delivery.js'; import { appendStepStartedIfAbsent } from '../utils/step-events.js'; +import { projectActivityBody } from '../utils/activity-body.js'; import { sessionIndexParam, contextTokensParam, @@ -1181,6 +1182,14 @@ export function registerWorkflowTools(server: McpServer, config: ServerConfig): ? `${activityBody}\n${stringifyForResponse({ artifacts: composedArtifacts })}` : activityBody; + // The definition is keyed in parts, not whole (#404 W10): the identity a worker checks the + // dispatched activity id against is always delivered in full, and the step list, transitions, + // outcome and synthesised artifact contract collapse to markers where this context already + // holds those bytes. Under full delivery every part ships — a definition section appears once + // in a response, so there is no earlier copy for a marker to point at. + const projectedBody = projectActivityBody(activityBodyWithArtifacts, state, scope, { readLedger: referenceMode }); + Object.assign(newDeliveries, projectedBody.newDeliveries); + // Payload-borne enforcement hints (#189 C7, R7): the enforcement model (schemas/README) lives // in docs that never ride the wire, so a payload-only reader still infers the SERVER executes // inert fields (guessing it applies `action: set`, unsure who owns auto-advance). Annotate, at @@ -1221,7 +1230,7 @@ export function registerWorkflowTools(server: McpServer, config: ServerConfig): // Assembled before the save so the dispatch event can record what this dispatch actually // cost — `chars` on an activity's fresh and resume events is the before/after measurement. - const responseBody = `${opsSection}${header}\n\n${activityRulesBlock}${enforcementBlock}${activityBodyWithArtifacts}`; + const responseBody = `${opsSection}${header}\n\n${activityRulesBlock}${enforcementBlock}${projectedBody.text}`; // Persist against a FRESH load, not the snapshot captured before composition: the session // store is last-writer-wins over the whole file, and composition awaits dozens of FS reads — @@ -1310,6 +1319,8 @@ export function registerWorkflowTools(server: McpServer, config: ServerConfig): bundled_steps_collapsed: bundledSteps.filter((b) => b.delivery === 'unchanged').length, bundled_resources: bundledResourceDeliveries.length, shared_blocks_collapsed_in_response: intraResponseCollapses, + body_fields_collapsed: projectedBody.collapsedFields.length, + body_chars_collapsed: projectedBody.collapsedChars, spent_chars: spentChars, eager_budget_chars: Math.floor(eagerBudgetChars), response_chars: responseText.length, diff --git a/src/utils/activity-body.ts b/src/utils/activity-body.ts new file mode 100644 index 000000000..b7152888e --- /dev/null +++ b/src/utils/activity-body.ts @@ -0,0 +1,151 @@ +import type { SessionFile } from '../schema/session.schema.js'; +import { contentHash, deliveredHash, unchangedMarker } from './delivery.js'; +import { stringifyForResponse } from './serialization.js'; + +/** + * The activity definition, delivered in parts (#404 W10). + * + * A delivery ends with the activity's own definition text, and every other part of the response can + * arrive as a short marker when the receiving context already holds the bytes: each bundled + * technique, the inherited rules block, each shared block of a composed technique, each eagerly + * bundled resource. The definition is keyed the same way, in parts rather than whole, because one + * part of it has to survive every collapse: a worker confirms that the activity id the server + * returned matches the one it was dispatched for, and stops without executing a step if they + * disagree. That check reads the definition. + * + * So every field the server does not key — the identity among them — is always delivered in full, in + * the position its author wrote it, and the step list, transitions, outcome and synthesised artifact + * contract are keyed separately. + * It is the treatment a composed technique already gets, where the invariant note and the item list + * are keyed apart so a shared preamble collapses even when the rest differs. + * + * Keys are content-keyed (`activity::`), so a changed section gets a different key and + * delivers in full with no invalidation logic — the same scheme as `bundle:rules:`. + */ + +/** + * Definition fields keyed separately from the identity. Each is a whole top-level block of the + * delivered YAML: `artifacts` is the contract the server synthesises from the steps' declared + * outputs and appends, and reaches a worker exactly like an authored field. + */ +export const COLLAPSIBLE_BODY_FIELDS: readonly string[] = ['steps', 'transitions', 'outcome', 'artifacts']; + +/** A top-level block of the delivered definition: its field name and its text, newline included. */ +export interface BodySection { + field: string; + text: string; +} + +/** + * One run of the delivered definition. `field` names a keyed block; its absence means text that always + * ships as authored. Parts are held in DOCUMENT ORDER, and the delivery emits them in that order — + * seven activities of the corpus carry an unkeyed field between two keyed ones, so hoisting the unkeyed + * text would hand the worker a definition whose fields are not in the order its author wrote them. + */ +export interface BodyPart { + field?: string; + text: string; +} + +/** The definition split at its top-level fields. */ +export interface SplitActivityBody { + /** Every run of the definition, in document order — the form the delivery emits. */ + parts: BodyPart[]; + /** The unkeyed text, concatenated: the identity a worker checks its dispatch against. */ + identity: string; + /** The keyed blocks, in document order. */ + sections: BodySection[]; +} + +/** A line opening a top-level YAML field, which is where one block ends and the next begins. */ +const TOP_LEVEL_FIELD = /^([A-Za-z$][A-Za-z0-9_-]*):/; + +/** + * Split the delivered definition text into runs at its top-level fields, in document order. Each keyed + * field carries its own block through to the next top-level field; consecutive unkeyed fields share a + * run. A field the server does not key is delivered as authored rather than dropped, so an unrecognised + * definition field reaches the worker whole. + */ +export function splitActivityBody(body: string): SplitActivityBody { + const parts: Array<{ field?: string; lines: string[] }> = []; + /** The run being accumulated. A new one opens at every top-level field that changes keyed-ness. */ + let current: { field?: string; lines: string[] } | undefined; + + const open = (field: string | undefined, line: string): void => { + current = field === undefined ? { lines: [line] } : { field, lines: [line] }; + parts.push(current); + }; + + for (const line of body.split('\n')) { + const opened = TOP_LEVEL_FIELD.exec(line); + if (opened) { + const field = opened[1]!; + if (COLLAPSIBLE_BODY_FIELDS.includes(field)) { open(field, line); continue; } + // An unkeyed field extends the run before it only when that run is unkeyed too; after a keyed + // block it opens a run of its own, so document order survives. + if (current && current.field === undefined) current.lines.push(line); + else open(undefined, line); + continue; + } + // A continuation line belongs to whatever run is open — a folded description, a nested list, a + // comment. A body opening with one has no run yet, so it opens an unkeyed one. + if (current) current.lines.push(line); + else open(undefined, line); + } + + const resolved: BodyPart[] = parts.map((p) => ( + p.field === undefined ? { text: p.lines.join('\n') } : { field: p.field, text: p.lines.join('\n') } + )); + return { + parts: resolved, + identity: resolved.filter((p) => p.field === undefined).map((p) => p.text).join('\n'), + sections: resolved.filter((p): p is BodySection => p.field !== undefined), + }; +} + +/** What a delivery of the definition carried, for the caller to report and to record. */ +export interface ProjectedActivityBody { + /** The definition text to send. */ + text: string; + /** Ledger entries for the sections sent in full, to commit with the rest of the delivery. */ + newDeliveries: Record; + /** Fields that arrived as markers. */ + collapsedFields: string[]; + /** Characters those markers stand for. */ + collapsedChars: number; +} + +/** + * The definition as this delivery sends it: identity in full, and each keyed section either in full + * or as an unchanged marker where this scope already holds those bytes. + * + * `readLedger: false` sends everything in full, which is what a forced full delivery and a freshly + * spawned worker get — a marker is unreadable to a context that never received the bytes, and unlike + * a technique's shared blocks a definition section appears once in a response, so there is no earlier + * copy in the same payload for a marker to point at. + */ +export function projectActivityBody( + body: string, + state: SessionFile, + scope: string, + opts: { readLedger: boolean }, +): ProjectedActivityBody { + const newDeliveries: Record = {}; + const collapsedFields: string[] = []; + let collapsedChars = 0; + + const emitted = splitActivityBody(body).parts.map((part) => { + if (part.field === undefined) return part.text; + const hash = contentHash(part.text); + const key = `activity:${part.field}:${hash}`; + if (opts.readLedger && deliveredHash(state, key, scope) === hash) { + collapsedFields.push(part.field); + collapsedChars += part.text.length; + return stringifyForResponse({ [part.field]: unchangedMarker(hash) }); + } + newDeliveries[key] = hash; + return part.text; + }); + + return { text: emitted.join('\n'), newDeliveries, collapsedFields, collapsedChars }; +} diff --git a/src/utils/delivery.ts b/src/utils/delivery.ts index 04ee2beff..f26aa9800 100644 --- a/src/utils/delivery.ts +++ b/src/utils/delivery.ts @@ -32,6 +32,10 @@ import { stringifyForResponse } from './serialization.js'; * - `bundle:` — one composed technique in the `get_activity` bundle * - `bundle:rules:` — the `get_activity` rules bundle * - `activity_rules:` — the inherited worker rules block + * - `activity::` — one top-level block of the delivered activity + * definition (`steps`, `transitions`, `outcome`, `artifacts`); its identity + * fields are never keyed, so the dispatched-activity check always has an id + * to read (see `activity-body.ts`) * - `technique:` — a full `get_technique` composed payload * - `technique::` — one shared block (`inherited_inputs` / * `inherited_outputs` / `rules`) of a composed technique diff --git a/tests/activity-body-delivery.test.ts b/tests/activity-body-delivery.test.ts new file mode 100644 index 000000000..16ee7358b --- /dev/null +++ b/tests/activity-body-delivery.test.ts @@ -0,0 +1,201 @@ +/** + * The activity definition collapses in parts, and its identity never does (#404 W10). + * + * Everything else a delivery carries can arrive as a marker when the receiving context already holds + * the bytes. The definition is keyed the same way, in parts, because a worker confirms the returned + * activity id against the one it was dispatched for and stops without executing a step if they + * disagree — so the identity has to survive whatever else collapses. + */ +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { existsSync, readFileSync, readdirSync } from 'node:fs'; +import { join } from 'node:path'; +import { parse } from 'yaml'; +import { corpusRoot } from './corpus-root.js'; +import { splitActivityBody, projectActivityBody, COLLAPSIBLE_BODY_FIELDS } from '../src/utils/activity-body.js'; +import { contentHash } from '../src/utils/delivery.js'; +import type { SessionFile } from '../src/schema/session.schema.js'; +import { createHarness, rawText, isError, parseToolResponse, type Harness } from './e2e/harness.js'; + +const BODY = [ + 'id: start-work-package', + 'version: 3.15.0', + 'name: Start Work Package', + 'description: Initialize the work package.', + 'required: true', + 'steps:', + ' - kind: action', + ' id: announce-start', + 'transitions:', + ' - to: design-philosophy', + 'outcome:', + ' - The work package has an identity', + 'artifacts:', + ' - name: 01-intake.md', +].join('\n'); + +/** A session whose ledger already holds `entries` under `scope`. */ +function sessionHolding(scope: string, entries: Record): SessionFile { + return { agentId: scope, deliveredContent: { [scope]: entries } } as unknown as SessionFile; +} + +describe('splitActivityBody', () => { + it('keeps every scalar field with the identity and separates each collapsible block', () => { + const { identity, sections } = splitActivityBody(BODY); + expect(identity).toContain('id: start-work-package'); + expect(identity).toContain('version: 3.15.0'); + expect(identity).toContain('name: Start Work Package'); + expect(identity).not.toContain('kind: action'); + expect(sections.map((s) => s.field)).toEqual(['steps', 'transitions', 'outcome', 'artifacts']); + expect(sections[0]!.text).toContain('id: announce-start'); + }); + + it('loses nothing: the parts rejoin into the definition they came from', () => { + expect(splitActivityBody(BODY).parts.map((p) => p.text).join('\n')).toBe(BODY); + }); + + it('keeps an unrecognised top-level field rather than dropping it', () => { + const withExtra = `${BODY}\nbundleTechniques:\n maxChars: 0`; + const { identity, sections } = splitActivityBody(withExtra); + expect(identity).toContain('bundleTechniques:'); + expect(identity).toContain(' maxChars: 0'); + expect(sections.map((s) => s.field)).toEqual(COLLAPSIBLE_BODY_FIELDS); + }); + + it('keeps an unkeyed field where its author put it, between two keyed ones', () => { + // Seven activities of the corpus carry `decisions:` between `steps` and `transitions`. Hoisting it + // to the front of the delivery would hand the worker a definition whose fields are not in the order + // they were authored in. + const withMiddle = [ + 'id: mid', 'version: 1.0.0', 'steps:', ' - kind: action', + 'decisions:', ' - id: d1', 'transitions:', ' - to: next', + ].join('\n'); + const { parts } = splitActivityBody(withMiddle); + expect(parts.map((p) => p.field ?? '-')).toEqual(['-', 'steps', '-', 'transitions']); + expect(parts.map((p) => p.text).join('\n')).toBe(withMiddle); + }); + + it('round-trips every activity of the corpus', () => { + const root = corpusRoot(); + let scanned = 0; + for (const workflow of readdirSync(root)) { + const dir = join(root, workflow, 'activities'); + if (!existsSync(dir)) continue; + for (const file of readdirSync(dir).filter((f) => f.endsWith('.yaml'))) { + const body = readFileSync(join(dir, file), 'utf8'); + scanned += 1; + expect( + splitActivityBody(body).parts.map((p) => p.text).join('\n'), + `${workflow}/${file} does not round-trip: the delivered definition differs from the authored one`, + ).toBe(body); + } + } + // A corpus that scanned nothing would pass every assertion above it. + expect(scanned).toBeGreaterThan(50); + }); +}); + +describe('projectActivityBody', () => { + it('sends every part in full when the context holds nothing, and stages a key for each', () => { + const projected = projectActivityBody(BODY, sessionHolding('w', {}), 'w', { readLedger: true }); + expect(projected.text).toBe(BODY); + expect(projected.collapsedFields).toEqual([]); + expect(Object.keys(projected.newDeliveries).sort()).toEqual( + COLLAPSIBLE_BODY_FIELDS.map((f) => `activity:${f}:${contentHash(splitActivityBody(BODY).sections.find((s) => s.field === f)!.text)}`).sort(), + ); + }); + + it('collapses a section this context already holds, and keeps the identity in full', () => { + const { sections } = splitActivityBody(BODY); + const held = Object.fromEntries(sections.map((s) => [`activity:${s.field}:${contentHash(s.text)}`, contentHash(s.text)])); + const projected = projectActivityBody(BODY, sessionHolding('w', held), 'w', { readLedger: true }); + + expect(projected.collapsedFields).toEqual(COLLAPSIBLE_BODY_FIELDS); + // The id survives a fully collapsed definition — the dispatched-activity check still has + // something to read. + const parsed = parse(projected.text) as Record; + expect(parsed['id']).toBe('start-work-package'); + expect(parsed['version']).toBe('3.15.0'); + for (const field of COLLAPSIBLE_BODY_FIELDS) { + expect(parsed[field]).toMatchObject({ delivery: 'unchanged' }); + } + // None of the collapsed content is in the payload. A marker has a floor of its own — on a + // definition this small the four of them cost more than the lines they replace, which is why the + // size claim is measured over the real corpus below rather than over this fixture. + expect(projected.text).not.toContain('id: announce-start'); + expect(projected.text).not.toContain('to: design-philosophy'); + expect(projected.collapsedChars).toBe( + splitActivityBody(BODY).sections.reduce((total, s) => total + s.text.length, 0), + ); + }); + + it('sends everything in full when the ledger is not consulted', () => { + const { sections } = splitActivityBody(BODY); + const held = Object.fromEntries(sections.map((s) => [`activity:${s.field}:${contentHash(s.text)}`, contentHash(s.text)])); + const projected = projectActivityBody(BODY, sessionHolding('w', held), 'w', { readLedger: false }); + expect(projected.text).toBe(BODY); + expect(projected.collapsedFields).toEqual([]); + }); + + it('delivers a changed section in full while its siblings collapse', () => { + const { sections } = splitActivityBody(BODY); + const held = Object.fromEntries( + sections.filter((s) => s.field !== 'steps').map((s) => [`activity:${s.field}:${contentHash(s.text)}`, contentHash(s.text)]), + ); + const projected = projectActivityBody(BODY, sessionHolding('w', held), 'w', { readLedger: true }); + expect(projected.collapsedFields).toEqual(['transitions', 'outcome', 'artifacts']); + expect(projected.text).toContain('id: announce-start'); + }); +}); + +describe('a resumed re-request over the real corpus', () => { + let h: Harness; + let sessionIndex: string; + + beforeAll(async () => { + h = await createHarness(); + const started = await h.client.callTool({ + name: 'start_session', + arguments: { + workflow_id: 'work-package', agent_id: 'orchestrator', + planning_folder: join(h.workspaceDir, '.engineering/artifacts/planning', 'activity-body'), + }, + }); + if (isError(started)) throw new Error(rawText(started)); + sessionIndex = parseToolResponse(started).session_index as string; + const entered = await h.client.callTool({ + name: 'next_activity', arguments: { session_index: sessionIndex, activity_id: 'start-work-package' }, + }); + if (isError(entered)) throw new Error(rawText(entered)); + }); + + afterAll(async () => { await h?.close(); }); + + const take = (agentId: string, extra: Record = {}) => + h.client.callTool({ + name: 'get_activity', + arguments: { session_index: sessionIndex, context_tokens: 200000, agent_id: agentId, ...extra }, + }); + + it('returns the identity in full and the remainder as markers, and shrinks the delivery', async () => { + const first = await take('resumed-worker'); + expect(isError(first)).toBe(false); + const firstChars = rawText(first).length; + + const second = await take('resumed-worker', { bundle: 'reference' }); + expect(isError(second)).toBe(false); + const secondText = rawText(second); + + // The id the worker checks its dispatch against is present on the collapsed delivery. + expect(secondText).toContain('id: start-work-package'); + expect(secondText).toMatch(/steps:\n\s+delivery: unchanged/); + expect(secondText.length).toBeLessThan(firstChars); + }); + + it('leaves a forced full delivery carrying the whole definition', async () => { + await take('forced-worker'); + const forced = await take('forced-worker', { bundle: 'full' }); + const text = rawText(forced); + expect(text).toContain('id: start-work-package'); + expect(text).not.toMatch(/steps:\n\s+delivery: unchanged/); + }); +});