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.
This commit is contained in:
Ryder Sun committed 2026-09-09 20:57:13 +08:00
1 parent ee0bd66708
commit eb0e1bb2ef
8 files changed
+360 -12

No files matched your search

+3 -1
View File
@@ -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)
+7 -4
View File
@@ -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'
@@ -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)) {
+110 -1
View File
@@ -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)
@@ -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()
@@ -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 }))
@@ -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)
})
+16
View File
@@ -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 环境覆盖与最终版本验收继续推进;此输入流程不代表人类采用或写入授权。