mirror of
https://github.com/Ryder-Sun/Meldwork.git
synced 2026-11-07 06:56:03 -10:00
refactor: split the invocation's session, outputs, budget, prompt and failure handling into modules
The 1500-line invokeLeased still mixed session bookkeeping, workspace output capture, budget accounting, prompt assembly and failure classification with the run lifecycle. Each is now a small module with unit tests: - InvocationSession: one place that resumes, adopts, discards (rebuilds) and completes the native session; the three copies of the 'start a fresh session' dance, one of them dead code, are gone - InvocationOutputs: workspace baseline, media import and outcome descriptors - invocation-budget: usage recording; a payload without a recorded wire size no longer fails the turn - invocation-prompt: the prompt one Agent receives (frozen, direct, group) - invocation-failure: why an invocation ended (stopped, timed out, interrupted, cancelled, failed) and the hidden error metadata - permission helpers for resumed decisions moved next to the permission policy invokeLeased is about 1000 lines shorter and reads as the run lifecycle. Behaviour is unchanged; focused invocation tests and the new module tests pass.
This commit is contained in:
Ryder Sun committed
1 parent
68145eb319
commit
3ce566f152
9 files changed
+949
-507
No files matched your search
@@ -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,
|
||||
}
|
||||
@@ -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 }
|
||||
@@ -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 }
|
||||
@@ -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 }
|
||||
@@ -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 }
|
||||
@@ -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,
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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](</home/dev/build logs/out.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/)
|
||||
})
|
||||
@@ -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)
|
||||
})
|
||||
Reference in new issue
Block a user