diff --git a/agent-run-deliverable-check.mjs b/agent-run-deliverable-check.mjs index eeba16c..1641368 100644 --- a/agent-run-deliverable-check.mjs +++ b/agent-run-deliverable-check.mjs @@ -1,6 +1,7 @@ export const RECOVERABLE_FINISH_ERROR_CODES = new Set([ 'AGENT_RUN_TIMEOUT', 'SESSION_REPLY_TIMEOUT', + 'SESSION_REPLY_IDLE_TIMEOUT', 'SESSION_REPLY_INCOMPLETE', 'AGENT_RUN_STALE_RECOVERY', ]); diff --git a/agent-run-deliverable-check.test.mjs b/agent-run-deliverable-check.test.mjs index b335e63..42038a5 100644 --- a/agent-run-deliverable-check.test.mjs +++ b/agent-run-deliverable-check.test.mjs @@ -14,6 +14,7 @@ import { test('isRecoverableFinishError recognizes finish and run timeout codes', () => { assert.equal(isRecoverableFinishError({ code: 'SESSION_REPLY_INCOMPLETE' }), true); assert.equal(isRecoverableFinishError({ code: 'SESSION_REPLY_TIMEOUT' }), true); + assert.equal(isRecoverableFinishError({ code: 'SESSION_REPLY_IDLE_TIMEOUT' }), true); assert.equal(isRecoverableFinishError({ code: 'AGENT_RUN_TIMEOUT' }), true); assert.equal(isRecoverableFinishError({ code: 'TOOL_GATEWAY_VALIDATION_FAILED' }), false); }); diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index b7f18fe..8f9c1bf 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -2566,6 +2566,7 @@ export function createAgentRunGateway({ ); await appendEvent(runId, 'session_finished', { sessionId, + finishReason: finish.finishReason ?? finish.finishEvent?.reason ?? null, tokenState: finish.tokenState ?? null, toolCalls: finish.toolEvidence?.calls ?? [], generateImage: finish.toolEvidence?.generateImage ?? null, diff --git a/session-reply-wait.mjs b/session-reply-wait.mjs index a3c843b..937c06c 100644 --- a/session-reply-wait.mjs +++ b/session-reply-wait.mjs @@ -102,6 +102,7 @@ function parseGenerateImageResult(toolResult) { function createToolEvidenceCollector() { const requestNames = new Map(); + const pendingToolIds = new Set(); const calls = new Set(); const successfulCalls = new Set(); let generateImage = { called: false, succeeded: false }; @@ -110,6 +111,7 @@ function createToolEvidenceCollector() { observe(event) { for (const item of messageContentItems(event)) { if (item?.type === 'toolRequest') { + if (item.id) pendingToolIds.add(String(item.id)); const name = String(item?.toolCall?.value?.name ?? '').trim(); if (!name) continue; calls.add(name); @@ -118,6 +120,7 @@ function createToolEvidenceCollector() { continue; } if (item?.type !== 'toolResponse') continue; + pendingToolIds.delete(String(item.id ?? '')); const name = requestNames.get(String(item.id ?? '')) ?? ''; const result = item.toolResult ?? {}; const succeeded = result.status === 'success' && result?.value?.isError !== true; @@ -129,6 +132,9 @@ function createToolEvidenceCollector() { : { called: true, succeeded: false }; } }, + hasPendingTools() { + return pendingToolIds.size > 0; + }, snapshot() { return { calls: [...calls], @@ -139,6 +145,13 @@ function createToolEvidenceCollector() { }; } +export function resolveSessionReplyIdleTimeoutMs(env = process.env) { + const value = Number(env.MEMIND_SESSION_REPLY_IDLE_TIMEOUT_MS ?? 0); + return Number.isFinite(value) && value > 0 ? Math.max(30_000, Math.floor(value)) : 0; +} + +const IDLE_RESETTING_EVENT_TYPES = new Set(['Message', 'Notification', 'UpdateConversation']); + export async function consumeSessionEventsUntilFinish( body, { @@ -147,6 +160,7 @@ export async function consumeSessionEventsUntilFinish( onEvent = null, requireRequestActivityBeforeUnscopedFinish = false, emptyFinishGraceMs = 20_000, + idleTimeoutMs = 0, } = {}, ) { if (!body) { @@ -181,7 +195,22 @@ export async function consumeSessionEventsUntilFinish( }); }; - armOriginalDeadline(); + // Idle only counts while waiting on the model; tool calls (aider/openhands/image) + // can legitimately run silent for minutes and stay bounded by the overall deadline. + const idleMs = Math.max(0, Number(idleTimeoutMs) || 0); + const armActivityTimeout = () => { + const remaining = Math.max(1, deadline - Date.now()); + if (idleMs > 0 && idleMs < remaining && !toolEvidence.hasPendingTools()) { + armTimeout(idleMs, { + code: 'SESSION_REPLY_IDLE_TIMEOUT', + message: `session reply produced no events for ${idleMs}ms while waiting on the model`, + }); + return; + } + armOriginalDeadline(); + }; + + armActivityTimeout(); try { for await (const chunk of reader) { @@ -214,8 +243,8 @@ export async function consumeSessionEventsUntilFinish( err.retryable = false; throw err; } - if (event.type === 'Message') { - armOriginalDeadline(); + if (IDLE_RESETTING_EVENT_TYPES.has(event.type)) { + armActivityTimeout(); } if (event.type !== 'Finish') continue; @@ -252,6 +281,7 @@ export async function consumeSessionEventsUntilFinish( if (pendingProviderError) throw pendingProviderError; return { finishEvent: event, + finishReason: String(event.reason ?? '').trim() || 'stop', tokenState, toolEvidence: toolEvidence.snapshot(), }; diff --git a/session-reply-wait.test.mjs b/session-reply-wait.test.mjs index 47093be..e5336b0 100644 --- a/session-reply-wait.test.mjs +++ b/session-reply-wait.test.mjs @@ -6,8 +6,90 @@ import { consumeSessionEventsUntilFinish, eventMatchesRequest, parseSessionStreamEvent, + resolveSessionReplyIdleTimeoutMs, } from './session-reply-wait.mjs'; +function timedStream(steps) { + return new ReadableStream({ + start(controller) { + let at = 0; + for (const step of steps) { + at += step.afterMs ?? 0; + setTimeout(() => { + if (step.close) controller.close(); + else controller.enqueue(new TextEncoder().encode(step.frame)); + }, at); + } + }, + }); +} + +const toolRequestFrame = (id) => + `data: ${JSON.stringify({ + type: 'Message', + request_id: 'req-1', + message: { role: 'assistant', content: [{ type: 'toolRequest', id, toolCall: { status: 'success', value: { name: 'sandbox-fs__write_file' } } }] }, + })}\n\n`; +const toolResponseFrame = (id) => + `data: ${JSON.stringify({ + type: 'Message', + request_id: 'req-1', + message: { role: 'user', content: [{ type: 'toolResponse', id, toolResult: { status: 'success', value: { content: [] } } }] }, + })}\n\n`; + +test('resolveSessionReplyIdleTimeoutMs is off by default and floors small values', () => { + assert.equal(resolveSessionReplyIdleTimeoutMs({}), 0); + assert.equal(resolveSessionReplyIdleTimeoutMs({ MEMIND_SESSION_REPLY_IDLE_TIMEOUT_MS: 'abc' }), 0); + assert.equal(resolveSessionReplyIdleTimeoutMs({ MEMIND_SESSION_REPLY_IDLE_TIMEOUT_MS: '0' }), 0); + assert.equal(resolveSessionReplyIdleTimeoutMs({ MEMIND_SESSION_REPLY_IDLE_TIMEOUT_MS: '5' }), 30_000); + assert.equal(resolveSessionReplyIdleTimeoutMs({ MEMIND_SESSION_REPLY_IDLE_TIMEOUT_MS: '300000' }), 300_000); +}); + +test('consumeSessionEventsUntilFinish fails fast when the model goes silent', async () => { + const stream = timedStream([ + { frame: 'data: {"type":"Message","request_id":"req-1","message":{"role":"assistant"}}\n\n' }, + ]); + const started = Date.now(); + await assert.rejects( + consumeSessionEventsUntilFinish(stream, { requestId: 'req-1', timeoutMs: 5000, idleTimeoutMs: 60 }), + (error) => error.code === 'SESSION_REPLY_IDLE_TIMEOUT', + ); + assert.ok(Date.now() - started < 2000); +}); + +test('consumeSessionEventsUntilFinish does not idle-timeout while a tool call is running', async () => { + const stream = timedStream([ + { frame: toolRequestFrame('t1') }, + { afterMs: 200, frame: toolResponseFrame('t1') }, + { afterMs: 10, frame: 'data: {"type":"Finish","request_id":"req-1","reason":"stop","token_state":{"totalTokens":3}}\n\n' }, + ]); + const result = await consumeSessionEventsUntilFinish(stream, { + requestId: 'req-1', + timeoutMs: 5000, + idleTimeoutMs: 60, + }); + assert.equal(result.finishReason, 'stop'); + assert.deepEqual(result.toolEvidence.calls, ['sandbox-fs__write_file']); +}); + +test('consumeSessionEventsUntilFinish exposes the Goose finish reason', async () => { + const stream = timedStream([ + { frame: 'data: {"type":"Message","request_id":"req-1","message":{"role":"assistant"}}\n\n' }, + { frame: 'data: {"type":"Finish","request_id":"req-1","reason":"max_turns","token_state":{"totalTokens":9}}\n\n' }, + ]); + const result = await consumeSessionEventsUntilFinish(stream, { requestId: 'req-1', timeoutMs: 5000 }); + assert.equal(result.finishReason, 'max_turns'); +}); + +test('consumeSessionEventsUntilFinish defaults finishReason to stop for older goosed', async () => { + const stream = timedStream([ + { frame: 'data: {"type":"Message","request_id":"req-1","message":{"role":"assistant"}}\n\n' }, + { frame: 'data: {"type":"Finish","request_id":"req-1","token_state":{"totalTokens":9}}\n\n' }, + ]); + const result = await consumeSessionEventsUntilFinish(stream, { requestId: 'req-1', timeoutMs: 5000 }); + assert.equal(result.finishReason, 'stop'); +}); + test('parseSessionStreamEvent parses SSE data payload', () => { const event = parseSessionStreamEvent('id: 1\ndata: {"type":"Finish","token_state":null}\n'); assert.equal(event?.type, 'Finish'); diff --git a/tkmind-proxy.mjs b/tkmind-proxy.mjs index 7804c18..00f8ae6 100644 --- a/tkmind-proxy.mjs +++ b/tkmind-proxy.mjs @@ -20,7 +20,10 @@ import { } from './user-publish.mjs'; import { buildCurrentTimeAgentPrefix, buildTaskRoutingAgentText } from './user-memory-profile.mjs'; import { reconcileAgentSession } from './session-reconcile.mjs'; -import { consumeSessionEventsUntilFinish } from './session-reply-wait.mjs'; +import { + consumeSessionEventsUntilFinish, + resolveSessionReplyIdleTimeoutMs, +} from './session-reply-wait.mjs'; import { createSessionAccess, isSessionBrokerEnabled } from './session-broker.mjs'; import { createSessionBrokerMetrics, isSessionBrokerMetricsEnabled } from './session-broker-metrics.mjs'; import { @@ -2294,6 +2297,7 @@ export function createTkmindProxy({ // agent run complete. requireRequestActivityBeforeUnscopedFinish: true, emptyFinishGraceMs: 1_500, + idleTimeoutMs: resolveSessionReplyIdleTimeoutMs(), }); // Prevent unhandled rejection from killing the Portal process when reply // setup fails before we await Finish (e.g. provider Error frames arrive later). @@ -2342,6 +2346,7 @@ export function createTkmindProxy({ requestId, timeoutMs: Math.max(1, timeoutMs - 1_500), requireRequestActivityBeforeUnscopedFinish: true, + idleTimeoutMs: resolveSessionReplyIdleTimeoutMs(), }, ); finishResult = { ok: true, value: retriedFinish };