From eb0e1bb2ef19d5fc9630d50aa775c57c6229b68c Mon Sep 17 00:00:00 2001 From: Ryder Sun Date: Wed, 9 Sep 2026 20:57:13 +0800 Subject: [PATCH] fix: resume group tasks after human clarification Bind clarification gates to completed task decisions and preserve them across restart. Make repeated gate responses idempotent to prevent invocation replay. Validation: desktop 1549/1549; real Electron pending restart, input response, completion and duplicate-response checks passed with exactly one continuation call. --- desktop/src/gates/human-gate-coordinator.cjs | 4 +- desktop/src/runs/run-ledger-records.cjs | 11 +- .../workspace/local-workspace-auto-runner.cjs | 26 ++- desktop/src/workspace/local-workspace.cjs | 111 ++++++++++- .../gates/human-gate-coordinator.test.cjs | 19 ++ ...l-workspace-v4-natural-discussion.test.cjs | 2 +- .../local-workspace-v4-task-decision.test.cjs | 183 ++++++++++++++++++ docs/releases/Meldwork-V1.0.5-progress.md | 16 ++ 8 files changed, 360 insertions(+), 12 deletions(-) create mode 100644 desktop/test/workspace/local-workspace-v4-task-decision.test.cjs diff --git a/desktop/src/gates/human-gate-coordinator.cjs b/desktop/src/gates/human-gate-coordinator.cjs index 9184f6d..a70063d 100644 --- a/desktop/src/gates/human-gate-coordinator.cjs +++ b/desktop/src/gates/human-gate-coordinator.cjs @@ -208,9 +208,11 @@ class HumanGateCoordinator { status, optionId: option.optionId, actorId: input.actorId || 'local-user', - decidedAt: input.decidedAt || this.now(), + decidedAt: input.decidedAt || current.decision?.decidedAt || this.now(), ...(response ? { response } : {}), }) + // Recovery owns recorded decisions; repeated submissions must not dispatch them again. + if (current.status !== 'pending') return publicHumanGate(record) this.clearExpiry(record.gateId) try { this.onResumed(record, record.decision) diff --git a/desktop/src/runs/run-ledger-records.cjs b/desktop/src/runs/run-ledger-records.cjs index 65854fa..925e79d 100644 --- a/desktop/src/runs/run-ledger-records.cjs +++ b/desktop/src/runs/run-ledger-records.cjs @@ -67,7 +67,7 @@ const CONTINUATION_V4_PHASES = new Set([ ]) const CONTINUATION_GATE_TYPES = new Set(['permission', 'budget', 'decision', 'retry', 'input']) const CONTINUATION_RESUME_KINDS = new Set([ - 'agent_slot', 'role_review_decision', 'v4_human_gate', 'v4_synthesis_recovery', + 'agent_slot', 'role_review_decision', 'v4_human_gate', 'v4_synthesis_recovery', 'v4_task_decision', ]) const CONTINUATION_STATES = new Set([ 'pending', 'ready', 'resuming', 'completed', 'failed', 'cancelled', @@ -405,16 +405,18 @@ function normalizeContinuation(input) { || !CONTINUATION_STATES.has(state) || !agentRunId || !agentKind || (hasPublicAgentRunId && !publicAgentRunId) || (resumeKind === 'role_review_decision' && gateType !== 'decision') + || (resumeKind === 'v4_task_decision' + && (gateType !== 'input' || !hasInvocationBinding || phase !== 'discussion')) || (resumeKind === 'v4_human_gate' && (gateType !== 'decision' || !hasStateEpoch || stateEpoch < 1)) || (resumeKind === 'v4_synthesis_recovery' && (gateType !== 'decision' || !hasStateEpoch)) || (!['v4_human_gate', 'v4_synthesis_recovery'].includes(resumeKind) && hasStateEpoch) || (invocationFieldCount > 0 && !hasInvocationBinding) - || (hasInvocationBinding && (resumeKind !== 'agent_slot' + || (hasInvocationBinding && (!['agent_slot', 'v4_task_decision'].includes(resumeKind) || !CONTINUATION_V4_PHASES.has(phase) || !slotId || !operationId || !SHA256.test(snapshotHash))) - || (gateType === 'input' && !hasRequestBinding) + || (gateType === 'input' && resumeKind !== 'v4_task_decision' && !hasRequestBinding) || (!hasRequestBinding && [requestId, requestHash, sessionRefHash, sessionProvenanceHash] .some(Boolean))) return undefined return { @@ -443,7 +445,8 @@ function continuationBindingMatchesOrchestration(continuation, orchestration) { } const hasBinding = CONTINUATION_INVOCATION_FIELDS.every(field => hasOwn(continuation || {}, field)) if (!hasBinding) return true - if (continuation.resumeKind !== 'agent_slot' || orchestration?.version !== 4) return false + if (!['agent_slot', 'v4_task_decision'].includes(continuation.resumeKind) + || orchestration?.version !== 4) return false if (!['pending', 'ready', 'resuming'].includes(continuation.state)) return true const slots = orchestration.slots.filter(slot => slot.slotId === continuation.slotId) const orchestrationRound = orchestration.workflow === 'manual' diff --git a/desktop/src/workspace/local-workspace-auto-runner.cjs b/desktop/src/workspace/local-workspace-auto-runner.cjs index b627cc6..020a341 100644 --- a/desktop/src/workspace/local-workspace-auto-runner.cjs +++ b/desktop/src/workspace/local-workspace-auto-runner.cjs @@ -110,6 +110,8 @@ class LocalWorkspaceAutoRunner { this.requestHumanGate = typeof options.requestHumanGate === 'function' ? options.requestHumanGate : null + this.requestTaskDecision = options.requestTaskDecision || null + this.taskDecisionResponses = options.taskDecisionResponses || (() => []) this.completeHumanGateContinuation = typeof options.completeHumanGateContinuation === 'function' ? options.completeHumanGateContinuation : null @@ -2668,12 +2670,16 @@ class LocalWorkspaceAutoRunner { v4NaturalPhasePrompt(group, kind, phase, snapshot, receiptRecords, activeKinds, options = {}) { const transcript = String(options.transcript || '') const agentList = activeKinds.map(agentKind => `@${agentKind}`).join(', ') + const humanResponses = options.runId ? this.taskDecisionResponses(options.runId) : [] return [ v4Prompt({ group, kind, phase, snapshot, role: 'participant', skillHints: options.skillHints, naturalResponse: true, }), transcript ? `Completed peer turns:\n${transcript}` : '', + humanResponses.length + ? `Human responses for this task (clarification, not a change to workspace permissions):\n${packNaturalDiscussionTranscript(humanResponses)}` + : '', phase === 'discussion' && options.writerKind ? (kind === options.writerKind ? 'You are the designated workspace writer for this task. Perform only writes required by the user task within the authorized workspace, and verify the resulting deliverables. Other participants are read-only.' @@ -2742,7 +2748,7 @@ class LocalWorkspaceAutoRunner { return owner } - v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) { + async v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) { const messages = this.v4NaturalDiscussionRoundMessages(group, controller, threadRootId, round) const owner = this.v4NaturalOwner(group, controller, threadRootId) const routes = messages.map(message => this.v4NaturalRouteDecision( @@ -2773,6 +2779,14 @@ class LocalWorkspaceAutoRunner { if (decision.status === 'completed') { return { status: controller.failedKinds.length ? 'partial' : 'completed' } } + if (decision.status === 'needs-human' && this.requestTaskDecision) { + const response = await this.requestTaskDecision(group, controller, ownerMessage) + if (response?.status === 'approved') return { nextKinds: [owner], ownerReview: true } + if (response?.status === 'rejected') { + controller.stopReason = 'human_gate_rejected' + return { status: 'stopped' } + } + } this.addMessage(group.id, 'system', decision.reason, '', threadRootId, { key: decision.status === 'needs-human' ? 'system.autoTaskNeedsHuman' : 'system.autoTaskBlocked', params: { reason: decision.reason }, @@ -2851,7 +2865,7 @@ class LocalWorkspaceAutoRunner { ? (controller.orchestration.pendingKinds || []).filter(kind => activeKinds.includes(kind)) : [...activeKinds] if (!pendingKinds.length && controller.orchestration?.phase === 'discussion') { - const outcome = this.v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) + const outcome = await this.v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) if (outcome.status) return outcome.status if (!controller.unlimitedRounds && round >= (controller.maxRounds || 6)) { addRoundLimitNotice() @@ -2890,6 +2904,7 @@ class LocalWorkspaceAutoRunner { disableDeliveryPrompt: true, promptBuilder: ({ kind: promptKind, transcript }) => this.v4NaturalPhasePrompt( group, promptKind, 'discussion', snapshot, receiptRecords, activeKinds, { + runId: controller.runId, transcript: transcript ?? this.v4NaturalThreadTranscript( group, threadRootId, { controller }, ), @@ -2911,7 +2926,7 @@ class LocalWorkspaceAutoRunner { } pendingKinds = remainingKinds.filter(agentKind => activeKinds.includes(agentKind)) } - const outcome = this.v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) + const outcome = await this.v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) if (outcome.status) return outcome.status if (!controller.unlimitedRounds && round >= (controller.maxRounds || 6)) { addRoundLimitNotice() @@ -2963,7 +2978,7 @@ class LocalWorkspaceAutoRunner { if (this.v4NaturalStopRepeatingDiscussion( group, controller, threadRootId, round, activeKinds, )) return 'partial' - const outcome = this.v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) + const outcome = await this.v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) if (outcome.status) return outcome.status nextKinds = outcome.nextKinds if (!controller.unlimitedRounds && round >= (controller.maxRounds || 6)) { @@ -3005,6 +3020,7 @@ class LocalWorkspaceAutoRunner { disableDeliveryPrompt: true, promptBuilder: ({ kind, transcript }) => this.v4NaturalPhasePrompt( group, kind, 'discussion', snapshot, receiptRecords, activeKinds, { + runId: controller.runId, transcript: transcript ?? this.v4NaturalThreadTranscript(group, threadRootId, { controller }), skillHints: context.rootSkillsByKind.get(kind) || [], allowRouting: true, @@ -3023,7 +3039,7 @@ class LocalWorkspaceAutoRunner { if (this.v4NaturalStopRepeatingDiscussion( group, controller, threadRootId, round, activeKinds, )) return 'partial' - const outcome = this.v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) + const outcome = await this.v4NaturalRoundOutcome(group, controller, threadRootId, round, activeKinds) if (outcome.status) return outcome.status nextKinds = outcome.nextKinds if (!controller.unlimitedRounds && round >= (controller.maxRounds || 6)) { diff --git a/desktop/src/workspace/local-workspace.cjs b/desktop/src/workspace/local-workspace.cjs index 24420f5..ffbe1c2 100644 --- a/desktop/src/workspace/local-workspace.cjs +++ b/desktop/src/workspace/local-workspace.cjs @@ -343,6 +343,8 @@ class LocalWorkspace extends EventEmitter { hasRunLedger: () => Boolean(this.runLedger), requestHumanGate: (...args) => this.requestHumanGate(...args), completeHumanGateContinuation: (...args) => this.completeHumanGateContinuation(...args), + requestTaskDecision: (...args) => this.requestV4TaskDecision(...args), + taskDecisionResponses: runId => this.v4TaskDecisionResponses(runId), retryContract: kind => ({ idempotencyMode: this.detectedAgents.find(agent => agent.kind === kind)?.idempotencyMode === 'durable' @@ -806,7 +808,7 @@ class LocalWorkspace extends EventEmitter { const invocationFields = ['phase', 'slotId', 'operationId', 'snapshotHash'] const invocationFieldCount = invocationFields.filter(field => Object.hasOwn(input, field)).length if (invocationFieldCount > 0 - && (input.resumeKind !== 'agent_slot' + && (!['agent_slot', 'v4_task_decision'].includes(input.resumeKind) || invocationFieldCount !== invocationFields.length)) { throw new Error('LOCAL_RUN_CONTINUATION_INVALID') } @@ -1141,6 +1143,103 @@ class LocalWorkspace extends EventEmitter { return { ...match, state: continuation.state } } + v4TaskDecisionSource(durable, gate, request) { + if (durable?.mode !== 'auto' || durable.orchestration?.version !== 4 + || gate.type !== 'input' || gate.runId !== durable.runId + || !request || Object.keys(request).length !== 7 + || request.source !== 'task-decision' || request.phase !== 'discussion' + || request.snapshotHash !== durable.orchestration.snapshotHash) return null + const sources = durable.agentRuns.filter(attempt => ( + attempt.agentRunId === gate.agentRunId && attempt.kind === gate.agentKind + && attempt.status === 'completed' && attempt.round === request.round + && attempt.context?.operationId === request.operationId + && attempt.context?.snapshotHash === request.snapshotHash + && attempt.context?.taskDecision?.status === 'needs-human' + && hashValue(attempt.context.taskDecision) === request.decisionHash + )) + if (sources.length !== 1 || request.operationId !== this.autoRunner.v4OperationId( + durable, gate.agentKind, `discussion:${request.round}`, request.slotId, request.round, + )) return null + return sources[0] + } + + canResumeV4TaskDecision(durable, gate, request) { + const source = this.v4TaskDecisionSource(durable, gate, request) + const continuation = durable.continuation + const cursor = durable.orchestration + const group = this.state.groups.find(item => item.id === durable.groupId) + const slot = cursor?.slots.find(item => item.slotId === request?.slotId) + return Boolean(source && group && gate.options.length === 2 + && gate.options[0].optionId === 'respond' && gate.options[0].kind === 'respond' + && gate.options[1].optionId === 'cancel' && gate.options[1].kind === 'reject' + && continuation?.resumeKind === 'v4_task_decision' + && cursor.phase === 'discussion' && cursor.round === request.round + && durable.currentRound === request.round && continuation.round === request.round + && continuation.slotId === request.slotId + && continuation.operationId === request.operationId + && continuation.snapshotHash === request.snapshotHash + && slot?.agentRunId === source.agentRunId && slot.agentKind === source.kind + && slot.operationId === request.operationId && slot.commitStatus === 'committed' + && this.autoRunner.v4NaturalOwner(group, durable, durable.threadRootId) === source.kind) + } + + v4TaskDecisionResponses(runId) { + const durable = this.runLedger?.get?.(runId) + if (!durable) return [] + return this.humanGateStore.list({ runId }).flatMap(item => { + if (item.type !== 'input' || item.status !== 'approved') return [] + try { + const gate = this.humanGateStore.get(item.gateId) + const request = this.humanGateStore.request(item.gateId) + const source = this.v4TaskDecisionSource(durable, gate, request) + return source && gate.decision?.response ? [{ + id: gate.gateId, round: source.round, agentKind: 'human', + text: gate.decision.response, + }] : [] + } catch { return [] } + }) + } + + async requestV4TaskDecision(group, controller, message) { + const durable = this.runLedger?.get?.(controller.runId) + if (!durable) return null + const source = durable.agentRuns.find(attempt => ( + attempt.agentRunId === message.trace?.agentRunId && attempt.kind === message.agentKind + )) + const slot = durable.orchestration?.slots.find(item => item.agentRunId === source?.agentRunId) + if (!source || !slot || source.context?.taskDecision?.status !== 'needs-human' + || slot.commitStatus !== 'committed') throw new Error('LOCAL_RUN_CONTINUATION_INVALID') + const request = { + source: 'task-decision', phase: 'discussion', round: source.round, + slotId: slot.slotId, operationId: slot.operationId, + snapshotHash: durable.orchestration.snapshotHash, + decisionHash: hashValue(source.context.taskDecision), + } + const decision = await this.requestHumanGate({ + type: 'input', runId: durable.runId, agentRunId: source.agentRunId, agentKind: source.kind, + createdAt: new Date(slot.finishedAt).toISOString(), + summary: source.context.taskDecision.reason, + options: [ + { optionId: 'respond', name: 'Respond', kind: 'respond' }, + { optionId: 'cancel', name: 'Cancel', kind: 'reject' }, + ], + request, + }, { + signal: controller.signal, + preserveOnAbort: () => controller.stopReason === 'shutdown', + continuation: { + resumeKind: 'v4_task_decision', agentRunId: source.agentRunId, agentKind: source.kind, + round: source.round, phase: 'discussion', slotId: slot.slotId, + operationId: slot.operationId, snapshotHash: durable.orchestration.snapshotHash, + }, + }) + if (this.completeHumanGateContinuation(durable.runId, decision.gateId, + decision.status === 'approved' ? 'completed' : 'cancelled') !== true) { + throw new Error('LOCAL_RUN_PERSIST_FAILED') + } + return decision + } + async requestHumanGate(input, options = {}) { const runId = String(input?.runId || '') const previous = this.humanGateWaitTails.get(runId) || Promise.resolve() @@ -1248,6 +1347,11 @@ class LocalWorkspace extends EventEmitter { if (continuation.state !== 'pending' && gate.status === 'pending') return false try { const request = this.humanGateStore.request(gate.gateId) + if (continuation.resumeKind === 'v4_task_decision') { + const pack = this.contextPackStore.get(runRecord.contextPackId) + return pack.taskId === runRecord.taskId + && this.canResumeV4TaskDecision(runRecord, gate, request) + } const connectorBound = gate.type === 'input' || request?.source === 'connector' if (connectorBound && ( continuation.requestId !== request.requestId @@ -1760,6 +1864,11 @@ class LocalWorkspace extends EventEmitter { try { controller = this.runCoordinator.resume(durable) const group = this.getGroup(durable.groupId) + if (durable.continuation.resumeKind === 'v4_task_decision') { + const finalStatus = await this.autoRunner.resume(group, durable, controller) + await this.finishRun(durable.groupId, controller, finalStatus) + return + } const resumableManual = this.canResumeManualOrchestration(durable) const resumableAuto = this.canResumeAutoOrchestration(durable) const resumableV4ManualSlot = this.canResumeV4ManualAgentSlot(durable, gate) diff --git a/desktop/test/gates/human-gate-coordinator.test.cjs b/desktop/test/gates/human-gate-coordinator.test.cjs index 3696086..08548be 100644 --- a/desktop/test/gates/human-gate-coordinator.test.cjs +++ b/desktop/test/gates/human-gate-coordinator.test.cjs @@ -64,6 +64,25 @@ test('persists a pending gate and resumes its live waiter after one decision', a assert.deepEqual(events.map(value => value.split(':', 1)[0]), ['waiting', 'resumed']) }) +test('repeated decisions return the persisted result without replaying resume callbacks', async t => { + let resumed = 0 + let orphaned = 0 + let tick = 0 + const { coordinator, store } = fixture(t, { + now: () => new Date(Date.UTC(2026, 7, 4, 0, 0, tick++)).toISOString(), + onResumed: () => { resumed += 1 }, + onOrphanDecision: () => { orphaned += 1 }, + }) + const gate = store.create({ ...request(), createdAt: '2026-08-04T00:00:00.000Z' }) + const first = coordinator.decide(gate.gateId, { optionId: 'allow-once' }) + assert.deepEqual(coordinator.decide(gate.gateId, { optionId: 'allow-once' }), first) + await coordinator.resumePromises.get(gate.gateId) + assert.deepEqual(coordinator.decide(gate.gateId, { optionId: 'allow-once' }), first) + assert.equal(resumed, 1) + assert.equal(orphaned, 1) + assert.throws(() => coordinator.decide(gate.gateId, { optionId: 'reject-once' })) +}) + test('aborting a waiter rejects the request and durably selects a reject option', async (t) => { const { coordinator } = fixture(t) const controller = new AbortController() diff --git a/desktop/test/workspace/local-workspace-v4-natural-discussion.test.cjs b/desktop/test/workspace/local-workspace-v4-natural-discussion.test.cjs index fe7ab38..20a40ac 100644 --- a/desktop/test/workspace/local-workspace-v4-natural-discussion.test.cjs +++ b/desktop/test/workspace/local-workspace-v4-natural-discussion.test.cjs @@ -138,7 +138,7 @@ test('Natural routing retains a request beyond the stored message excerpt', asyn assert.deepEqual(turns, ['codex', 'hermes', 'codex']) }) -for (const status of ['completed', 'blocked', 'needs-human']) { +for (const status of ['completed', 'blocked']) { test(`Natural single-Agent task persists the owner's ${status} judgment`, async (t) => { const { directory, options } = fixture() t.after(() => fs.rmSync(directory, { recursive: true, force: true })) diff --git a/desktop/test/workspace/local-workspace-v4-task-decision.test.cjs b/desktop/test/workspace/local-workspace-v4-task-decision.test.cjs new file mode 100644 index 0000000..4fb919a --- /dev/null +++ b/desktop/test/workspace/local-workspace-v4-task-decision.test.cjs @@ -0,0 +1,183 @@ +const test = require('node:test') +const assert = require('node:assert/strict') +const fs = require('node:fs') +const path = require('node:path') +const { LocalWorkspace } = require('../../src/workspace/local-workspace.cjs') +const { RunLedger } = require('../../src/runs/run-ledger.cjs') +const { fixture } = require('../support/local-workspace-test-helpers.cjs') + +function reply(status, reason) { + return `${reason}\n[[MELDWORK_COLLABORATION:${JSON.stringify({ summary: reason, taskDecision: { + status, reason, deliverables: status === 'completed' ? ['The requested answer.'] : [], nextKinds: [], + } })}]]` +} + +async function until(read, timeout = 10000) { + const deadline = Date.now() + timeout + while (Date.now() < deadline) { + const result = read() + if (result) return result + await new Promise(resolve => setTimeout(resolve, 20)) + } + throw new Error('TEST_TASK_DECISION_TIMEOUT') +} + +async function start(t, style = 'agent-led') { + const { directory, options } = fixture() + const storagePath = path.join(directory, 'run-ledger.json') + const calls = [] + const runAgent = async (agent, prompt, _workdir, input) => { + const phase = prompt.includes('Work independently on the user task') ? 'proposal' : 'discussion' + calls.push({ kind: agent.kind, phase, prompt, sessionRef: input.sessionRef }) + return { outcome: 'completed', sessionRef: input.sessionRef || 'human-task-session', + text: phase === 'proposal' ? 'I need the user to choose the target.' + : prompt.includes('USER_CHOICE_B') ? reply('completed', 'The answer uses USER_CHOICE_B.') + : reply('needs-human', 'Which target should be used?') } + } + const workspaceOptions = { ...options, naturalAgentResponses: true, runAgent } + const ledger = new RunLedger({ storagePath }) + const workspace = new LocalWorkspace({ ...workspaceOptions, runLedger: ledger }) + const workspaces = [workspace] + t.after(async () => { + for (const instance of workspaces) await instance.stopAll() + fs.rmSync(directory, { recursive: true, force: true }) + }) + await workspace.refreshAgents() + const group = workspace.createGroup({ name: 'Human task decision', agentKinds: ['codex'], workdir: directory, allowWrite: false }) + await workspace.sendMessage({ groupId: group.id, text: 'Answer after I choose the target.', + mode: 'auto', protocol: 'v4', discussionStyle: style, targetKinds: ['codex'], maxRounds: 5 }) + const controller = workspace.activeRuns.get(group.id) + const gate = await until(() => workspace.listHumanGates({ pendingOnly: true })[0]) + return { workspace, workspaceOptions, workspaces, ledger, storagePath, group, controller, gate, calls } +} + +for (const style of ['agent-led', 'sequential']) { + test(`Natural ${style} needs-human waits for input and continues the same task`, async t => { + const { workspace, ledger, controller, gate, calls } = await start(t, style) + assert.equal(gate.type, 'input') + const waiting = ledger.get(controller.runId) + assert.equal(waiting.status, 'waiting') + assert.equal(waiting.continuation.resumeKind, 'v4_task_decision') + assert.ok(workspace.snapshot().humanGates.some(item => item.gateId === gate.gateId)) + const before = calls.length + workspace.decideHumanGate(gate.gateId, { optionId: 'respond', response: 'USER_CHOICE_B' }) + await controller.promise + const completed = ledger.get(controller.runId) + assert.equal(completed.status, 'completed', completed.reason) + assert.equal(calls.length, before + 1) + assert.match(calls.at(-1).prompt, /Human responses for this task[\s\S]*USER_CHOICE_B/) + assert.equal(completed.continuation.state, 'completed') + assert.equal(calls.at(-1).sessionRef, 'human-task-session') + }) +} + +test('Natural needs-human cancellation stops without invoking another Agent', async t => { + const { workspace, ledger, controller, gate, calls } = await start(t) + const before = calls.length + workspace.decideHumanGate(gate.gateId, { optionId: 'cancel' }) + await controller.promise + assert.equal(ledger.get(controller.runId).status, 'stopped') + assert.equal(calls.length, before) +}) + +test('Natural needs-human survives shutdown and resumes after the user responds without replaying the request', async t => { + const { workspace, workspaceOptions, workspaces, storagePath, controller, gate, calls } = await start(t) + await workspace.stopAll() + await controller.promise + const ledger = new RunLedger({ storagePath }) + assert.equal(ledger.get(controller.runId).continuation.state, 'pending') + const reopened = new LocalWorkspace({ ...workspaceOptions, runLedger: ledger }) + workspaces.push(reopened) + await reopened.refreshAgents() + const before = calls.length + assert.equal(reopened.canResumeHumanGate(reopened.humanGateStore.get(gate.gateId)), true) + reopened.decideHumanGate(gate.gateId, { optionId: 'respond', response: 'USER_CHOICE_B' }) + const final = await until(() => { + const run = ledger.get(controller.runId) + return ['completed', 'partial', 'failed', 'stopped'].includes(run.status) ? run : null + }) + assert.equal(final.status, 'completed', final.reason) + assert.equal(calls.length, before + 1) + assert.match(calls.at(-1).prompt, /USER_CHOICE_B/) + assert.equal(reopened.listHumanGates({ pendingOnly: true }).length, 0) +}) + +test('Natural needs-human rejects altered task, source, snapshot and owner bindings', async t => { + const { workspace, ledger, controller, gate } = await start(t) + const stored = workspace.humanGateStore.get(gate.gateId) + const durable = ledger.get(controller.runId) + const request = workspace.humanGateStore.request(gate.gateId) + assert.equal(workspace.canResumeV4TaskDecision(durable, stored, request), true) + for (const field of ['source', 'phase', 'round', 'slotId', 'operationId', 'snapshotHash', 'decisionHash']) { + assert.equal(workspace.canResumeV4TaskDecision(durable, stored, { + ...request, [field]: field === 'round' ? request.round + 1 : 'different', + }), false, field) + } + assert.equal(workspace.canResumeV4TaskDecision(durable, { ...stored, runId: 'other-run' }, request), false) + assert.equal(workspace.canResumeV4TaskDecision(durable, { ...stored, agentRunId: 'other-call' }, request), false) + assert.equal(workspace.canResumeV4TaskDecision(durable, { ...stored, agentKind: 'hermes' }, request), false) + assert.equal(workspace.canResumeV4TaskDecision(durable, { + ...stored, options: [{ optionId: 'respond', kind: 'allow_always' }, stored.options[1]], + }, request), false) + const altered = structuredClone(durable) + altered.agentRuns.at(-1).context.taskDecision.reason = 'A different question.' + assert.equal(workspace.canResumeV4TaskDecision(altered, stored, request), false) +}) + +test('Natural needs-human resumes an approved decision after shutdown before continuation dispatch', async t => { + const { workspace, workspaceOptions, workspaces, storagePath, controller, gate, calls } = await start(t) + await workspace.stopAll() + await controller.promise + const ledger = new RunLedger({ storagePath }) + const deciding = new LocalWorkspace({ ...workspaceOptions, runLedger: ledger }) + workspaces.push(deciding) + await deciding.refreshAgents() + deciding.humanGateCoordinator.onOrphanDecision = () => {} + deciding.decideHumanGate(gate.gateId, { optionId: 'respond', response: 'USER_CHOICE_B' }) + assert.equal(ledger.get(controller.runId).continuation.state, 'ready') + await deciding.stopAll() + const before = calls.length + const recoveredLedger = new RunLedger({ storagePath }) + const recovered = new LocalWorkspace({ ...workspaceOptions, runLedger: recoveredLedger }) + workspaces.push(recovered) + await recovered.refreshAgents() + const final = await until(() => { + const run = recoveredLedger.get(controller.runId) + return ['completed', 'partial', 'failed', 'stopped'].includes(run.status) ? run : null + }) + assert.equal(final.status, 'completed', final.reason) + assert.equal(calls.length, before + 1) + recovered.decideHumanGate(gate.gateId, { optionId: 'respond', response: 'USER_CHOICE_B' }) + await new Promise(resolve => setTimeout(resolve, 50)) + assert.equal(calls.length, before + 1) +}) + +test('Natural needs-human recovers after applying input but before starting the next turn', async t => { + const { workspace, workspaceOptions, workspaces, storagePath, controller, gate, calls } = await start(t) + const recoveryPath = path.join(path.dirname(storagePath), 'recovery-ledger.json') + const complete = workspace.completeHumanGateContinuation.bind(workspace) + workspace.completeHumanGateContinuation = (...args) => { + const result = complete(...args) + if (args[2] === 'completed') { + fs.copyFileSync(storagePath, recoveryPath) + if (fs.existsSync(`${storagePath}.journal`)) fs.copyFileSync(`${storagePath}.journal`, `${recoveryPath}.journal`) + throw new Error('TEST_CRASH_AFTER_INPUT_CHECKPOINT') + } + return result + } + workspace.decideHumanGate(gate.gateId, { optionId: 'respond', response: 'USER_CHOICE_B' }) + await controller.promise + const ledger = new RunLedger({ storagePath: recoveryPath }) + assert.equal(ledger.get(controller.runId).continuation.state, 'completed') + const before = calls.length + const recovered = new LocalWorkspace({ ...workspaceOptions, runLedger: ledger }) + workspaces.push(recovered) + await recovered.refreshAgents() + const final = await until(() => { + const run = ledger.get(controller.runId) + return ['completed', 'partial', 'failed', 'stopped'].includes(run.status) ? run : null + }) + assert.equal(final.status, 'completed', final.reason) + assert.equal(calls.length, before + 1) + assert.equal(recovered.listHumanGates({ runId: controller.runId }).length, 1) +}) diff --git a/docs/releases/Meldwork-V1.0.5-progress.md b/docs/releases/Meldwork-V1.0.5-progress.md index db1c1f0..0d9f66e 100644 --- a/docs/releases/Meldwork-V1.0.5-progress.md +++ b/docs/releases/Meldwork-V1.0.5-progress.md @@ -212,3 +212,19 @@ nextKinds 使用最多 32 个唯一标识，与现有 V4 成员上限一致； - `git diff --check` 通过。本轮未改前端源码，未重跑前端测试、构建或打包；桌面实际运行与历史重开验证如上。 本检查点未改变人类采用的含义，needs-human 仍待接入可恢复人类决定流程。尚未更新版本号、构建、打包或替换日常应用。 + +## 自然讨论的人类输入与重启恢复 + +2026-09-09：交付负责人返回 needs-human 后，通过现有输入 Gate 等待用户澄清，回复后继续同一任务，不再直接结束为 partial。新增 v4_task_decision continuation，绑定实际已完成的来源调用、讨论轮次、slot、operation、任务快照和判断哈希；重启恢复验证来源与当前负责人，避免重放已完成的提问。原生 CLI 暂停输入仍使用既有 session/request 绑定，用户澄清不会增加工作区写权限。 + +重复提交已记录 Gate 决定时保留原决定时间，并在校验内容一致后返回，不重复触发恢复回调；冲突决定仍被拒绝。关闭应用时保留待输入 Gate，取消输入则停止任务。已批准的回答经来源校验后进入现有有界上下文打包。 + +验证： + +- 新增任务输入测试 7/7，覆盖 agent-led/sequential、取消、等待时重启、来源与选项篡改、批准后分发前恢复、重复提交，以及 continuation 检查点后下一次调用前恢复。Gate/账本/任务输入针对性验证 74/74。 +- 最终桌面全量 1549/1549，失败和跳过均为 0，退出码 0，约 311 秒。命令为 `MELDWORK_TEST_CLAUDE_EXECUTABLE=/Users/rydersun/.local/opt/npm-global/bin/claude npm --prefix desktop test`，日志 `/tmp/meldwork-105-human-task-desktop.log`；该轮包含最终产品代码。 +- 真实 Electron 脚本 `/tmp/meldwork-105-human-task-ui.cjs /tmp/meldwork-105-human-task-3UxpAQ` 退出 0，正常关闭应用。复用已有任务，两次调用完成后账本 waiting；关闭并重开仍保留输入，界面提交后仅新增一次调用，答案为 HUMAN_DECISION_105_OK，任务和 continuation 均 completed。重复通过真实 preload 提交相同回答，调用总数仍为 3；群组 allowWrite 仍为 false。 +- 已查看 `/tmp/meldwork-105-human-task-waiting.png` 和 `/tmp/meldwork-105-human-task-completed.png`，确认恢复后的输入入口与最终完成卡片。等待截图捕获在面板过渡过程中，不能用其透明度判断静止界面样式。 +- 前期验证脚本曾因 onboarding 遮挡、异步 waitForFunction 判断及误读活动快照 status 而退出或超时，未计入成功。最终使用真实账本及 waitingGateIds 判断，并复用已有等待任务，没有重发任务制造新样本。 + +本检查点未修改前端源码，未重新构建或打包，版本仍为 0.1.4。剩余通用调度耦合、CLI 环境覆盖与最终版本验收继续推进；此输入流程不代表人类采用或写入授权。