diff --git a/desktop/src/agents/agent-permission-policy.cjs b/desktop/src/agents/agent-permission-policy.cjs index fb4bfe9..a171489 100644 --- a/desktop/src/agents/agent-permission-policy.cjs +++ b/desktop/src/agents/agent-permission-policy.cjs @@ -1,3 +1,10 @@ +const { createHash } = require('node:crypto') +const { canonicalJson } = require('../collaboration/context-pack-records.cjs') + +function sha256(value) { + return createHash('sha256').update(String(value || '')).digest('hex') +} + // Tool kinds an Agent may use without approval during a read-only turn. const READ_ONLY_SAFE_TOOL_KINDS = new Set(['read', 'search', 'fetch', 'think']) @@ -22,4 +29,49 @@ function automaticPermissionResponse(request, { allowWrite }) { return option?.optionId ? { status: 'approved', optionId: option.optionId } : undefined } -module.exports = { READ_ONLY_SAFE_TOOL_KINDS, automaticPermissionResponse } +// A person's decision applies only to the exact request it was made for. +function matchesResumedPermission(request, resumedGate) { + if (!request || !resumedGate?.request) return false + try { + if (canonicalJson(request) !== canonicalJson(resumedGate.request)) return false + } catch { + return false + } + const selected = request.options?.find(option => option.optionId === resumedGate.optionId) + if (!selected) return false + return resumedGate.status === 'approved' + ? selected.kind.startsWith('allow_') + : selected.kind.startsWith('reject_') +} + +function permissionContinuationRequestId(requestHash, binding) { + return sha256(canonicalJson({ + requestHash, + sessionRefHash: binding.sessionRefHash, + sessionProvenanceHash: binding.sessionProvenanceHash, + })) +} + +// A resumed permission decision must belong to the native session the +// request came from; anything else is refused before the Agent runs. +function assertResumedPermissionBinding(resumed, currentBinding, sessionRef) { + const requestHash = sha256(canonicalJson(resumed.request)) + const persistedBinding = { + sessionRefHash: resumed.sessionRefHash, + sessionProvenanceHash: resumed.sessionProvenanceHash, + } + if (resumed.requestId !== permissionContinuationRequestId(requestHash, persistedBinding) + || resumed.requestHash !== requestHash + || resumed.sessionProvenanceHash !== currentBinding.sessionProvenanceHash + || (sessionRef && resumed.sessionRefHash !== currentBinding.sessionRefHash)) { + throw new Error('LOCAL_RUN_PERMISSION_REQUEST_MISMATCH') + } +} + +module.exports = { + READ_ONLY_SAFE_TOOL_KINDS, + assertResumedPermissionBinding, + automaticPermissionResponse, + matchesResumedPermission, + permissionContinuationRequestId, +} diff --git a/desktop/src/workspace/invocation-budget.cjs b/desktop/src/workspace/invocation-budget.cjs new file mode 100644 index 0000000..1f79042 --- /dev/null +++ b/desktop/src/workspace/invocation-budget.cjs @@ -0,0 +1,37 @@ +const { outboundWirePayloadBytes } = require('../collaboration/outbound-payload.cjs') + +// What the Agent was sent: wire bytes are reported, tokens are an estimate. +function recordOutboundUsage(budget, outbound) { + if (!budget) return + const bytes = outboundWirePayloadBytes(outbound) + // A payload that was not built by the transport has no recorded wire size; + // skipping it is better than failing the Agent's turn over bookkeeping. + if (!bytes) return + budget.addUsageBatch([ + { dimension: 'outboundBytes', amount: bytes.length, source: 'reported' }, + { dimension: 'inputTokens', amount: Math.ceil(bytes.length / 4), source: 'estimated' }, + ]) +} + +// What the Agent produced: prefer the usage the CLI reported over an estimate. +function recordReplyUsage(budget, { usage, replyText, toolCallCount }) { + if (!budget) return + const reportedOutputTokens = Number(usage?.outputTokens) + const hasReportedOutputTokens = Number.isSafeInteger(reportedOutputTokens) + && reportedOutputTokens >= 0 + const reportedCostMicros = Number(usage?.costMicros) + budget.addUsageBatch([ + { + dimension: 'outputTokens', + amount: hasReportedOutputTokens ? reportedOutputTokens : Math.ceil(replyText.length / 4), + source: hasReportedOutputTokens ? 'reported' : 'estimated', + }, + { dimension: 'toolCalls', amount: toolCallCount, source: 'estimated' }, + ...(Number.isSafeInteger(reportedCostMicros) && reportedCostMicros >= 0 + ? [{ dimension: 'costMicros', amount: reportedCostMicros, source: 'reported' }] + : []), + { dimension: 'elapsedMs', value: budget.elapsedValue(), source: 'reported' }, + ]) +} + +module.exports = { recordOutboundUsage, recordReplyUsage } diff --git a/desktop/src/workspace/invocation-failure.cjs b/desktop/src/workspace/invocation-failure.cjs new file mode 100644 index 0000000..8a4203e --- /dev/null +++ b/desktop/src/workspace/invocation-failure.cjs @@ -0,0 +1,72 @@ +const { agentStoppedError } = require('./local-workspace-inputs.cjs') + +// Marks how far an invocation got before it failed, so the run can decide +// whether the operation may be retried safely. +function attachInvocationFailure(error, input) { + if (!error || (typeof error !== 'object' && typeof error !== 'function')) return error + const existing = error.invocationFailure && typeof error.invocationFailure === 'object' + ? error.invocationFailure + : {} + Object.defineProperty(error, 'invocationFailure', { + value: Object.freeze({ + outcomeCertainty: existing.outcomeCertainty || input.outcomeCertainty, + sideEffectsPossible: existing.sideEffectsPossible === true || input.sideEffectsPossible === true, + operationId: existing.operationId || input.operationId, + idempotencyMode: existing.idempotencyMode || input.idempotencyMode, + }), + enumerable: false, + configurable: true, + }) + return error +} + +// Hidden properties on an error that downstream code reads but that must not +// leak into serialised error messages. +function attachErrorMetadata(error, metadata) { + if (!error || (typeof error !== 'object' && typeof error !== 'function')) return error + for (const [name, value] of Object.entries(metadata)) { + if (!value) continue + Object.defineProperty(error, name, { value, enumerable: false, configurable: true }) + } + return error +} + +// Why an invocation ended and how to report it: +// - the parent run was stopped, timed out or is shutting down; +// - the watchdog cut a silent Agent off; +// - the Agent was cancelled on its own (e.g. a superseded attempt); +// - or it simply failed with `caughtError`. +// `settleOutstanding` tells the caller to wait briefly for capture and run +// promises that may still be winding down. +function resolveInvocationStop({ + caughtError, signal, parentAbortObserved, activeRun, agentController, watchdog, +}) { + const parentTimedOut = Boolean(signal?.aborted && activeRun?.stopReason === 'timeout') + const parentStopped = Boolean(signal?.aborted || parentAbortObserved) + const parentInterrupted = parentStopped && activeRun?.stopReason === 'shutdown' + const agentCancelled = agentController.signal.aborted && !parentStopped && !watchdog.timedOut + let error = caughtError + let status = 'failed' + if (parentTimedOut) { + error = new Error('LOCAL_AGENT_TIMEOUT') + status = 'timeout' + } else if (parentStopped) { + error = caughtError?.message === 'LOCAL_AGENT_EXECUTION_STOPPED' + ? caughtError + : agentStoppedError() + status = parentInterrupted ? 'interrupted' : 'stopped' + } else if (watchdog.timedOut) { + error = watchdog.error || new Error('LOCAL_AGENT_TIMEOUT') + status = 'timeout' + } else if (agentCancelled) { + error = agentStoppedError() + status = 'stopped' + } + return { + error, + status, + settleOutstanding: parentStopped || watchdog.timedOut || agentCancelled, + } +} + +module.exports = { attachErrorMetadata, attachInvocationFailure, resolveInvocationStop } diff --git a/desktop/src/workspace/invocation-outputs.cjs b/desktop/src/workspace/invocation-outputs.cjs new file mode 100644 index 0000000..822a194 --- /dev/null +++ b/desktop/src/workspace/invocation-outputs.cjs @@ -0,0 +1,113 @@ +const { + MAX_MESSAGE_ATTACHMENTS, + abortableOperation, + normalizeAttachmentMetadata, +} = require('./local-workspace-inputs.cjs') + +// Absolute paths an Agent mentions in its reply. Quoted, backticked or +// Markdown-linked paths may contain spaces; bare paths end at whitespace. +function extractReportedOutputPaths(text) { + if (typeof text !== 'string') return [] + const root = '\\/(?:Users|home|tmp|private\\/tmp|Volumes)\\/' + const delimited = [...text.matchAll(new RegExp(`[\`"'(<](${root}[^\`"'<>\\n)]+)[\`"')>]`, 'g'))] + .map(match => match[1]) + const bare = (text.match(new RegExp(`${root}[^\\s"'<>\`\\]\\[)]+`, 'g')) || []) + .filter(value => !delimited.some(full => full !== value && full.startsWith(value))) + return [...new Set([...delimited, ...bare].map(value => value.trim().replace(/[.,;:!?]+$/, '')))] +} + +// Files a writable Agent turn leaves behind: a snapshot of the workspace is +// taken before the turn and compared afterwards to import generated media and +// to describe produced artifacts. Capturing is best effort; only a stop +// request is allowed to interrupt it. +class InvocationOutputs { + constructor({ enabled, workdir, signal, agentKind, startedAt, deps }) { + this.enabled = enabled === true + this.workdir = workdir + this.signal = signal + this.agentKind = agentKind + this.startedAt = startedAt + this.deps = deps + this.baseline = null + this.artifactBaseline = null + this.capturePromise = null + this.artifactCapturePromise = null + this.importPromise = null + this.outcomePromise = null + } + + // Promises that may still be running when the invocation is cut short. + pending() { + return [ + this.capturePromise, + this.artifactCapturePromise, + this.importPromise, + this.outcomePromise, + ].filter(Boolean) + } + + async tolerate(operation) { + try { + return await abortableOperation(operation, this.signal) + } catch (error) { + if (this.signal.aborted) throw error + return undefined + } + } + + async begin() { + if (!this.enabled) return + this.artifactCapturePromise = Promise.resolve().then( + () => this.deps.captureArtifactOutputs(this.workdir, { signal: this.signal }), + ) + this.artifactCapturePromise.catch(() => {}) + try { + this.capturePromise = Promise.resolve().then( + () => this.deps.captureAgentOutputs(this.workdir, { signal: this.signal }), + ) + this.capturePromise.catch(() => {}) + this.baseline = await abortableOperation(() => this.capturePromise, this.signal) + } catch (error) { + if (this.signal.aborted) throw error + } + this.artifactBaseline = (await this.tolerate(() => this.artifactCapturePromise)) ?? null + } + + // Media the Agent produced during this turn, as message attachments. + async importMedia(replyText) { + if (!this.enabled) return [] + this.importPromise = Promise.resolve().then(() => this.deps.importAgentOutputs({ + workdir: this.workdir, + baseline: this.baseline, + startedAt: this.startedAt, + agentKind: this.agentKind, + reportedPaths: extractReportedOutputPaths(replyText), + signal: this.signal, + })) + this.importPromise.catch(() => {}) + const imported = await this.tolerate(() => this.importPromise) + return (Array.isArray(imported) ? imported : []) + .slice(0, MAX_MESSAGE_ATTACHMENTS) + .map(normalizeAttachmentMetadata) + .filter(Boolean) + } + + // Descriptors of the artifacts the Agent wrote, for the evidence record. + async describeOutcomes({ runId, agentRunId }) { + if (!this.enabled || !this.artifactBaseline) return [] + this.outcomePromise = Promise.resolve().then(() => this.deps.captureAgentOutcomeDescriptors({ + workdir: this.workdir, + baseline: this.artifactBaseline, + startedAt: this.startedAt, + agentKind: this.agentKind, + runId, + agentRunId, + signal: this.signal, + })) + this.outcomePromise.catch(() => {}) + const captured = await this.tolerate(() => this.outcomePromise) + return Array.isArray(captured) ? captured : [] + } +} + +module.exports = { InvocationOutputs, extractReportedOutputPaths } diff --git a/desktop/src/workspace/invocation-prompt.cjs b/desktop/src/workspace/invocation-prompt.cjs new file mode 100644 index 0000000..f8171d0 --- /dev/null +++ b/desktop/src/workspace/invocation-prompt.cjs @@ -0,0 +1,59 @@ +const { stagedAgentInputPrompt } = require('../agents/agent-input-staging.cjs') +const { directPromptFor } = require('./local-workspace-context.cjs') +const { cleanText } = require('./local-workspace-inputs.cjs') + +const RESPONSE_SCOPE_INSTRUCTION = [ + 'Final response scope:', + 'Answer only the current user task stated above.', + 'Use earlier messages only as supporting evidence when they help answer that current task.', + 'Do not separately mention, summarize, continue, or complete any previous user task.', + 'Do not append an answer to an older task after the current-task conclusion.', +].join('\n') + +// Builds the prompt one Agent receives for one turn. +// - frozen / isolated invocations send exactly the approved prompt. +// - direct chats lead with the user's request; a resumed native session +// already holds the conversation, a new or rebuilt one gets the history. +// - group turns use the V4 prompt when the orchestrator supplied one, else +// the workspace prompt for the current mode. +// Staged attachments and harness recovery instructions are appended to both. +function assembleInvocationPrompt({ + group, kind, mode, threadRootId, context, activeRun, isolated, frozen, + v4Prompt, promptMode, afterKind, packed, session, stagedInputs, promptFor, agentLabel, +}) { + if (isolated || frozen) return v4Prompt + const runtimeInstruction = cleanText(context.runtimeInstruction, 3000) + const regenerationInstruction = cleanText(context.regenerationInstruction, 3000) + const additions = [ + stagedAgentInputPrompt(stagedInputs), + runtimeInstruction ? `Harness recovery task:\n${runtimeInstruction}` : '', + regenerationInstruction ? `Fresh response instruction:\n${regenerationInstruction}` : '', + ] + if (group.conversationType === 'direct') { + return [ + directPromptFor({ + group, + label: agentLabel(kind), + packed, + skillHints: context.skillHints || [], + knowledgeBaseHints: context.knowledgeBaseHints || [], + includeHistory: session.rotated || !session.ref, + }), + ...additions, + ].filter(Boolean).join('\n\n') + } + return [ + context.v4 === true && v4Prompt + ? v4Prompt + : promptFor( + group, kind, context.completionPolicy === 'typed' ? 'manual' : mode, + threadRootId, context.skillHints || [], + context.knowledgeBaseHints || [], afterKind, packed, promptMode, + activeRun?.unlimitedRounds === true, + ), + ...additions, + RESPONSE_SCOPE_INSTRUCTION, + ].filter(Boolean).join('\n') +} + +module.exports = { assembleInvocationPrompt } diff --git a/desktop/src/workspace/invocation-session.cjs b/desktop/src/workspace/invocation-session.cjs new file mode 100644 index 0000000..2bcd846 --- /dev/null +++ b/desktop/src/workspace/invocation-session.cjs @@ -0,0 +1,156 @@ +const { createHash, randomUUID } = require('node:crypto') +const { canonicalJson } = require('../collaboration/context-pack-records.cjs') +const { nextSessionMeta, normalizeSessionMeta } = require('../runs/run-harness.cjs') + +const MAX_INHERITED_TASK_IDS = 64 + +function sha256(value) { + return createHash('sha256').update(String(value || '')).digest('hex') +} + +function createdSessionProvenance(taskId, stateless = false) { + if (stateless) { + return { + scope: 'none', reuse: false, origin: 'none', originTaskId: null, + inheritedTaskIds: [], completeness: 'complete', + } + } + return { + scope: 'conversation', + reuse: false, + origin: 'created', + originTaskId: taskId || null, + inheritedTaskIds: [], + completeness: 'complete', + } +} + +function provenanceMeta(provenance) { + if (provenance.scope === 'unknown-legacy') { + return { + sessionScope: 'unknown-legacy', + originTaskId: '', + inheritedTaskIds: [], + provenanceCompleteness: 'unknown-legacy', + } + } + if (provenance.scope === 'none') return {} + return { + sessionScope: provenance.scope, + originTaskId: provenance.originTaskId || '', + inheritedTaskIds: provenance.inheritedTaskIds || [], + provenanceCompleteness: provenance.completeness, + } +} + +function completedSessionMeta(meta, provenance, taskId, usage) { + let inheritedTaskIds = [...(provenance.inheritedTaskIds || [])] + let completeness = provenance.completeness + if (provenance.scope === 'conversation' && taskId + && taskId !== provenance.originTaskId && !inheritedTaskIds.includes(taskId)) { + inheritedTaskIds.push(taskId) + if (inheritedTaskIds.length > MAX_INHERITED_TASK_IDS) { + inheritedTaskIds = inheritedTaskIds.slice(-MAX_INHERITED_TASK_IDS) + completeness = 'partial' + } + } + return nextSessionMeta(meta, { + ...usage, + ...provenanceMeta({ ...provenance, inheritedTaskIds, completeness }), + }) +} + +// Binds a pending permission or connector request to the exact native session +// it was raised in, so a decision cannot be replayed against another session. +function connectorSessionBinding(sessionRef, sessionProvenance) { + const stableProvenance = { + scope: String(sessionProvenance?.scope || 'none'), + originTaskId: String(sessionProvenance?.originTaskId || ''), + inheritedTaskIds: Array.isArray(sessionProvenance?.inheritedTaskIds) + ? [...sessionProvenance.inheritedTaskIds] + : [], + completeness: String(sessionProvenance?.completeness || 'complete'), + } + return { + sessionRefHash: sha256(sessionRef), + sessionProvenanceHash: sha256(canonicalJson(stableProvenance)), + } +} + +// The native Agent session one invocation runs in. A conversation owns one +// native session for its lifetime: it is resumed on every turn and rebuilt +// only when the Agent reports that the stored session is no longer valid. +class InvocationSession { + // `resolved` is what the workspace stored for this conversation; `deps` + // carries the state hooks so this class never touches workspace state itself. + constructor({ resolved, group, kind, taskId, deps }) { + this.group = group + this.kind = kind + this.taskId = taskId + this.deps = deps + this.key = resolved.key + this.storedRef = String(resolved.sessionRef || '') + this.ref = this.storedRef + this.meta = normalizeSessionMeta(resolved.sessionMeta) + this.provenance = resolved.provenance + this.rotated = resolved.sessionReset === true + this.transport = this.ref ? String(this.meta.transport || '') : '' + } + + binding() { + return connectorSessionBinding(this.ref, this.provenance) + } + + // Forget the native session so the next turn starts a fresh one. OpenClaw + // addresses its sessions by name, so it gets a new name straight away. + discard() { + this.meta = {} + this.provenance = createdSessionProvenance(this.taskId) + this.transport = '' + this.rotated = true + this.deps.commitSessionState((nextState) => { + delete nextState.sessions[this.key] + delete nextState.sessionMeta[this.key] + if (this.kind === 'openclaw') { + const generation = randomUUID().replace(/[^a-zA-Z0-9]/g, '').slice(0, 12) || 'session' + this.ref = this.deps.openClawSessionRef( + this.group, generation, this.group.conversationType === 'direct' ? '' : this.taskId, + ) + nextState.sessions[this.key] = this.ref + nextState.sessionMeta[this.key] = normalizeSessionMeta(provenanceMeta(this.provenance)) + } else { + this.ref = '' + } + }) + } + + // The Agent reported (or created) its native session id. + adopt(nextRef, metadata = {}) { + const transport = ['legacy', 'acp'].includes(metadata?.transport) ? metadata.transport : '' + if (transport) this.transport = transport + this.ref = String(nextRef || '') + this.deps.persistSessionState(this.key, this.ref, normalizeSessionMeta({ + ...this.meta, + ...provenanceMeta(this.provenance), + ...(this.transport ? { transport: this.transport } : {}), + })) + } + + // A turn finished: record what it cost and which message it had seen. + complete(resultRef, usage) { + this.deps.persistSessionState( + this.key, + resultRef || this.ref, + completedSessionMeta(this.meta, this.provenance, this.taskId, usage), + ) + } +} + +module.exports = { + InvocationSession, + completedSessionMeta, + connectorSessionBinding, + createdSessionProvenance, + provenanceMeta, + sha256, +} diff --git a/desktop/src/workspace/local-workspace-agent-invocation.cjs b/desktop/src/workspace/local-workspace-agent-invocation.cjs index fe48eec..582c44d 100644 --- a/desktop/src/workspace/local-workspace-agent-invocation.cjs +++ b/desktop/src/workspace/local-workspace-agent-invocation.cjs @@ -1,51 +1,60 @@ const { createHash, randomUUID } = require('node:crypto') const fs = require('node:fs') const path = require('node:path') -const { createAnswerMarkerFilter } = require('../collaboration/answer-marker-filter.cjs') const { - V4_AGENT_PHASES, - parseV4CollaborationReceipt, -} = require('../collaboration/collaboration-receipt.cjs') -const { automaticPermissionResponse } = require('../agents/agent-permission-policy.cjs') -const { createInvocationWatchdog } = require('../runs/invocation-watchdog.cjs') + assertResumedPermissionBinding, + automaticPermissionResponse, + matchesResumedPermission, + permissionContinuationRequestId, +} = require('../agents/agent-permission-policy.cjs') const { agentRuntimeCapabilities, isCodeReviewAgentKind, requireTerminalAgentResult, } = require('../agents/agent-runtime-contract.cjs') const { - nextSessionMeta, - normalizeOutcomeRefs, - normalizeSessionMeta, -} = require('../runs/run-harness.cjs') + cleanupStagedAgentInputs, + stageAgentInputs, +} = require('../agents/agent-input-staging.cjs') +const { createAnswerMarkerFilter } = require('../collaboration/answer-marker-filter.cjs') +const { + V4_AGENT_PHASES, + parseV4CollaborationReceipt, +} = require('../collaboration/collaboration-receipt.cjs') const { canonicalJson, normalizeContextPackId, } = require('../collaboration/context-pack-records.cjs') +const { normalizeOutcomeRefs, normalizeSessionMeta } = require('../runs/run-harness.cjs') +const { createInvocationWatchdog } = require('../runs/invocation-watchdog.cjs') +const { processRunScheduler } = require('../runs/run-scheduler.cjs') +const { assertLocalSkillExecution } = require('../skills/local-skill-contract.cjs') const { defaultAgentLabel, HERMES_WORKSPACE_ACP_ENABLED, - MAX_MESSAGE_ATTACHMENTS, - abortableOperation, agentStoppedError, attachmentType, cleanText, cleanProgressSteps, - normalizeAttachmentMetadata, parseAutoReply, settleWithin, } = require('./local-workspace-inputs.cjs') -const { assertLocalSkillExecution } = require('../skills/local-skill-contract.cjs') -const { directPromptFor } = require('./local-workspace-context.cjs') -const { outboundWirePayloadBytes } = require('../collaboration/outbound-payload.cjs') -const { processRunScheduler } = require('../runs/run-scheduler.cjs') +const { recordOutboundUsage, recordReplyUsage } = require('./invocation-budget.cjs') const { - cleanupStagedAgentInputs, - stageAgentInputs, - stagedAgentInputPrompt, -} = require('../agents/agent-input-staging.cjs') + attachErrorMetadata, + attachInvocationFailure, + resolveInvocationStop, +} = require('./invocation-failure.cjs') +const { InvocationOutputs } = require('./invocation-outputs.cjs') +const { assembleInvocationPrompt } = require('./invocation-prompt.cjs') +const { + InvocationSession, + connectorSessionBinding, + createdSessionProvenance, + provenanceMeta, + sha256, +} = require('./invocation-session.cjs') -const MAX_INHERITED_TASK_IDS = 64 const MAX_ISOLATED_PROMPT_BYTES = 4 * 1024 * 1024 const OUTCOME_REF_FIELDS = Object.freeze([ 'artifactIds', @@ -56,18 +65,6 @@ const OUTCOME_REF_FIELDS = Object.freeze([ 'workflowOutcomeRefs', ]) -function extractReportedOutputPaths(text) { - if (typeof text !== 'string') return [] - const root = '\\/(?:Users|home|tmp|private\\/tmp|Volumes)\\/' - // Quoted, backticked or Markdown-linked paths may contain spaces; bare - // paths end at whitespace. - const delimited = [...text.matchAll(new RegExp(`[\`"'(<](${root}[^\`"'<>\\n)]+)[\`"')>]`, 'g'))] - .map(match => match[1]) - const bare = (text.match(new RegExp(`${root}[^\\s"'<>\`\\]\\[)]+`, 'g')) || []) - .filter(value => !delimited.some(full => full !== value && full.startsWith(value))) - return [...new Set([...delimited, ...bare].map(value => value.trim().replace(/[.,;:!?]+$/, '')))] -} - function mergeOutcomeRefs(sources, options = {}) { const merged = {} const seen = new Map(OUTCOME_REF_FIELDS.map(field => [field, new Set()])) @@ -96,24 +93,6 @@ const SHA256 = /^[a-f0-9]{64}$/ -function attachInvocationFailure(error, input) { - if (!error || (typeof error !== 'object' && typeof error !== 'function')) return error - const existing = error.invocationFailure && typeof error.invocationFailure === 'object' - ? error.invocationFailure - : {} - Object.defineProperty(error, 'invocationFailure', { - value: Object.freeze({ - outcomeCertainty: existing.outcomeCertainty || input.outcomeCertainty, - sideEffectsPossible: existing.sideEffectsPossible === true || input.sideEffectsPossible === true, - operationId: existing.operationId || input.operationId, - idempotencyMode: existing.idempotencyMode || input.idempotencyMode, - }), - enumerable: false, - configurable: true, - }) - return error -} - function isolatedInvocationContext(context) { if (!['isolated', 'frozen'].includes(context?.sessionPolicy) && !(context?.v4 === true && typeof context.promptOverride === 'string')) return null @@ -133,84 +112,6 @@ function isolatedInvocationContext(context) { } } -function createdSessionProvenance(group, taskId, stateless = false) { - if (stateless) { - return { - scope: 'none', reuse: false, origin: 'none', originTaskId: null, - inheritedTaskIds: [], completeness: 'complete', - } - } - return { - scope: 'conversation', - reuse: false, - origin: 'created', - originTaskId: taskId || null, - inheritedTaskIds: [], - completeness: 'complete', - } -} - -function provenanceMeta(provenance) { - if (provenance.scope === 'unknown-legacy') { - return { - sessionScope: 'unknown-legacy', - originTaskId: '', - inheritedTaskIds: [], - provenanceCompleteness: 'unknown-legacy', - } - } - if (provenance.scope === 'none') return {} - return { - sessionScope: provenance.scope, - originTaskId: provenance.originTaskId || '', - inheritedTaskIds: provenance.inheritedTaskIds || [], - provenanceCompleteness: provenance.completeness, - } -} - -function completedSessionMeta(meta, provenance, taskId, usage) { - let inheritedTaskIds = [...(provenance.inheritedTaskIds || [])] - let completeness = provenance.completeness - if (provenance.scope === 'conversation' && taskId - && taskId !== provenance.originTaskId && !inheritedTaskIds.includes(taskId)) { - inheritedTaskIds.push(taskId) - if (inheritedTaskIds.length > MAX_INHERITED_TASK_IDS) { - inheritedTaskIds = inheritedTaskIds.slice(-MAX_INHERITED_TASK_IDS) - completeness = 'partial' - } - } - return nextSessionMeta(meta, { - ...usage, - ...provenanceMeta({ ...provenance, inheritedTaskIds, completeness }), - }) -} - -function matchesResumedPermission(request, resumedGate) { - if (!request || !resumedGate?.request) return false - try { - if (canonicalJson(request) !== canonicalJson(resumedGate.request)) return false - } catch { - return false - } - const selected = request.options?.find(option => option.optionId === resumedGate.optionId) - if (!selected) return false - return resumedGate.status === 'approved' - ? selected.kind.startsWith('allow_') - : selected.kind.startsWith('reject_') -} - -function sha256(value) { - return createHash('sha256').update(String(value || '')).digest('hex') -} - -function permissionContinuationRequestId(requestHash, binding) { - return sha256(canonicalJson({ - requestHash, - sessionRefHash: binding.sessionRefHash, - sessionProvenanceHash: binding.sessionProvenanceHash, - })) -} - function deliveredPackedContext(packedContext, promptMode) { if (promptMode !== 'continuation') return packedContext return { @@ -258,21 +159,6 @@ function assertRequiredContextFits(packedContext) { throw error } -function connectorSessionBinding(sessionRef, sessionProvenance) { - const stableProvenance = { - scope: String(sessionProvenance?.scope || 'none'), - originTaskId: String(sessionProvenance?.originTaskId || ''), - inheritedTaskIds: Array.isArray(sessionProvenance?.inheritedTaskIds) - ? [...sessionProvenance.inheritedTaskIds] - : [], - completeness: String(sessionProvenance?.completeness || 'complete'), - } - return { - sessionRefHash: sha256(sessionRef), - sessionProvenanceHash: sha256(canonicalJson(stableProvenance)), - } -} - class LocalWorkspaceAgentInvocation { constructor(options) { this.state = options.state @@ -528,22 +414,26 @@ class LocalWorkspaceAgentInvocation { const idempotencyMode = agent.idempotencyMode === 'durable' ? 'durable' : 'none' // Isolated invocations must see only their approved prompt, so they never // resume or persist the conversation's native session. - const resolvedSession = reviewOnly || isolated - ? { - key: this.sessionKey(group.id, kind), - sessionRef: '', - sessionMeta: {}, - provenance: createdSessionProvenance(group, taskId, true), - } - : this.sessionState( - group, kind, context.sessionThreadRootId || threadRootId, taskId, - ) - const key = resolvedSession.key - const storedSessionRef = String(resolvedSession.sessionRef || '') - let sessionRef = storedSessionRef - let sessionMeta = normalizeSessionMeta(resolvedSession.sessionMeta) - let sessionProvenance = resolvedSession.provenance - let sessionRotated = resolvedSession.sessionReset === true + const session = new InvocationSession({ + resolved: reviewOnly || isolated + ? { + key: this.sessionKey(group.id, kind), + sessionRef: '', + sessionMeta: {}, + provenance: createdSessionProvenance(taskId, true), + } + : this.sessionState( + group, kind, context.sessionThreadRootId || threadRootId, taskId, + ), + group, + kind, + taskId, + deps: { + commitSessionState: mutator => this.commitSessionState(mutator), + openClawSessionRef: (...args) => this.openClawSessionRef(...args), + persistSessionState: (...args) => this.persistSessionState(...args), + }, + }) const resumedConnectorGate = ['input', 'permission'].includes(context.resumedGate?.type) && context.resumedGate?.request?.source === 'connector' ? context.resumedGate @@ -552,77 +442,25 @@ class LocalWorkspaceAgentInvocation { && !resumedConnectorGate ? context.resumedGate : null - // A conversation owns one native session for its lifetime. Rebuild only - // when the Agent reports that the existing native session is invalid. - const sessionNeedsRotation = false if (resumedPermission) { - const requestHash = sha256(canonicalJson(resumedPermission.request)) - const persistedBinding = { - sessionRefHash: resumedPermission.sessionRefHash, - sessionProvenanceHash: resumedPermission.sessionProvenanceHash, - } - const binding = connectorSessionBinding(sessionRef, sessionProvenance) - if (resumedPermission.requestId !== permissionContinuationRequestId( - requestHash, persistedBinding, - ) || resumedPermission.requestHash !== requestHash - || resumedPermission.sessionProvenanceHash !== binding.sessionProvenanceHash - || (sessionRef && resumedPermission.sessionRefHash !== binding.sessionRefHash)) { - throw new Error('LOCAL_RUN_PERMISSION_REQUEST_MISMATCH') - } + assertResumedPermissionBinding(resumedPermission, session.binding(), session.ref) + if (!session.ref) throw new Error('LOCAL_RUN_PERMISSION_RESUME_UNAVAILABLE') } - if (resumedPermission && (!sessionRef || sessionNeedsRotation)) { - if (context.v4 === true && !allowWrite && sessionRef) resumedPermission = null - else throw new Error('LOCAL_RUN_PERMISSION_RESUME_UNAVAILABLE') - } - if (sessionNeedsRotation) { - sessionMeta = {} - sessionProvenance = createdSessionProvenance(group, taskId) - this.commitSessionState((nextState) => { - delete nextState.sessions[key] - delete nextState.sessionMeta[key] - if (kind === 'openclaw') { - const generation = randomUUID().replace(/[^a-zA-Z0-9]/g, '').slice(0, 12) || 'session' - sessionRef = this.openClawSessionRef( - group, generation, group.conversationType === 'direct' ? '' : taskId, - ) - nextState.sessions[key] = sessionRef - nextState.sessionMeta[key] = normalizeSessionMeta( - provenanceMeta(sessionProvenance), - ) - } else { - sessionRef = '' - } - }) - sessionRotated = true - } - let sessionTransport = sessionRef ? String(sessionMeta.transport || '') : '' - const hermesNeedsLegacy = kind === 'hermes' && sessionTransport === 'acp' - && (!HERMES_WORKSPACE_ACP_ENABLED - || agent.acpAvailable === false - ) + const hermesNeedsLegacy = kind === 'hermes' && session.transport === 'acp' + && (!HERMES_WORKSPACE_ACP_ENABLED || agent.acpAvailable === false) if (resumedPermission && hermesNeedsLegacy) { throw new Error('LOCAL_RUN_PERMISSION_RESUME_UNAVAILABLE') } - if (sessionRef && hermesNeedsLegacy) { - this.commitSessionState((nextState) => { - delete nextState.sessions[key] - delete nextState.sessionMeta[key] - }) - sessionRef = '' - sessionMeta = {} - sessionProvenance = createdSessionProvenance(group, taskId) - sessionTransport = '' - sessionRotated = true - } + if (session.ref && hermesNeedsLegacy) session.discard() let v4Prompt = invocationContext?.promptOverride || '' let v4Delivery = null const buildV4Prompt = () => { if (context.v4 !== true || typeof context.v4PromptBuilder !== 'function') return v4Prompt const built = context.v4PromptBuilder({ - sessionRefHash: sha256(sessionRef), - sessionProvenanceHash: connectorSessionBinding(sessionRef, sessionProvenance).sessionProvenanceHash, - sessionRotated, - hasSession: Boolean(sessionRef), + sessionRefHash: sha256(session.ref), + sessionProvenanceHash: session.binding().sessionProvenanceHash, + sessionRotated: session.rotated, + hasSession: Boolean(session.ref), }) if (!built || typeof built.prompt !== 'string' || !built.prompt || built.prompt.includes('\u0000') || Buffer.byteLength(built.prompt) > MAX_ISOLATED_PROMPT_BYTES) { @@ -643,26 +481,26 @@ class LocalWorkspaceAgentInvocation { }) } const knownSessionHistory = group.conversationType === 'direct' - || sessionProvenance.completeness !== 'unknown-legacy' + || session.provenance.completeness !== 'unknown-legacy' const sameTaskSession = !taskId - || taskId === sessionMeta.originTaskId - || (Array.isArray(sessionMeta.inheritedTaskIds) && sessionMeta.inheritedTaskIds.includes(taskId)) - let transcriptAfterKind = knownSessionHistory && !sessionRotated && storedSessionRef && storedSessionRef === sessionRef - && (!sessionMeta.lastDeliveredMessageId || !sameTaskSession) + || taskId === session.meta.originTaskId + || (Array.isArray(session.meta.inheritedTaskIds) && session.meta.inheritedTaskIds.includes(taskId)) + let transcriptAfterKind = knownSessionHistory && !session.rotated && session.storedRef && session.storedRef === session.ref + && (!session.meta.lastDeliveredMessageId || !sameTaskSession) ? kind : '' // Group Sessions already contain the immutable bootstrap instructions after // a successful turn. Keep direct chat unchanged and send only the compact // Harness context on later group turns. let promptMode = group.conversationType !== 'direct' - && Boolean(sessionRef) - && !sessionRotated + && Boolean(session.ref) + && !session.rotated && knownSessionHistory - && sessionMeta.turns > 0 + && session.meta.turns > 0 ? 'continuation' : 'bootstrap' const sessionContextOptions = (!isolated && !frozen && promptMode === 'continuation' && sameTaskSession) - ? { ...(context.contextOptions || {}), afterMessageId: sessionMeta.lastDeliveredMessageId || '' } + ? { ...(context.contextOptions || {}), afterMessageId: session.meta.lastDeliveredMessageId || '' } : (context.contextOptions || {}) let packedContext = isolated || frozen ? { @@ -700,7 +538,7 @@ class LocalWorkspaceAgentInvocation { let deliveryContext = { contextPackId: invocationContext?.contextPackId || activeRun?.contextPackId || '', deliveryRecordIds: [], - sessionProvenance, + sessionProvenance: session.provenance, } if (harnessRun) { const liveHarnessRun = harness.current(kind, round, harnessRun.agentRunId) @@ -708,7 +546,7 @@ class LocalWorkspaceAgentInvocation { liveHarnessRun.context = { ...deliveredContext.context, contextMode: promptMode, - sessionRotated, + sessionRotated: session.rotated, ...deliveryContext, } } @@ -728,11 +566,7 @@ class LocalWorkspaceAgentInvocation { let agentCallbacksClosed = false let parentAbortHandler = null let parentAbortPromise = null - let capturePromise = null - let artifactCapturePromise = null let runPromise = null - let importPromise = null - let outcomeCapturePromise = null let attemptProgress = [] const resolvedGateIds = [] const pendingPermissionCallbacks = new Set() @@ -766,6 +600,19 @@ class LocalWorkspaceAgentInvocation { agentCallbacksClosed = true watchdog.close() } + const outputs = new InvocationOutputs({ + enabled: allowWrite, + workdir: group.workdir, + signal: agentController.signal, + agentKind: kind, + startedAt, + deps: { + captureAgentOutputs: (...args) => this.captureAgentOutputs(...args), + captureArtifactOutputs: (...args) => this.captureArtifactOutputs(...args), + importAgentOutputs: (...args) => this.importAgentOutputs(...args), + captureAgentOutcomeDescriptors: (...args) => this.captureAgentOutcomeDescriptors(...args), + }, + }) const waitForHumanGate = async (operation) => { watchdog.pause() try { @@ -839,8 +686,6 @@ class LocalWorkspaceAgentInvocation { const rest = answerFilter.flush(finalText) if (rest) emitHarnessEvent({ type: 'answer_delta', status: 'running', delta: rest }) } - let outputBaseline = null - let artifactOutputBaseline = null let stagedInputs = null let result let operationStarted = false @@ -851,7 +696,7 @@ class LocalWorkspaceAgentInvocation { let connectorResume = null if (resumedConnectorGate) { const request = resumedConnectorGate.request - const binding = connectorSessionBinding(sessionRef, sessionProvenance) + const binding = session.binding() const requestHash = sha256(canonicalJson(request)) if (!agent.connectorInstanceId || request.connectorInstanceId !== agent.connectorInstanceId @@ -887,7 +732,7 @@ class LocalWorkspaceAgentInvocation { const finished = harness.finishAgent(kind, round, status, finalText, { ...deliveredContext.context, contextMode: promptMode, - sessionRotated, + sessionRotated: session.rotated, ...(context.phase ? { phase: context.phase } : {}), ...(Number.isInteger(context.executionSequence) ? { executionSequence: context.executionSequence } @@ -921,36 +766,7 @@ class LocalWorkspaceAgentInvocation { sideEffectsPossible: allowWrite, }) } - if (allowWrite) { - artifactCapturePromise = Promise.resolve().then(() => this.captureArtifactOutputs( - group.workdir, - { signal: agentController.signal }, - )) - artifactCapturePromise.catch(() => {}) - try { - capturePromise = Promise.resolve().then(() => this.captureAgentOutputs( - group.workdir, - { signal: agentController.signal }, - )) - capturePromise.catch(() => {}) - outputBaseline = await abortableOperation( - () => capturePromise, - agentController.signal, - ) - } catch (error) { - if (agentController.signal.aborted) throw error - /* output capture is best effort */ - } - try { - artifactOutputBaseline = await abortableOperation( - () => artifactCapturePromise, - agentController.signal, - ) - } catch (error) { - if (agentController.signal.aborted) throw error - /* artifact output capture is best effort */ - } - } + await outputs.begin() const nativeImageLimit = Math.max( 0, Math.floor(Number(this.attachmentSupport(kind)?.image) || 0), @@ -958,68 +774,28 @@ class LocalWorkspaceAgentInvocation { stagedInputs = isolated || frozen ? null : stageAgentInputs(group.workdir, context.attachmentSnapshots || [], nativeImageLimit) - const runtimeInstruction = cleanText(context.runtimeInstruction, 3000) - const regenerationInstruction = cleanText(context.regenerationInstruction, 3000) - const responseScopeInstruction = group.conversationType === 'direct' - ? '' - : [ - 'Final response scope:', - 'Answer only the current user task stated above.', - 'Use earlier messages only as supporting evidence when they help answer that current task.', - 'Do not separately mention, summarize, continue, or complete any previous user task.', - 'Do not append an answer to an older task after the current-task conclusion.', - ].join('\n') - const buildPrompt = (afterKind, contextPackage) => isolated || frozen - ? v4Prompt - : group.conversationType === 'direct' - ? [ - // A resumed native session already holds the conversation. A new - // or rebuilt session needs the earlier turns and, always, the - // current request plus the per-turn delivery contract. - directPromptFor({ - group, - label: this.agentLabel(kind), - packed: contextPackage, - skillHints: context.skillHints || [], - knowledgeBaseHints: context.knowledgeBaseHints || [], - includeHistory: sessionRotated || !sessionRef, - }), - stagedAgentInputPrompt(stagedInputs), - runtimeInstruction ? `Harness recovery task:\n${runtimeInstruction}` : '', - regenerationInstruction - ? `Fresh response instruction:\n${regenerationInstruction}` - : '', - ].filter(Boolean).join('\n\n') - : [ - context.v4 === true && v4Prompt - ? v4Prompt - : this.promptFor( - group, kind, context.completionPolicy === 'typed' ? 'manual' : mode, - threadRootId, context.skillHints || [], - context.knowledgeBaseHints || [], afterKind, contextPackage, promptMode, - activeRun?.unlimitedRounds === true, - ), - stagedAgentInputPrompt(stagedInputs), - runtimeInstruction ? `Harness recovery task:\n${runtimeInstruction}` : '', - regenerationInstruction - ? `Fresh response instruction:\n${regenerationInstruction}` - : '', - responseScopeInstruction, - ].filter(Boolean).join('\n') + const buildPrompt = (afterKind, contextPackage) => assembleInvocationPrompt({ + group, + kind, + mode, + threadRootId, + context, + activeRun, + isolated, + frozen, + v4Prompt, + promptMode, + afterKind, + packed: contextPackage, + session, + stagedInputs, + promptFor: (...args) => this.promptFor(...args), + agentLabel: label => this.agentLabel(label), + }) let prompt = buildPrompt(transcriptAfterKind, packedContext) const recordOutboundPayload = (outbound = {}) => { watchdog.progress() - if (activeRun?.budget) { - const bytes = outboundWirePayloadBytes(outbound) - activeRun.budget.addUsageBatch([ - { dimension: 'outboundBytes', amount: bytes.length, source: 'reported' }, - { - dimension: 'inputTokens', - amount: Math.ceil(bytes.length / 4), - source: 'estimated', - }, - ]) - } + recordOutboundUsage(activeRun?.budget, outbound) const baseContextPackId = isolated?.contextPackId || activeRun?.contextPackId || '' if (!baseContextPackId || !harness || !harnessRun) return const deliveredPrompt = typeof outbound?.prompt === 'string' ? outbound.prompt : prompt @@ -1049,7 +825,7 @@ class LocalWorkspaceAgentInvocation { permissionMode: allowWrite ? 'workspace-write' : 'read-only', skills: isolated || frozen ? [] : (context.skillHints || []), runtimeAdditions: attempt.runtimeAdditions, - sessionProvenance, + sessionProvenance: session.provenance, }) deliveryContext = { contextPackId: attempt.record.contextPackId, @@ -1057,7 +833,7 @@ class LocalWorkspaceAgentInvocation { ...deliveryContext.deliveryRecordIds, delivery.deliveryRecordId, ].slice(-8), - sessionProvenance, + sessionProvenance: session.provenance, ...contextSourceProof(attempt.record), promptChars: deliveredPrompt.length, promptBytes: Buffer.byteLength(deliveredPrompt, 'utf8'), @@ -1070,7 +846,7 @@ class LocalWorkspaceAgentInvocation { liveHarnessRun.context = { ...deliveredContext.context, contextMode: promptMode, - sessionRotated, + sessionRotated: session.rotated, ...deliveryContext, ...connectorContext, } @@ -1079,29 +855,10 @@ class LocalWorkspaceAgentInvocation { this.emitChanged() } const rebuildFreshSession = () => { - sessionMeta = {} - sessionProvenance = createdSessionProvenance(group, taskId) + session.discard() promptMode = 'bootstrap' - deliveryContext = { ...deliveryContext, sessionProvenance } - this.commitSessionState((nextState) => { - delete nextState.sessions[key] - delete nextState.sessionMeta[key] - if (kind === 'openclaw') { - const generation = randomUUID().replace(/[^a-zA-Z0-9]/g, '').slice(0, 12) || 'session' - sessionRef = this.openClawSessionRef( - group, generation, group.conversationType === 'direct' ? '' : taskId, - ) - nextState.sessions[key] = sessionRef - nextState.sessionMeta[key] = normalizeSessionMeta( - provenanceMeta(sessionProvenance), - ) - } else { - sessionRef = '' - } - }) - sessionTransport = '' + deliveryContext = { ...deliveryContext, sessionProvenance: session.provenance } transcriptAfterKind = '' - sessionRotated = true buildV4Prompt() packedContext = frozen ? { @@ -1131,7 +888,7 @@ class LocalWorkspaceAgentInvocation { liveHarnessRun.context = { ...deliveredContext.context, contextMode: promptMode, - sessionRotated, + sessionRotated: session.rotated, ...deliveryContext, ...connectorContext, } @@ -1142,7 +899,7 @@ class LocalWorkspaceAgentInvocation { return prompt } const runOptions = { - sessionRef, + sessionRef: session.ref, ...(Number.isInteger(context.executionSequence) ? { executionSequence: context.executionSequence } : {}), @@ -1150,16 +907,7 @@ class LocalWorkspaceAgentInvocation { if (agentCallbacksClosed || agentController.signal.aborted) return watchdog.progress() if (reviewOnly || isolated) return - const transport = ['legacy', 'acp'].includes(metadata?.transport) - ? metadata.transport - : '' - if (transport) sessionTransport = transport - sessionRef = String(nextSessionRef || '') - this.persistSessionState(key, sessionRef, normalizeSessionMeta({ - ...sessionMeta, - ...provenanceMeta(sessionProvenance), - ...(sessionTransport ? { transport: sessionTransport } : {}), - })) + session.adopt(nextSessionRef, metadata) }, onSessionInvalidated: () => { if (kind !== 'hermes' || reviewOnly || isolated || resumedPermission) return null @@ -1204,7 +952,7 @@ class LocalWorkspaceAgentInvocation { } } const requestHash = sha256(canonicalJson(request)) - const binding = connectorSessionBinding(sessionRef, sessionProvenance) + const binding = session.binding() const decision = await waitForHumanGate(() => this.requestHumanGate({ type: 'permission', runId: activeRun.runId, @@ -1237,7 +985,7 @@ class LocalWorkspaceAgentInvocation { ) return pending }, - sessionTransport, + sessionTransport: session.transport, ...(kind === 'workbuddy' && Number.isInteger(context.maxInternalTurns) ? { maxTurns: context.maxInternalTurns } : {}), @@ -1245,13 +993,15 @@ class LocalWorkspaceAgentInvocation { // ACP runtimes are shared by the stable group/task session key. OpenClaw // is excluded because its gateway wrapper owns the process lifetime and // is intentionally torn down after each invocation. - ...(kind !== 'openclaw' ? { acpPersistenceKey: key } : {}), + ...(kind !== 'openclaw' ? { acpPersistenceKey: session.key } : {}), ...(kind === 'hermes' ? { hermesAcpAvailable: HERMES_WORKSPACE_ACP_ENABLED && agent.acpAvailable !== false } : {}), } const runCurrentSession = () => { - const currentRunOptions = { ...runOptions, sessionRef, sessionTransport } + const currentRunOptions = { + ...runOptions, sessionRef: session.ref, sessionTransport: session.transport, + } if (!agent.connectorInstanceId) { return this.runAgent(agent, prompt, group.workdir, currentRunOptions) } @@ -1281,7 +1031,7 @@ class LocalWorkspaceAgentInvocation { liveHarnessRun.context = { ...deliveredContext.context, contextMode: promptMode, - sessionRotated, + sessionRotated: session.rotated, ...deliveryContext, ...connectorContext, } @@ -1326,7 +1076,7 @@ class LocalWorkspaceAgentInvocation { } if (agentController.signal.aborted) throw agentStoppedError() const executeCurrentSession = async () => { - const reusedSessionRef = sessionRef + const reusedSessionRef = session.ref try { operationStarted = true if (allowWrite) sideEffectsStarted = true @@ -1359,11 +1109,11 @@ class LocalWorkspaceAgentInvocation { } const waiting = result.waitingRequest const waitingSessionRef = String(result.sessionRef || '') - if (waitingSessionRef && waitingSessionRef !== sessionRef) { + if (waitingSessionRef && waitingSessionRef !== session.ref) { runOptions.onSessionRef(waitingSessionRef) - sessionRef = waitingSessionRef + session.ref = waitingSessionRef } - const binding = connectorSessionBinding(sessionRef, sessionProvenance) + const binding = session.binding() const request = { version: 1, source: 'connector', @@ -1467,81 +1217,19 @@ class LocalWorkspaceAgentInvocation { })) if (activeRun) activeRun.progress = toolCalls if (activeRun?.budget) { - const reportedOutputTokens = Number(result.usage?.outputTokens) - const hasReportedOutputTokens = Number.isSafeInteger(reportedOutputTokens) - && reportedOutputTokens >= 0 - const reportedCostMicros = Number(result.usage?.costMicros) - activeRun.budget.addUsageBatch([ - { - dimension: 'outputTokens', - amount: hasReportedOutputTokens - ? reportedOutputTokens - : Math.ceil(reply.text.length / 4), - source: hasReportedOutputTokens ? 'reported' : 'estimated', - }, - { dimension: 'toolCalls', amount: toolCalls.length, source: 'estimated' }, - ...(Number.isSafeInteger(reportedCostMicros) && reportedCostMicros >= 0 - ? [{ dimension: 'costMicros', amount: reportedCostMicros, source: 'reported' }] - : []), - { - dimension: 'elapsedMs', - value: activeRun.budget.elapsedValue(), - source: 'reported', - }, - ]) + recordReplyUsage(activeRun.budget, { + usage: result.usage, + replyText: reply.text, + toolCallCount: toolCalls.length, + }) this.checkpointRun(group.id, activeRun) } - let attachments = [] - if (allowWrite) { - try { - importPromise = Promise.resolve().then(() => this.importAgentOutputs({ - workdir: group.workdir, - baseline: outputBaseline, - startedAt, - agentKind: kind, - reportedPaths: extractReportedOutputPaths(reply.text), - signal: agentController.signal, - })) - importPromise.catch(() => {}) - const imported = await abortableOperation( - () => importPromise, - agentController.signal, - ) - attachments = (Array.isArray(imported) ? imported : []) - .slice(0, MAX_MESSAGE_ATTACHMENTS) - .map(normalizeAttachmentMetadata) - .filter(Boolean) - } catch (error) { - if (agentController.signal.aborted) throw error - /* the reply remains available when no valid media was produced */ - } - } + const attachments = await outputs.importMedia(reply.text) if (agentController.signal.aborted) throw agentStoppedError() - let descriptors = [] - if (allowWrite && artifactOutputBaseline) { - try { - outcomeCapturePromise = Promise.resolve().then(() => ( - this.captureAgentOutcomeDescriptors({ - workdir: group.workdir, - baseline: artifactOutputBaseline, - startedAt, - agentKind: kind, - runId: activeRun?.runId, - agentRunId: harnessRun?.agentRunId, - signal: agentController.signal, - }) - )) - outcomeCapturePromise.catch(() => {}) - const captured = await abortableOperation( - () => outcomeCapturePromise, - agentController.signal, - ) - descriptors = Array.isArray(captured) ? captured : [] - } catch (error) { - if (agentController.signal.aborted) throw error - /* the conclusion remains durable when explicit outputs cannot be captured */ - } - } + const descriptors = await outputs.describeOutcomes({ + runId: activeRun?.runId, + agentRunId: harnessRun?.agentRunId, + }) if (agentController.signal.aborted) throw agentStoppedError() const requestedOutcomeRefs = !internal && context.outcomeRefs && typeof context.outcomeRefs === 'object' @@ -1612,15 +1300,13 @@ class LocalWorkspaceAgentInvocation { ) : null if (!reviewOnly && !isolated) { - this.persistSessionState(key, result.sessionRef || sessionRef, completedSessionMeta( - sessionMeta, sessionProvenance, taskId, { + session.complete(result.sessionRef, { promptChars: prompt.length, replyChars: reply.text.length, - rotated: sessionRotated, - transport: sessionTransport, - lastDeliveredMessageId: packedContext.latestMessageId || sessionMeta.lastDeliveredMessageId, - }, - )) + rotated: session.rotated, + transport: session.transport, + lastDeliveredMessageId: packedContext.latestMessageId || session.meta.lastDeliveredMessageId, + }) } return { message, @@ -1632,71 +1318,31 @@ class LocalWorkspaceAgentInvocation { consensus: reply.consensus && result.outcome === 'completed', ...(v4Delivery ? { v4Delivery, - v4SessionBinding: connectorSessionBinding(result.sessionRef || sessionRef, sessionProvenance), + v4SessionBinding: connectorSessionBinding(result.sessionRef || session.ref, session.provenance), } : {}), } } catch (caughtError) { - const parentTimedOut = Boolean(signal?.aborted && activeRun?.stopReason === 'timeout') - const parentStopped = Boolean(signal?.aborted || parentAbortObserved) - const parentInterrupted = parentStopped && activeRun?.stopReason === 'shutdown' - const agentCancelled = agentController.signal.aborted - && !parentStopped && !watchdog.timedOut - if (parentStopped || watchdog.timedOut || agentCancelled) { - const cleanupPromises = [ - capturePromise, - artifactCapturePromise, - runPromise, - importPromise, - outcomeCapturePromise, - ].filter(Boolean) + const stop = resolveInvocationStop({ + caughtError, signal, parentAbortObserved, activeRun, agentController, watchdog, + }) + if (stop.settleOutstanding) { + const cleanupPromises = [runPromise, ...outputs.pending()].filter(Boolean) if (cleanupPromises.length) { await settleWithin(Promise.allSettled(cleanupPromises), this.runAbortGraceMs) } } - const error = parentTimedOut - ? new Error('LOCAL_AGENT_TIMEOUT') - : parentStopped - ? (caughtError?.message === 'LOCAL_AGENT_EXECUTION_STOPPED' - ? caughtError - : agentStoppedError()) - : watchdog.timedOut - ? (watchdog.error || new Error('LOCAL_AGENT_TIMEOUT')) - : agentCancelled - ? agentStoppedError() - : caughtError + const { error } = stop // Only an explicit sign-in failure changes readiness. Network, quota and // provider refusals stay run-scoped so a healthy Agent remains usable. if (error?.code === 'LOCAL_AGENT_AUTH_REQUIRED' && context.deferCredentialFailure !== true) { this.markRuntimeCredential(kind, 'missing') } - const status = parentTimedOut - ? 'timeout' - : parentInterrupted - ? 'interrupted' - : parentStopped - ? 'stopped' - : watchdog.timedOut - ? 'timeout' - : agentCancelled - ? 'stopped' - : 'failed' - const trace = finishHarness(status) - if (trace && error && (typeof error === 'object' || typeof error === 'function')) { - Object.defineProperty(error, 'runTrace', { - value: trace, - enumerable: false, - configurable: true, - }) - } // Failure and stop notices take the turn's slot in the timeline. - if (context.turnSeq && error && (typeof error === 'object' || typeof error === 'function')) { - Object.defineProperty(error, 'turnSeq', { - value: context.turnSeq, - enumerable: false, - configurable: true, - }) - } + attachErrorMetadata(error, { + runTrace: finishHarness(stop.status), + turnSeq: context.turnSeq, + }) throw attachInvocationFailure(error, { outcomeCertainty: terminalFailureKnown ? 'known_failed' @@ -1748,7 +1394,7 @@ class LocalWorkspaceAgentInvocation { } if (kind === 'openclaw' && rotateOpenClaw) { const generation = randomUUID().replace(/[^a-zA-Z0-9]/g, '').slice(0, 12) || 'session' - const provenance = createdSessionProvenance(group, cleanText(taskId, 120)) + const provenance = createdSessionProvenance(cleanText(taskId, 120)) state.sessions[key] = this.openClawSessionRef(group, generation) state.sessionMeta[key] = normalizeSessionMeta(provenanceMeta(provenance)) changed = true diff --git a/desktop/test/workspace/invocation-helpers.test.cjs b/desktop/test/workspace/invocation-helpers.test.cjs new file mode 100644 index 0000000..a5c9940 --- /dev/null +++ b/desktop/test/workspace/invocation-helpers.test.cjs @@ -0,0 +1,212 @@ +const test = require('node:test') +const assert = require('node:assert/strict') +const { + attachErrorMetadata, + attachInvocationFailure, + resolveInvocationStop, +} = require('../../src/workspace/invocation-failure.cjs') +const { + recordOutboundUsage, + recordReplyUsage, +} = require('../../src/workspace/invocation-budget.cjs') +const { + InvocationOutputs, + extractReportedOutputPaths, +} = require('../../src/workspace/invocation-outputs.cjs') +const { assembleInvocationPrompt } = require('../../src/workspace/invocation-prompt.cjs') +const { createLegacyOutboundPayload } = require('../../src/collaboration/outbound-payload.cjs') + +function stopInputs({ parentAborted = false, stopReason = '', agentAborted = false, timedOut = false } = {}) { + const signal = { aborted: parentAborted } + return { + caughtError: new Error('something broke'), + signal, + parentAbortObserved: false, + activeRun: { stopReason }, + agentController: { signal: { aborted: agentAborted || parentAborted || timedOut } }, + watchdog: { timedOut, error: timedOut ? new Error('LOCAL_AGENT_TIMEOUT') : null }, + } +} + +test('an ordinary failure keeps its own error and status', () => { + const inputs = stopInputs() + const stop = resolveInvocationStop(inputs) + assert.equal(stop.error, inputs.caughtError) + assert.equal(stop.status, 'failed') + assert.equal(stop.settleOutstanding, false) +}) + +test('a stopped run, a timed-out run and a shutdown are told apart', () => { + const stopped = resolveInvocationStop(stopInputs({ parentAborted: true })) + assert.equal(stopped.status, 'stopped') + assert.equal(stopped.error.message, 'LOCAL_AGENT_EXECUTION_STOPPED') + assert.equal(stopped.settleOutstanding, true) + + const timedOutRun = resolveInvocationStop(stopInputs({ parentAborted: true, stopReason: 'timeout' })) + assert.equal(timedOutRun.status, 'timeout') + assert.equal(timedOutRun.error.message, 'LOCAL_AGENT_TIMEOUT') + + const shutdown = resolveInvocationStop(stopInputs({ parentAborted: true, stopReason: 'shutdown' })) + assert.equal(shutdown.status, 'interrupted') +}) + +test('the watchdog timeout and a self-cancelled Agent are reported distinctly', () => { + const watchdog = resolveInvocationStop(stopInputs({ timedOut: true })) + assert.equal(watchdog.status, 'timeout') + assert.equal(watchdog.error.message, 'LOCAL_AGENT_TIMEOUT') + assert.equal(watchdog.settleOutstanding, true) + + const cancelled = resolveInvocationStop(stopInputs({ agentAborted: true })) + assert.equal(cancelled.status, 'stopped') + assert.equal(cancelled.settleOutstanding, true) +}) + +test('error metadata is attached as hidden properties and empty values are skipped', () => { + const error = new Error('x') + attachErrorMetadata(error, { runTrace: { id: 1 }, turnSeq: 0 }) + assert.deepEqual(error.runTrace, { id: 1 }) + assert.equal('turnSeq' in error, false) + assert.equal(Object.keys(error).includes('runTrace'), false) + + attachInvocationFailure(error, { + outcomeCertainty: 'unknown_outcome', sideEffectsPossible: true, operationId: 'op', idempotencyMode: 'none', + }) + attachInvocationFailure(error, { + outcomeCertainty: 'not_started', sideEffectsPossible: false, operationId: 'other', idempotencyMode: 'durable', + }) + assert.equal(error.invocationFailure.outcomeCertainty, 'unknown_outcome') + assert.equal(error.invocationFailure.operationId, 'op') + assert.equal(attachInvocationFailure('not an object', {}), 'not an object') +}) + +test('usage is recorded from what the CLI reported, falling back to estimates', () => { + const batches = [] + const budget = { addUsageBatch: batch => batches.push(batch), elapsedValue: () => 42 } + recordReplyUsage(budget, { usage: { outputTokens: 7, costMicros: 99 }, replyText: 'abcdefgh', toolCallCount: 3 }) + assert.deepEqual(batches[0].map(item => [item.dimension, item.source]), [ + ['outputTokens', 'reported'], ['toolCalls', 'estimated'], ['costMicros', 'reported'], ['elapsedMs', 'reported'], + ]) + assert.equal(batches[0][0].amount, 7) + + recordReplyUsage(budget, { usage: {}, replyText: 'abcdefgh', toolCallCount: 0 }) + assert.deepEqual(batches[1].map(item => item.dimension), ['outputTokens', 'toolCalls', 'elapsedMs']) + assert.equal(batches[1][0].amount, 2) + assert.equal(batches[1][0].source, 'estimated') + + recordOutboundUsage(budget, createLegacyOutboundPayload({ + prompt: 'x'.repeat(40), command: 'codex', args: ['exec'], cwd: '/w', stdin: '', promptMode: 'argv', + })) + assert.deepEqual(batches[2].map(item => item.dimension), ['outboundBytes', 'inputTokens']) + recordOutboundUsage(budget, { prompt: 'built elsewhere' }) + assert.equal(batches.length, 3) + recordOutboundUsage(null, {}) + recordReplyUsage(undefined, { replyText: '' }) +}) + +test('reported output paths keep spaces inside quotes and links but not trailing punctuation', () => { + const text = [ + 'Saved to `/Users/me/My Reports/final draft.md`, and also /tmp/plain.txt.', + 'See [the log]() and "/Volumes/Data/a b.png".', + ].join('\n') + assert.deepEqual(extractReportedOutputPaths(text).sort(), [ + '/Users/me/My Reports/final draft.md', + '/Volumes/Data/a b.png', + '/home/dev/build logs/out.log', + '/tmp/plain.txt', + ]) + assert.deepEqual(extractReportedOutputPaths(undefined), []) +}) + +test('output capture only runs for writable turns and shrugs off capture errors', async () => { + const calls = [] + const controller = new AbortController() + const deps = { + captureAgentOutputs: async () => { calls.push('baseline'); return { files: [] } }, + captureArtifactOutputs: async () => { calls.push('artifacts'); throw new Error('disk gone') }, + importAgentOutputs: async (input) => { + calls.push(['import', input.reportedPaths]) + return [{ id: 'a1', name: 'x.png', mimeType: 'image/png', size: 10 }, { bogus: true }] + }, + captureAgentOutcomeDescriptors: async () => { calls.push('describe'); return [{ kind: 'file' }] }, + } + const readOnly = new InvocationOutputs({ + enabled: false, workdir: '/w', signal: controller.signal, agentKind: 'codex', startedAt: 1, deps, + }) + await readOnly.begin() + assert.deepEqual(await readOnly.importMedia('/tmp/a.txt'), []) + assert.deepEqual(await readOnly.describeOutcomes({}), []) + assert.deepEqual(calls, []) + + const writable = new InvocationOutputs({ + enabled: true, workdir: '/w', signal: controller.signal, agentKind: 'codex', startedAt: 1, deps, + }) + await writable.begin() + assert.deepEqual(writable.baseline, { files: [] }) + assert.equal(writable.artifactBaseline, null) + const attachments = await writable.importMedia('Wrote /tmp/a.txt') + assert.deepEqual(attachments.map(item => item.id), ['a1']) + assert.deepEqual(await writable.describeOutcomes({ runId: 'r', agentRunId: 'a' }), []) + assert.equal(writable.pending().length, 3) +}) + +test('a stop request interrupts output capture instead of being swallowed', async () => { + const controller = new AbortController() + const outputs = new InvocationOutputs({ + enabled: true, workdir: '/w', signal: controller.signal, agentKind: 'codex', startedAt: 1, + deps: { + captureAgentOutputs: () => new Promise(() => {}), + captureArtifactOutputs: () => new Promise(() => {}), + importAgentOutputs: async () => [], + captureAgentOutcomeDescriptors: async () => [], + }, + }) + const pending = outputs.begin() + controller.abort() + await assert.rejects(pending) +}) + +function promptInputs(overrides = {}) { + return { + group: { id: 'g', conversationType: 'group' }, + kind: 'codex', + mode: 'manual', + threadRootId: 'root', + context: {}, + activeRun: null, + isolated: false, + frozen: false, + v4Prompt: '', + promptMode: 'bootstrap', + afterKind: '', + packed: { currentTaskText: 'Do the thing', recentText: '', stableText: '' }, + session: { rotated: false, ref: 'native-1' }, + stagedInputs: null, + promptFor: (...args) => `GROUP PROMPT mode=${args[2]} continuation=${args[8]}`, + agentLabel: kind => `Label ${kind}`, + ...overrides, + } +} + +test('frozen and isolated invocations send exactly the approved prompt', () => { + assert.equal(assembleInvocationPrompt(promptInputs({ frozen: true, v4Prompt: 'APPROVED' })), 'APPROVED') + assert.equal(assembleInvocationPrompt(promptInputs({ isolated: true, v4Prompt: 'ISOLATED' })), 'ISOLATED') +}) + +test('group prompts append recovery text and the response scope, in that order', () => { + const prompt = assembleInvocationPrompt(promptInputs({ + context: { runtimeInstruction: 'Retry carefully', regenerationInstruction: 'Try another angle' }, + promptMode: 'continuation', + })) + const lines = prompt.split('\n') + assert.equal(lines[0], 'GROUP PROMPT mode=manual continuation=continuation') + assert.ok(prompt.indexOf('Harness recovery task:\nRetry carefully') + < prompt.indexOf('Fresh response instruction:\nTry another angle')) + assert.ok(prompt.indexOf('Fresh response instruction') < prompt.indexOf('Final response scope:')) +}) + +test('a V4 prompt replaces the workspace prompt, and a typed policy runs as a manual turn', () => { + const v4 = assembleInvocationPrompt(promptInputs({ context: { v4: true }, v4Prompt: 'V4 PROMPT' })) + assert.equal(v4.split('\n')[0], 'V4 PROMPT') + const typed = assembleInvocationPrompt(promptInputs({ mode: 'auto', context: { completionPolicy: 'typed' } })) + assert.match(typed, /mode=manual/) +}) diff --git a/desktop/test/workspace/invocation-session.test.cjs b/desktop/test/workspace/invocation-session.test.cjs new file mode 100644 index 0000000..faa2f28 --- /dev/null +++ b/desktop/test/workspace/invocation-session.test.cjs @@ -0,0 +1,95 @@ +const test = require('node:test') +const assert = require('node:assert/strict') +const { + InvocationSession, + connectorSessionBinding, + createdSessionProvenance, +} = require('../../src/workspace/invocation-session.cjs') + +function sessionFor({ kind = 'codex', conversationType = 'group', stored = {} } = {}) { + const state = { + sessions: { 'group-1:codex': 'native-1', ...stored.sessions }, + sessionMeta: { 'group-1:codex': { turns: 2, transport: 'acp' }, ...stored.sessionMeta }, + } + const persisted = [] + const session = new InvocationSession({ + resolved: { + key: 'group-1:codex', + sessionRef: state.sessions['group-1:codex'], + sessionMeta: state.sessionMeta['group-1:codex'], + provenance: createdSessionProvenance('task-1'), + }, + group: { id: 'group-1', conversationType }, + kind, + taskId: 'task-9', + deps: { + commitSessionState: mutator => mutator(state), + openClawSessionRef: (group, generation, taskId) => `openclaw:${group.id}:${generation.length}:${taskId}`, + persistSessionState: (...args) => persisted.push(args), + }, + }) + return { session, state, persisted } +} + +test('a stored native session is resumed with its transport', () => { + const { session } = sessionFor() + assert.equal(session.ref, 'native-1') + assert.equal(session.storedRef, 'native-1') + assert.equal(session.transport, 'acp') + assert.equal(session.rotated, false) + assert.equal(session.meta.turns, 2) +}) + +test('discarding forgets the stored session and marks the turn as rebuilt', () => { + const { session, state } = sessionFor() + session.discard() + assert.equal(session.ref, '') + assert.equal(session.transport, '') + assert.equal(session.rotated, true) + assert.deepEqual(session.meta, {}) + assert.equal(session.provenance.originTaskId, 'task-9') + assert.equal('group-1:codex' in state.sessions, false) + assert.equal('group-1:codex' in state.sessionMeta, false) +}) + +test('OpenClaw gets a fresh named session straight away', () => { + const group = sessionFor({ kind: 'openclaw' }) + group.session.discard() + assert.match(group.session.ref, /^openclaw:group-1:12:task-9$/) + assert.equal(group.state.sessions['group-1:codex'], group.session.ref) + + const direct = sessionFor({ kind: 'openclaw', conversationType: 'direct' }) + direct.session.discard() + assert.match(direct.session.ref, /^openclaw:group-1:12:$/) +}) + +test('adopting a session id persists it with the provenance and a known transport only', () => { + const { session, persisted } = sessionFor() + session.adopt('native-2', { transport: 'legacy' }) + assert.equal(session.ref, 'native-2') + assert.equal(session.transport, 'legacy') + assert.equal(persisted.at(-1)[0], 'group-1:codex') + assert.equal(persisted.at(-1)[1], 'native-2') + assert.equal(persisted.at(-1)[2].transport, 'legacy') + + session.adopt('native-3', { transport: 'carrier-pigeon' }) + assert.equal(session.transport, 'legacy') +}) + +test('completing a turn stores the id the Agent returned, or keeps the current one', () => { + const { session, persisted } = sessionFor() + session.complete('native-9', { promptChars: 10, replyChars: 5, rotated: false, transport: 'acp' }) + assert.equal(persisted.at(-1)[1], 'native-9') + assert.equal(persisted.at(-1)[2].turns, 3) + session.complete('', { promptChars: 1, replyChars: 1 }) + assert.equal(persisted.at(-1)[1], 'native-1') +}) + +test('the binding changes with the native session so a decision cannot move to another one', () => { + const { session } = sessionFor() + const before = session.binding() + assert.deepEqual(before, connectorSessionBinding('native-1', session.provenance)) + session.ref = 'native-2' + assert.notEqual(session.binding().sessionRefHash, before.sessionRefHash) + assert.equal(session.binding().sessionProvenanceHash, before.sessionProvenanceHash) +})