diff --git a/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts b/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts index a2d24ccae1..fa854d3cd8 100644 --- a/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts +++ b/apps/desktop/src/main/__tests__/desktop-session-projection.test.ts @@ -24,6 +24,7 @@ import type { SessionSummary } from '@maka/core/session'; import { projectDesktopSessionEvent, projectDesktopSessionSummary, + projectDesktopStoredMessage, projectDesktopTurnRecord, } from '../../shared/desktop-session-projection.js'; @@ -145,6 +146,32 @@ test('projects queued Session attachments into the Desktop host namespace', () = ); }); +test('projects durable WorkHub delegation targets into the Desktop host namespace', () => { + const projected = projectDesktopStoredMessage( + { hostId: 'remote-root' }, + { + type: 'workhub_coordination', + id: 'delegation-commit-message', + turnId: 'coordination-turn', + ts: 2, + schemaVersion: 1, + kind: 'delegation_committed', + actionId: 'action-id', + actionFingerprint: `sha256:${'a'.repeat(64)}`, + coordinationTurnId: 'coordination-turn', + targetSessionId: 'payments', + disposition: 'delegate_existing', + delegationId: 'delegation-id', + targetTurnId: 'payments-turn', + }, + ); + + assert.equal(projected.type, 'workhub_coordination'); + if (projected.type === 'workhub_coordination') { + assert.equal(projected.targetSessionId, JSON.stringify(['remote-root', 'payments'])); + } +}); + function summary(id: string): SessionSummary { return { id, diff --git a/apps/desktop/src/shared/desktop-session-projection.ts b/apps/desktop/src/shared/desktop-session-projection.ts index c2c4b0a0a6..1f04b7b54e 100644 --- a/apps/desktop/src/shared/desktop-session-projection.ts +++ b/apps/desktop/src/shared/desktop-session-projection.ts @@ -124,6 +124,11 @@ export function projectDesktopStoredMessage( return message.parentSessionId ? { ...message, parentSessionId: projectSessionId(host, message.parentSessionId) } : message; + case 'workhub_coordination': + return { + ...message, + targetSessionId: projectSessionId(host, message.targetSessionId), + }; default: return message; } diff --git a/docs/architecture/workhub-coordination-session-adr.md b/docs/architecture/workhub-coordination-session-adr.md index 18db78e671..2ed812375a 100644 --- a/docs/architecture/workhub-coordination-session-adr.md +++ b/docs/architecture/workhub-coordination-session-adr.md @@ -114,6 +114,15 @@ authoritative result. WorkHub may display a bounded projection or record a coordination summary, but it does not copy the ordinary Session's complete transcript into the Coordination Session. +Delegation linkage uses closed, typed `workhub_coordination` records in the +existing Coordination Session transcript. An immutable `delegation_intent` is +appended before the target Session effect so an opaque candidate remains +recoverable after the candidate set changes or the Runtime Host restarts. A +`delegation_committed` record then binds that intent to the accepted target Turn +and acts as the durable action-replay result. The records carry an action +fingerprint to reject conflicting reuse of an action identity. They do not form a +general workflow state machine and do not persist target execution lifecycle. + ## Consequences, costs, and reevaluation - WorkHub gains persistent conversational continuity without adding another @@ -129,9 +138,10 @@ transcript into the Coordination Session. entity remains unresolved. - Cross-Runtime-Host coordination remains deferred. - Coordination Session role representation, lazy creation, durable lookup, - recovery, and per-Host UI resolution are implemented. Coordination transcript, - disposition, delegation-link, and Action Gate behavior remain later work; this - ADR defines their authority boundaries without implementing them. + recovery, per-Host UI resolution, persistent transcript, closed dispositions, + and the Action Gate are implemented. Durable delegation linkage is encoded in + that transcript; target lifecycle projection and destructive replacement/Stop + recovery remain later work. Reevaluate the per-Host decision if supported workflows require one WorkHub conversation to coordinate ordinary Sessions on multiple Runtime Hosts, or if Host diff --git a/packages/core/src/__tests__/workhub-coordination-record.test.ts b/packages/core/src/__tests__/workhub-coordination-record.test.ts new file mode 100644 index 0000000000..522c0af041 --- /dev/null +++ b/packages/core/src/__tests__/workhub-coordination-record.test.ts @@ -0,0 +1,81 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import { decodeCanonicalMessage } from '../session.js'; + +const FINGERPRINT = `sha256:${'a'.repeat(64)}`; + +describe('WorkHub Coordination stored records', () => { + test('decodes exact delegation intent and commit records', () => { + const intent = { + type: 'workhub_coordination', + id: 'intent-id', + turnId: 'coordination-turn', + ts: 1, + schemaVersion: 1, + kind: 'delegation_intent', + actionId: 'action-id', + actionFingerprint: FINGERPRINT, + coordinationTurnId: 'coordination-turn', + targetSessionId: 'payments', + disposition: 'delegate_existing', + } as const; + const committed = { + ...intent, + id: 'commit-id', + ts: 2, + kind: 'delegation_committed', + delegationId: 'delegation-id', + targetTurnId: 'target-turn', + steered: true, + } as const; + + assert.deepEqual(decodeCanonicalMessage(intent), intent); + assert.deepEqual(decodeCanonicalMessage(committed), committed); + }); + + test('rejects malformed or widened coordination records', () => { + const base = { + type: 'workhub_coordination', + id: 'intent-id', + turnId: 'coordination-turn', + ts: 1, + schemaVersion: 1, + kind: 'delegation_intent', + actionId: 'action-id', + actionFingerprint: FINGERPRINT, + coordinationTurnId: 'coordination-turn', + targetSessionId: 'payments', + disposition: 'delegate_existing', + } as const; + + for (const invalid of [ + { ...base, coordinationTurnId: 'different-turn' }, + { ...base, actionFingerprint: 'not-a-digest' }, + { ...base, disposition: 'replace' }, + { ...base, sourceSessionId: 'injected' }, + { ...base, kind: 'delegation_committed' }, + { ...base, schemaVersion: 2 }, + ]) { + assert.throws(() => decodeCanonicalMessage(invalid), /Invalid stored message schema/u); + } + }); +}); diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index e9896dc2e2..f7f19986d1 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -756,6 +756,7 @@ export type StoredMessage = | PermissionDecisionMessage | TokenUsageMessage | TurnStateMessage + | WorkHubCoordinationMessage | SystemNoteMessage; export interface UserMessage extends MessageContent { @@ -908,6 +909,41 @@ export interface TurnStateMessage { partialOutputRetained: boolean; } +export const WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION = 1 as const; + +export type WorkHubDelegationDisposition = 'delegate_existing' | 'create_new'; + +interface WorkHubCoordinationMessageEnvelope { + type: 'workhub_coordination'; + id: string; + /** The Coordination Turn that owns this action. */ + turnId: string; + ts: number; + schemaVersion: typeof WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION; + actionId: string; + actionFingerprint: `sha256:${string}`; + coordinationTurnId: string; + targetSessionId: string; + disposition: WorkHubDelegationDisposition; +} + +/** Durable target choice written before a delegated Session effect is attempted. */ +export interface WorkHubDelegationIntentMessage extends WorkHubCoordinationMessageEnvelope { + kind: 'delegation_intent'; +} + +/** Durable proof that one Coordination action owns one accepted target Turn. */ +export interface WorkHubDelegationCommittedMessage extends WorkHubCoordinationMessageEnvelope { + kind: 'delegation_committed'; + delegationId: string; + targetTurnId: string; + steered?: true; +} + +export type WorkHubCoordinationMessage = + | WorkHubDelegationIntentMessage + | WorkHubDelegationCommittedMessage; + export interface TurnRecord { turnId: string; firstSequence?: number; @@ -1030,6 +1066,41 @@ const TURN_STATE_MESSAGE_SHAPE = defineObjectShape()( 'errorClass', ], ); +const WORKHUB_DELEGATION_INTENT_MESSAGE_SHAPE = defineObjectShape()( + [ + 'type', + 'id', + 'turnId', + 'ts', + 'schemaVersion', + 'kind', + 'actionId', + 'actionFingerprint', + 'coordinationTurnId', + 'targetSessionId', + 'disposition', + ], + [], +); +const WORKHUB_DELEGATION_COMMITTED_MESSAGE_SHAPE = + defineObjectShape()( + [ + 'type', + 'id', + 'turnId', + 'ts', + 'schemaVersion', + 'kind', + 'actionId', + 'actionFingerprint', + 'coordinationTurnId', + 'targetSessionId', + 'disposition', + 'delegationId', + 'targetTurnId', + ], + ['steered'], + ); const SYSTEM_NOTE_MESSAGE_SHAPE = defineObjectShape()( ['type', 'id', 'ts', 'kind'], ['turnId', 'data'], @@ -1177,6 +1248,11 @@ function decodeMessage( ) return message as unknown as TurnStateMessage; break; + case 'workhub_coordination': + if (isWorkHubCoordinationMessage(message)) { + return message as unknown as WorkHubCoordinationMessage; + } + break; case 'system_note': if ( hasExactShape(message, SYSTEM_NOTE_MESSAGE_SHAPE) && @@ -1190,6 +1266,30 @@ function decodeMessage( throw new Error('Invalid stored message schema'); } +function isWorkHubCoordinationMessage(message: Record): boolean { + const common = + hasMessageEnvelope(message, true) && + message.schemaVersion === WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION && + typeof message.actionId === 'string' && + typeof message.actionFingerprint === 'string' && + /^sha256:[a-f0-9]{64}$/u.test(message.actionFingerprint) && + typeof message.coordinationTurnId === 'string' && + message.turnId === message.coordinationTurnId && + typeof message.targetSessionId === 'string' && + (message.disposition === 'delegate_existing' || message.disposition === 'create_new'); + if (!common) return false; + if (message.kind === 'delegation_intent') { + return hasExactShape(message, WORKHUB_DELEGATION_INTENT_MESSAGE_SHAPE); + } + return ( + message.kind === 'delegation_committed' && + hasExactShape(message, WORKHUB_DELEGATION_COMMITTED_MESSAGE_SHAPE) && + typeof message.delegationId === 'string' && + typeof message.targetTurnId === 'string' && + (message.steered === undefined || message.steered === true) + ); +} + function decodeStoredMessageContent( value: unknown, decodeToolResultContent: (content: unknown) => ToolResultContent, diff --git a/packages/core/src/thread-search.ts b/packages/core/src/thread-search.ts index 881124d3e5..f09df85a39 100644 --- a/packages/core/src/thread-search.ts +++ b/packages/core/src/thread-search.ts @@ -465,6 +465,7 @@ export function threadSearchMatchKind(message: StoredMessage): ThreadSearchMatch case 'permission_decision': case 'token_usage': case 'turn_state': + case 'workhub_coordination': case 'system_note': throw new Error(`Message type ${message.type} is not searchable`); } @@ -488,6 +489,8 @@ export function formatSearchResultSummary(message: StoredMessage): string { return '用量记录'; case 'turn_state': return '回合状态'; + case 'workhub_coordination': + return 'WorkHub 协调记录'; case 'system_note': return '系统记录'; } @@ -543,6 +546,7 @@ export function collectSearchableText(message: StoredMessage): string | undefine case 'permission_decision': case 'token_usage': case 'turn_state': + case 'workhub_coordination': case 'system_note': // Excluded — not user-typed / not user-visible content. return undefined; diff --git a/packages/runtime-host/src/__tests__/workhub-coordination-action-gate.test.ts b/packages/runtime-host/src/__tests__/workhub-coordination-action-gate.test.ts index 9d36dae540..b50996e300 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-action-gate.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-action-gate.test.ts @@ -25,6 +25,9 @@ import { WorkHubCoordinationActionGate, type WorkHubActionGateEffects, type WorkHubActionGateSession, + type WorkHubDelegationCommit, + type WorkHubDelegationIntent, + type WorkHubDelegationRecord, } from '../server/workhub-coordination-action-gate.js'; import type { ConnectionContext } from '../server/operation-dispatcher.js'; @@ -202,8 +205,10 @@ describe('WorkHub Coordination Action Gate', () => { const first = await gate.act(input, CONTEXT); const replay = await gate.act(input, CONTEXT); + const restartedReplay = await new WorkHubCoordinationActionGate(effects).act(input, CONTEXT); assert.deepEqual(replay, first); + assert.deepEqual(restartedReplay, first); assert.equal(effects.creations.length, 1); assert.equal(effects.submissions.length, 1); assert.match(effects.creations[0]?.sessionId ?? '', /^whs_[a-f0-9]{48}$/u); @@ -223,7 +228,7 @@ describe('WorkHub Coordination Action Gate', () => { assert.equal(effects.creations.length, 1); }); - test('effect rejection grants no root ownership and releases the action identity', async () => { + test('effect rejection grants no root ownership and lets the durable intent retry', async () => { const effects = fakeEffects([session('ordinary')]); const gate = new WorkHubCoordinationActionGate(effects); const snapshot = await gate.candidates(); @@ -249,7 +254,31 @@ describe('WorkHub Coordination Action Gate', () => { assert.equal((await gate.act(input, CONTEXT)).disposition, 'delegate_existing'); }); - test('replays an ordinary delegation without submitting twice', async () => { + test('commits a delegation after recovering an unknown submit outcome', async () => { + const effects = fakeEffects([session('ordinary')]); + const gate = new WorkHubCoordinationActionGate(effects); + const snapshot = await gate.candidates(); + effects.submitUnknownAfterAdmission = true; + + const result = await gate.act( + { + actionId: 'unknown-submit-action', + userText: 'Continue ordinary work', + candidateSetId: snapshot.candidateSetId, + proposal: { + disposition: 'delegate_existing', + candidateRef: snapshot.candidates[0]!.candidateRef, + }, + }, + CONTEXT, + ); + + assert.equal(result.disposition, 'delegate_existing'); + assert.equal(effects.submissions.length, 1); + assert.equal(effects.delegations.get('unknown-submit-action')?.kind, 'delegation_committed'); + }); + + test('replays an ordinary delegation durably across Action Gate restart', async () => { const effects = fakeEffects([session('ordinary')]); const gate = new WorkHubCoordinationActionGate(effects); const snapshot = await gate.candidates(); @@ -264,11 +293,52 @@ describe('WorkHub Coordination Action Gate', () => { }; const first = await gate.act(input, CONTEXT); - const replay = await gate.act(input, CONTEXT); + const replay = await new WorkHubCoordinationActionGate(effects).act(input, CONTEXT); assert.deepEqual(replay, first); assert.equal(effects.submissions.length, 1); assert.equal(effects.submissions[0]?.sessionId, 'ordinary'); + assert.equal(effects.delegations.get(input.actionId)?.kind, 'delegation_committed'); + + await assert.rejects( + new WorkHubCoordinationActionGate(effects).act( + { ...input, userText: 'Different work' }, + CONTEXT, + ), + (error) => error instanceof WorkHubActionGateFailure && error.code === 'action_conflict', + ); + assert.equal(effects.submissions.length, 1); + }); + + test('resumes a durable intent after restart without re-admitting stale candidates', async () => { + const effects = fakeEffects([session('ordinary')]); + const gate = new WorkHubCoordinationActionGate(effects); + const snapshot = await gate.candidates(); + const input = { + actionId: 'interrupted-delegate-action', + userText: 'Continue ordinary work', + candidateSetId: snapshot.candidateSetId, + proposal: { + disposition: 'delegate_existing' as const, + candidateRef: snapshot.candidates[0]!.candidateRef, + }, + }; + effects.commitFailuresRemaining = 1; + + await assert.rejects( + gate.act(input, CONTEXT), + (error) => + error instanceof WorkHubActionEffectFailure && error.code === 'commit_outcome_unknown', + ); + assert.equal(effects.delegations.get(input.actionId)?.kind, 'delegation_intent'); + assert.equal(effects.submissions.length, 1); + + effects.sessions[0] = session('ordinary', { statusUpdatedAt: 99 }); + const recovered = await new WorkHubCoordinationActionGate(effects).act(input, CONTEXT); + + assert.equal(recovered.disposition, 'delegate_existing'); + assert.equal(effects.submissions.length, 1); + assert.equal(effects.delegations.get(input.actionId)?.kind, 'delegation_committed'); }); }); @@ -290,6 +360,13 @@ function session( } function fakeEffects(initialSessions: WorkHubActionGateSession[]) { + const submitted = new Map< + string, + { + readonly input: { sessionId: string; messageId: string; text: string }; + readonly turnId: string; + } + >(); const state = { sessions: [...initialSessions], answers: [] as Array<{ turnId: string; text: string }>, @@ -304,6 +381,9 @@ function fakeEffects(initialSessions: WorkHubActionGateSession[]) { title: string; }>, submissions: [] as Array<{ sessionId: string; messageId: string; text: string }>, + delegations: new Map(), + commitFailuresRemaining: 0, + submitUnknownAfterAdmission: false as boolean, async listSessions() { return this.sessions; }, @@ -318,11 +398,56 @@ function fakeEffects(initialSessions: WorkHubActionGateSession[]) { workspace: { kind: 'project'; projectId: string } | { kind: 'host_path'; path: string }; title: string; }) { - this.creations.push(input); + if (!this.creations.some(({ sessionId }) => sessionId === input.sessionId)) { + this.creations.push(input); + } }, async submit(input: { sessionId: string; messageId: string; text: string }) { + const existing = submitted.get(input.messageId); + if (existing) { + assert.deepEqual(existing.input, input); + return { turnId: existing.turnId }; + } this.submissions.push(input); - return { turnId: `turn-${input.sessionId}` }; + const turnId = `turn-${input.sessionId}`; + submitted.set(input.messageId, { input, turnId }); + if (this.submitUnknownAfterAdmission) { + this.submitUnknownAfterAdmission = false; + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'Target submit outcome is unknown', + ); + } + return { turnId }; + }, + async recoverSubmission(input: { sessionId: string; messageId: string; text: string }) { + const existing = submitted.get(input.messageId); + if (!existing) return undefined; + assert.deepEqual(existing.input, input); + return { turnId: existing.turnId }; + }, + async readDelegation(actionId: string) { + return this.delegations.get(actionId); + }, + async prepareDelegation(intent: WorkHubDelegationIntent) { + const existing = this.delegations.get(intent.actionId); + if (existing) { + assert.deepEqual(existing, intent); + return; + } + this.delegations.set(intent.actionId, intent); + }, + async commitDelegation(commit: WorkHubDelegationCommit) { + const existing = this.delegations.get(commit.actionId); + assert.equal(existing?.kind, 'delegation_intent'); + if (this.commitFailuresRemaining > 0) { + this.commitFailuresRemaining -= 1; + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'Delegation commit outcome is unknown', + ); + } + this.delegations.set(commit.actionId, commit); }, } satisfies WorkHubActionGateEffects & { sessions: WorkHubActionGateSession[]; @@ -334,6 +459,9 @@ function fakeEffects(initialSessions: WorkHubActionGateSession[]) { title: string; }>; submissions: Array<{ sessionId: string; messageId: string; text: string }>; + delegations: Map; + commitFailuresRemaining: number; + submitUnknownAfterAdmission: boolean; }; return state; } diff --git a/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts index d78d325d19..02f7c21020 100644 --- a/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/workhub-coordination-coordinator.test.ts @@ -38,6 +38,7 @@ import type { ConnectionContext } from '../server/operation-dispatcher.js'; import type { RootTurnCoordinator } from '../server/root-turn-coordinator.js'; import { SessionAdmissionGate } from '../server/session-admission-gate.js'; import { SessionOperationFailure } from '../server/session-catalog-coordinator.js'; +import type { WorkHubActionGateEffects } from '../server/workhub-coordination-action-gate.js'; import { HostWorkHubCoordinationCoordinator, type CoordinationCreateTarget, @@ -480,6 +481,106 @@ describe('Host WorkHub Coordination coordinator', () => { } }); + test('persists delegated action ownership and replays it after Host restart', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-workhub-delegation-')); + let store = createSessionStore(root); + try { + await store.create({ + cwd: root, + name: 'Payments', + llmConnectionSlug: 'test-connection', + model: 'test-model', + permissionMode: 'ask', + }); + const submissions: Array<{ sessionId: string; messageId: string; text: string }> = []; + const first = coordinator(root, store, () => undefined, undefined, undefined, undefined, { + create: async () => undefined, + submit: async (input) => { + submissions.push(input); + return { turnId: 'payments-turn' }; + }, + recoverSubmission: async () => undefined, + }); + assert.equal((await first.handlers['workhub.coordination.resolve']({}, CONTEXT)).ok, true); + const candidates = await first.handlers['workhub.coordination.candidates']({}, CONTEXT); + assert.equal(candidates.ok, true); + if (!candidates.ok) return; + const input = { + actionId: 'payments-action', + userText: 'Continue payment work', + candidateSetId: candidates.result.candidateSetId, + proposal: { + disposition: 'delegate_existing' as const, + candidateRef: candidates.result.candidates[0]!.candidateRef, + }, + }; + const admitted = await first.handlers['workhub.coordination.act'](input, CONTEXT); + assert.deepEqual(admitted, { + ok: true, + result: { + disposition: 'delegate_existing', + targetSessionId: candidates.result.candidates[0]!.sessionId, + targetTurnId: 'payments-turn', + }, + }); + assert.deepEqual( + (await store.readMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID)) + .filter((message) => message.type === 'workhub_coordination') + .map(({ kind, actionId, targetSessionId }) => ({ kind, actionId, targetSessionId })), + [ + { + kind: 'delegation_intent', + actionId: 'payments-action', + targetSessionId: candidates.result.candidates[0]!.sessionId, + }, + { + kind: 'delegation_committed', + actionId: 'payments-action', + targetSessionId: candidates.result.candidates[0]!.sessionId, + }, + ], + ); + assert.equal(submissions.length, 1); + } finally { + await store.close?.(); + } + + store = createSessionStore(root); + try { + const restarted = coordinator(root, store, () => undefined, undefined, undefined, undefined, { + create: async () => assert.fail('durable replay must not create a Session'), + submit: async () => assert.fail('durable replay must not submit another Turn'), + recoverSubmission: async () => + assert.fail('durable replay must not recover an already committed Turn'), + }); + const candidates = await restarted.handlers['workhub.coordination.candidates']({}, CONTEXT); + assert.equal(candidates.ok, true); + if (!candidates.ok) return; + const replayed = await restarted.handlers['workhub.coordination.act']( + { + actionId: 'payments-action', + userText: 'Continue payment work', + candidateSetId: candidates.result.candidateSetId, + proposal: { + disposition: 'delegate_existing', + candidateRef: candidates.result.candidates[0]!.candidateRef, + }, + }, + CONTEXT, + ); + assert.equal(replayed.ok, true); + if (replayed.ok) { + assert.equal(replayed.result.disposition, 'delegate_existing'); + if (replayed.result.disposition === 'delegate_existing') { + assert.equal(replayed.result.targetTurnId, 'payments-turn'); + } + } + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + test('refuses to merge a Turn identity shared across answer and record', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-workhub-turn-identity-')); const store = createSessionStore(root); @@ -614,6 +715,11 @@ function coordinator( hasRootTurnAdmission: async () => false, }, admission: SessionAdmissionGate = new SessionAdmissionGate(), + sessionActions: Pick = { + create: async () => undefined, + submit: async ({ sessionId }) => ({ turnId: `turn-${sessionId}` }), + recoverSubmission: async () => undefined, + }, ) { return new HostWorkHubCoordinationCoordinator({ stateRoot: root, @@ -621,10 +727,7 @@ function coordinator( admission, continuity: { refreshCanonical: async () => undefined }, executions, - sessionActions: { - create: async () => undefined, - submit: async ({ sessionId }) => ({ turnId: `turn-${sessionId}` }), - }, + sessionActions, resolveCreateTarget: resolveCreateTarget ?? (async () => ({ diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index cf46c22b9c..f29ea7830f 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -18,7 +18,7 @@ */ import { createHash, randomUUID } from 'node:crypto'; -import { normalizeMessageContent } from '@maka/core/events'; +import { messageContentDigest, normalizeMessageContent } from '@maka/core/events'; import { describeChatConfigurationReason, NO_REAL_CONNECTION_CODE, @@ -1264,9 +1264,69 @@ export async function createExecutionRuntimeHostComposition( outcome.error.message, ); } - return outcome.result.disposition === 'turn_started' - ? { turnId: outcome.result.turnId } - : { turnId: input.messageId, steered: true as const }; + if (outcome.result.disposition === 'turn_started') { + return { turnId: outcome.result.turnId }; + } + try { + const admission = await stores.sessionStore.readMessageAdmission( + input.sessionId, + input.messageId, + ); + if (admission) return { turnId: admission.turnId, steered: true as const }; + } catch { + // The submit already settled; losing its exact Turn identity makes + // the WorkHub linkage outcome uncertain rather than retryable. + } + context.requestDrain(); + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'WorkHub target Turn identity could not be proven', + ); + }, + recoverSubmission: async (input) => { + const expectedDigest = messageContentDigest( + normalizeMessageContent({ text: input.text }), + ); + const receipt = await stores.agentRunStore.readRootTurnSourceMessageReceipt( + input.sessionId, + input.messageId, + ); + if (receipt) { + const source = receipt.sourceMessage; + const actualDigest = + source.submittedContentDigest ?? messageContentDigest(source.content); + if (source.placement !== 'current_turn' || actualDigest !== expectedDigest) { + throw new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub target Message identity belongs to different content', + ); + } + return source.disposition === 'turn_started' + ? { turnId: receipt.admission.turnId } + : source.disposition === 'steering' + ? { turnId: receipt.admission.turnId, steered: true as const } + : undefined; + } + const admission = await stores.sessionStore.readMessageAdmission( + input.sessionId, + input.messageId, + ); + if (!admission) return undefined; + if ( + admission.submittedPlacement !== 'current_turn' || + admission.submittedContentDigest !== expectedDigest + ) { + throw new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub target Message admission belongs to different content', + ); + } + const root = await stores.agentRunStore.readRootTurnAdmission( + input.sessionId, + admission.turnId, + ); + if (!root || root.runId !== admission.runId) return undefined; + return { turnId: admission.turnId, steered: true as const }; }, }, resolveCreateTarget: async () => { diff --git a/packages/runtime-host/src/server/workhub-coordination-action-gate.ts b/packages/runtime-host/src/server/workhub-coordination-action-gate.ts index bf95b82ae8..1dd75ec489 100644 --- a/packages/runtime-host/src/server/workhub-coordination-action-gate.ts +++ b/packages/runtime-host/src/server/workhub-coordination-action-gate.ts @@ -77,8 +77,34 @@ export interface WorkHubActionGateEffects { }, context: ConnectionContext, ): Promise<{ readonly turnId: string; readonly steered?: true }>; + recoverSubmission(input: { + readonly sessionId: string; + readonly messageId: string; + readonly text: string; + }): Promise<{ readonly turnId: string; readonly steered?: true } | undefined>; + readDelegation(actionId: string): Promise; + prepareDelegation(intent: WorkHubDelegationIntent): Promise; + commitDelegation(commit: WorkHubDelegationCommit): Promise; +} + +export interface WorkHubDelegationIntent { + readonly kind: 'delegation_intent'; + readonly actionId: string; + readonly actionFingerprint: `sha256:${string}`; + readonly coordinationTurnId: string; + readonly targetSessionId: string; + readonly disposition: 'delegate_existing' | 'create_new'; +} + +export interface WorkHubDelegationCommit extends Omit { + readonly kind: 'delegation_committed'; + readonly delegationId: string; + readonly targetTurnId: string; + readonly steered?: true; } +export type WorkHubDelegationRecord = WorkHubDelegationIntent | WorkHubDelegationCommit; + export type WorkHubActionEffectFailureCode = | 'host_not_ready' | 'host_draining' @@ -161,12 +187,12 @@ export class WorkHubCoordinationActionGate { return replay.result; } - const result = this.#act(input, context); + const result = this.#act(input, fingerprint, context); const action = { fingerprint, result }; this.#actions.set(input.actionId, action); - // Successful actions remain replayable. A rejected admission does not own - // the action identity forever: callers must be able to refresh stale - // candidates or satisfy an actionable precondition and retry safely. + // Successful actions remain a Host-lifetime fast path. Rejections leave the + // in-memory slot so a pre-intent admission can retry; once an intent is + // durable, the journal independently keeps that action identity owned. void result.catch(() => { if (this.#actions.get(input.actionId) === action) { this.#actions.delete(input.actionId); @@ -178,6 +204,7 @@ export class WorkHubCoordinationActionGate { async #act( input: WorkHubCoordinationActInput, + fingerprint: `sha256:${string}`, context: ConnectionContext, ): Promise { const proposal = input.proposal; @@ -195,6 +222,20 @@ export class WorkHubCoordinationActionGate { }); return { disposition: 'clarify', coordinationTurnId: turnId }; } + const durable = await this.#effects.readDelegation(input.actionId); + if (durable) { + if (durable.actionFingerprint !== fingerprint) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub action identity belongs to a different proposal', + ); + } + if (durable.kind === 'delegation_committed') { + return committedResult(durable); + } + return this.#executeDelegation(input, durable, context); + } + if (proposal.disposition === 'create_new') { if (!input.create) { throw new WorkHubActionGateFailure( @@ -203,20 +244,9 @@ export class WorkHubCoordinationActionGate { ); } const sessionId = workHubCreatedSessionId(input.actionId); - await this.#effects.create({ - sessionId, - workspace: input.create.workspace, - title: proposal.title, - }); - const submitted = await this.#effects.submit( - { - sessionId, - messageId: actionMessageId(input.actionId), - text: input.userText, - }, - context, - ); - return executionResult('create_new', sessionId, submitted); + const intent = delegationIntent(input, fingerprint, sessionId); + await this.#effects.prepareDelegation(intent); + return this.#executeDelegation(input, intent, context); } const candidates = await this.candidates(); @@ -237,23 +267,63 @@ export class WorkHubCoordinationActionGate { } this.#assertTarget(target); - return this.#submitExisting(input, target, context); + const intent = delegationIntent(input, fingerprint, target.sessionId); + await this.#effects.prepareDelegation(intent); + return this.#executeDelegation(input, intent, context); } - async #submitExisting( + async #executeDelegation( input: WorkHubCoordinationActInput, - target: WorkHubCoordinationCandidate, + intent: WorkHubDelegationIntent, context: ConnectionContext, ): Promise { - const submitted = await this.#effects.submit( - { - sessionId: target.sessionId, - messageId: actionMessageId(input.actionId), - text: input.userText, - }, - context, - ); - return executionResult('delegate_existing', target.sessionId, submitted); + if (intent.disposition === 'create_new') { + if (input.proposal.disposition !== 'create_new' || !input.create) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub durable creation intent does not match the requested action', + ); + } + await this.#effects.create({ + sessionId: intent.targetSessionId, + workspace: input.create.workspace, + title: input.proposal.title, + }); + } else if (input.proposal.disposition !== 'delegate_existing') { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub durable delegation intent does not match the requested action', + ); + } + + const message = { + sessionId: intent.targetSessionId, + messageId: actionMessageId(input.actionId), + text: input.userText, + }; + let submitted: { readonly turnId: string; readonly steered?: true }; + try { + submitted = await this.#effects.submit(message, context); + } catch (error) { + if ( + !(error instanceof WorkHubActionEffectFailure) || + error.code !== 'commit_outcome_unknown' + ) { + throw error; + } + const recovered = await this.#effects.recoverSubmission(message); + if (!recovered) throw error; + submitted = recovered; + } + const commit: WorkHubDelegationCommit = { + ...intent, + kind: 'delegation_committed', + delegationId: delegationId(input.actionId), + targetTurnId: submitted.turnId, + ...(submitted.steered ? { steered: true as const } : {}), + }; + await this.#effects.commitDelegation(commit); + return committedResult(commit); } #assertTarget(target: WorkHubCoordinationCandidate): void { @@ -328,6 +398,34 @@ function actionMessageId(actionId: string): string { return `whm_${hash(actionId).slice(0, 48)}`; } +function delegationId(actionId: string): string { + return `whd_${hash(`delegation\0${actionId}`).slice(0, 48)}`; +} + +function delegationIntent( + input: WorkHubCoordinationActInput, + actionFingerprint: `sha256:${string}`, + targetSessionId: string, +): WorkHubDelegationIntent { + if ( + input.proposal.disposition !== 'delegate_existing' && + input.proposal.disposition !== 'create_new' + ) { + throw new WorkHubActionGateFailure( + 'action_conflict', + 'WorkHub local action cannot create a delegation intent', + ); + } + return { + kind: 'delegation_intent', + actionId: input.actionId, + actionFingerprint, + coordinationTurnId: input.actionId, + targetSessionId, + disposition: input.proposal.disposition, + }; +} + function workHubCreatedSessionId(actionId: string): string { return `whs_${hash(`create\0${actionId}`).slice(0, 48)}`; } @@ -350,16 +448,12 @@ function updatedAt(session: WorkHubActionGateSession): number { return session.lastMessageAt ?? session.statusUpdatedAt ?? session.createdAt; } -function executionResult( - disposition: 'delegate_existing' | 'create_new', - sessionId: string, - submitted: { readonly turnId: string; readonly steered?: true }, -): WorkHubCoordinationActResult { +function committedResult(commit: WorkHubDelegationCommit): WorkHubCoordinationActResult { return { - disposition, - targetSessionId: sessionId, - targetTurnId: submitted.turnId, - ...(submitted.steered ? { steered: true as const } : {}), + disposition: commit.disposition, + targetSessionId: commit.targetSessionId, + targetTurnId: commit.targetTurnId, + ...(commit.steered ? { steered: true as const } : {}), } as WorkHubCoordinationActResult; } diff --git a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts index a90e8449b9..8af2c61439 100644 --- a/packages/runtime-host/src/server/workhub-coordination-coordinator.ts +++ b/packages/runtime-host/src/server/workhub-coordination-coordinator.ts @@ -55,6 +55,7 @@ import { WorkHubCoordinationActionGate, type WorkHubActionGateEffects, } from './workhub-coordination-action-gate.js'; +import { WorkHubDelegationJournal } from './workhub-delegation-journal.js'; const CREATE_FINGERPRINT = `sha256:${createHash('sha256') .update('maka:workhub-coordination-session:v1', 'utf8') @@ -100,7 +101,10 @@ export interface HostWorkHubCoordinationCoordinatorOptions { readonly admission: SessionAdmissionGate; readonly continuity: Pick; readonly executions: CoordinationExecutions; - readonly sessionActions: Pick; + readonly sessionActions: Pick< + WorkHubActionGateEffects, + 'create' | 'submit' | 'recoverSubmission' + >; readonly resolveCreateTarget: () => Promise; readonly requestDrain: () => void; } @@ -123,6 +127,7 @@ export class HostWorkHubCoordinationCoordinator { readonly #resolveCreateTarget: () => Promise; readonly #requestDrain: () => void; readonly #actionGate: WorkHubCoordinationActionGate; + readonly #delegations: WorkHubDelegationJournal; constructor(options: HostWorkHubCoordinationCoordinatorOptions) { this.#coordinationCwd = join(options.stateRoot, COORDINATION_CWD_DIRECTORY); @@ -132,6 +137,12 @@ export class HostWorkHubCoordinationCoordinator { this.#executions = options.executions; this.#resolveCreateTarget = options.resolveCreateTarget; this.#requestDrain = options.requestDrain; + this.#delegations = new WorkHubDelegationJournal({ + stores: options.stores, + admission: options.admission, + continuity: options.continuity, + requestDrain: options.requestDrain, + }); this.#actionGate = new WorkHubCoordinationActionGate({ listSessions: () => this.#stores.listHeaders(), answer: async (input, context) => { @@ -152,6 +163,10 @@ export class HostWorkHubCoordinationCoordinator { }, create: options.sessionActions.create, submit: options.sessionActions.submit, + recoverSubmission: options.sessionActions.recoverSubmission, + readDelegation: (actionId) => this.#delegations.read(actionId), + prepareDelegation: (intent) => this.#delegations.prepare(intent), + commitDelegation: (commit) => this.#delegations.commit(commit), }); } diff --git a/packages/runtime-host/src/server/workhub-delegation-journal.ts b/packages/runtime-host/src/server/workhub-delegation-journal.ts new file mode 100644 index 0000000000..9d7937702b --- /dev/null +++ b/packages/runtime-host/src/server/workhub-delegation-journal.ts @@ -0,0 +1,300 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { createHash } from 'node:crypto'; +import { + WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION, + WORKHUB_COORDINATION_SESSION_ID, + isWorkHubCoordinationSession, + type StoredMessage, + type WorkHubDelegationCommittedMessage, + type WorkHubDelegationIntentMessage, +} from '@maka/core/session'; +import type { SessionAuthorityStore } from '@maka/storage/session-store'; +import type { SessionContinuityCoordinator } from './session-continuity-coordinator.js'; +import { + WorkHubActionEffectFailure, + type WorkHubDelegationCommit, + type WorkHubDelegationIntent, + type WorkHubDelegationRecord, +} from './workhub-coordination-action-gate.js'; +import type { SessionAdmissionGate } from './session-admission-gate.js'; + +const RECORD_KINDS = ['delegation_intent', 'delegation_committed'] as const; +const RECORD_READ_MAX_BYTES = 16 * 1024; + +type JournalStores = Pick< + SessionAuthorityStore, + | 'appendMessages' + | 'readHeaderSnapshot' + | 'readTranscriptHighWaterSnapshot' + | 'readTranscriptMessagesSnapshot' +>; + +export interface WorkHubDelegationJournalOptions { + readonly stores: JournalStores; + readonly admission: SessionAdmissionGate; + readonly continuity: Pick; + readonly requestDrain: () => void; +} + +/** + * Append-only authority for WorkHub action intent and committed delegation links. + * + * The interface exposes domain records only. Message identities, transcript + * snapshots, exact replay checks, and commit-outcome handling stay local here. + */ +export class WorkHubDelegationJournal { + readonly #stores: JournalStores; + readonly #admission: SessionAdmissionGate; + readonly #continuity: Pick; + readonly #requestDrain: () => void; + + constructor(options: WorkHubDelegationJournalOptions) { + this.#stores = options.stores; + this.#admission = options.admission; + this.#continuity = options.continuity; + this.#requestDrain = options.requestDrain; + } + + async read(actionId: string): Promise { + await this.#assertCoordinationSession(); + const messages = await this.#readMessages(actionId); + return this.#projectRecord(actionId, messages); + } + + prepare(intent: WorkHubDelegationIntent): Promise { + return this.#admission.run(WORKHUB_COORDINATION_SESSION_ID, async (lease) => { + await this.#assertCoordinationSession(); + const existing = this.#projectRecord( + intent.actionId, + await this.#readMessages(intent.actionId), + ); + if (existing) { + if (!sameIntent(existing, intent)) throw actionConflict(); + return; + } + try { + await this.#stores.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [intentMessage(intent)]); + await this.#continuity.refreshCanonical(WORKHUB_COORDINATION_SESSION_ID, lease); + } catch (error) { + if (error instanceof WorkHubActionEffectFailure) throw error; + this.#requestDrain(); + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'WorkHub delegation intent outcome is unknown', + ); + } + }); + } + + commit(commit: WorkHubDelegationCommit): Promise { + return this.#admission.run(WORKHUB_COORDINATION_SESSION_ID, async (lease) => { + await this.#assertCoordinationSession(); + const existing = this.#projectRecord( + commit.actionId, + await this.#readMessages(commit.actionId), + ); + if (existing?.kind === 'delegation_committed') { + if (!sameCommit(existing, commit)) throw actionConflict(); + return; + } + if (!existing || !sameIntent(existing, commit)) throw actionConflict(); + try { + await this.#stores.appendMessages(WORKHUB_COORDINATION_SESSION_ID, [ + committedMessage(commit), + ]); + await this.#continuity.refreshCanonical(WORKHUB_COORDINATION_SESSION_ID, lease); + } catch (error) { + if (error instanceof WorkHubActionEffectFailure) throw error; + this.#requestDrain(); + throw new WorkHubActionEffectFailure( + 'commit_outcome_unknown', + 'WorkHub delegation commit outcome is unknown', + ); + } + }); + } + + async #assertCoordinationSession(): Promise { + try { + const header = await this.#stores.readHeaderSnapshot(WORKHUB_COORDINATION_SESSION_ID); + if (isWorkHubCoordinationSession(header) && !header.isArchived) return; + throw new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub Coordination Session identity is unavailable', + ); + } catch (error) { + if (error instanceof WorkHubActionEffectFailure) throw error; + throw new WorkHubActionEffectFailure( + 'persistence_failed', + 'WorkHub Coordination Session state is unavailable', + ); + } + } + + async #readMessages(actionId: string): Promise { + try { + const throughSequence = await this.#stores.readTranscriptHighWaterSnapshot( + WORKHUB_COORDINATION_SESSION_ID, + ); + if (throughSequence === null) return []; + return await this.#stores.readTranscriptMessagesSnapshot(WORKHUB_COORDINATION_SESSION_ID, { + messageIds: RECORD_KINDS.map((kind) => recordMessageId(actionId, kind)), + throughSequence, + maxBytes: RECORD_READ_MAX_BYTES, + maxMessages: RECORD_KINDS.length, + }); + } catch (error) { + if (error instanceof WorkHubActionEffectFailure) throw error; + throw new WorkHubActionEffectFailure( + 'persistence_failed', + 'WorkHub delegation records are unavailable', + ); + } + } + + #projectRecord( + actionId: string, + messages: readonly StoredMessage[], + ): WorkHubDelegationRecord | undefined { + try { + return projectRecord(actionId, messages); + } catch (error) { + this.#requestDrain(); + throw error; + } + } +} + +function projectRecord( + actionId: string, + messages: readonly StoredMessage[], +): WorkHubDelegationRecord | undefined { + if (messages.length === 0) return undefined; + const intent = messages.find( + (message): message is WorkHubDelegationIntentMessage => + message.type === 'workhub_coordination' && message.kind === 'delegation_intent', + ); + const committed = messages.find( + (message): message is WorkHubDelegationCommittedMessage => + message.type === 'workhub_coordination' && message.kind === 'delegation_committed', + ); + if ( + messages.length !== Number(intent !== undefined) + Number(committed !== undefined) || + intent?.actionId !== actionId || + (committed !== undefined && (!intent || !sameMessageIntent(intent, committed))) + ) { + throw new WorkHubActionEffectFailure( + 'persistence_failed', + 'WorkHub delegation record chain is invalid', + ); + } + return committed ? commitRecord(committed) : intent ? intentRecord(intent) : undefined; +} + +function intentMessage(intent: WorkHubDelegationIntent): WorkHubDelegationIntentMessage { + return { + type: 'workhub_coordination', + id: recordMessageId(intent.actionId, 'delegation_intent'), + turnId: intent.coordinationTurnId, + ts: Date.now(), + schemaVersion: WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION, + ...intent, + }; +} + +function committedMessage(commit: WorkHubDelegationCommit): WorkHubDelegationCommittedMessage { + return { + type: 'workhub_coordination', + id: recordMessageId(commit.actionId, 'delegation_committed'), + turnId: commit.coordinationTurnId, + ts: Date.now(), + schemaVersion: WORKHUB_COORDINATION_RECORD_SCHEMA_VERSION, + ...commit, + }; +} + +function intentRecord(message: WorkHubDelegationIntentMessage): WorkHubDelegationIntent { + return { + kind: message.kind, + actionId: message.actionId, + actionFingerprint: message.actionFingerprint, + coordinationTurnId: message.coordinationTurnId, + targetSessionId: message.targetSessionId, + disposition: message.disposition, + }; +} + +function commitRecord(message: WorkHubDelegationCommittedMessage): WorkHubDelegationCommit { + return { + kind: 'delegation_committed', + actionId: message.actionId, + actionFingerprint: message.actionFingerprint, + coordinationTurnId: message.coordinationTurnId, + targetSessionId: message.targetSessionId, + disposition: message.disposition, + delegationId: message.delegationId, + targetTurnId: message.targetTurnId, + ...(message.steered ? { steered: true as const } : {}), + }; +} + +function sameMessageIntent( + intent: WorkHubDelegationIntentMessage, + committed: WorkHubDelegationCommittedMessage, +): boolean { + return sameIntent(intentRecord(intent), commitRecord(committed)); +} + +function sameIntent( + left: WorkHubDelegationRecord, + right: Omit, +): boolean { + return ( + left.actionId === right.actionId && + left.actionFingerprint === right.actionFingerprint && + left.coordinationTurnId === right.coordinationTurnId && + left.targetSessionId === right.targetSessionId && + left.disposition === right.disposition + ); +} + +function sameCommit(left: WorkHubDelegationCommit, right: WorkHubDelegationCommit): boolean { + return ( + sameIntent(left, right) && + left.delegationId === right.delegationId && + left.targetTurnId === right.targetTurnId && + left.steered === right.steered + ); +} + +function recordMessageId(actionId: string, kind: (typeof RECORD_KINDS)[number]): string { + return `whj_${createHash('sha256') + .update(`${actionId}\0${kind}`, 'utf8') + .digest('hex') + .slice(0, 48)}`; +} + +function actionConflict(): WorkHubActionEffectFailure { + return new WorkHubActionEffectFailure( + 'operation_conflict', + 'WorkHub action identity belongs to different durable delegation content', + ); +}