tkmind/memind
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

fix(agent): recover stale runs and improve new-user OA delivery

Browse Source 
Fix DEV logout cookie clearing, materialize selected MindSpace OA assets before agent runs, and recover zombie runs from synced workspace pages. Add client run wait timeout, harness retry limits, page-edit asset forwarding, and logout/john2 scenario tests.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
john
2026-07-10 20:05:52 +08:00
parent aede1e6fcb
commit 7f8d692d16
34 changed files with 1318 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent-run-gateway.mjs
+107 -20
View File
@@ -11,7 +11,10 @@ import {
persistSessionTranscriptMessages,
} from './conversation-transcript-persist.mjs';
import { ensureGooseUserMessageMetadata } from './goose-message.mjs';
import { tryRecoverRunFromSessionDeliverables } from './agent-run-deliverable-check.mjs';
import {
SESSION_FINISHED_STALE_GRACE_MS,
tryRecoverRunFromDeliverables,
} from './agent-run-deliverable-check.mjs';
const DEFAULT_RUN_RETRY_DELAYS_MS = [1500, 5000, 15000];
const TERMINAL_STATUSES = new Set(['succeeded', 'failed']);
@@ -391,12 +394,38 @@ export function createAgentRunGateway({
return projectRun(await getRunById(runId));
}
async function recoverRunFromDeliverables({ runId, userId, sessionId, err }) {
const recovered = await tryRecoverRunFromSessionDeliverables({
async function prepareRunDeliverables({ userId, sessionId, runId, runStartedAtMs = null }) {
if (typeof syncUserPagesOnSuccess !== 'function') return;
await syncUserPagesOnSuccess({
userId,
sessionId,
runId,
runStartedAtMs,
}).catch((err) => {
console.warn(
'[AgentRun] prepare deliverables sync failed:',
err instanceof Error ? err.message : err,
);
});
}
async function recoverRunFromDeliverables({
runId,
userId,
sessionId,
err,
runStartedAtMs = null,
requireRecoverableError = true,
}) {
const recovered = await tryRecoverRunFromDeliverables({
pool,
userId,
sessionId,
error: err,
requireRecoverableError,
runStartedAtMs,
prepareDeliverables: ({ userId: uid, sessionId: sid }) =>
prepareRunDeliverables({ userId: uid, sessionId: sid, runId, runStartedAtMs }),
});
if (!recovered) return false;
await appendEvent(runId, 'run_recovered_from_deliverables', {
@@ -780,6 +809,7 @@ export function createAgentRunGateway({
userId: row.user_id,
sessionId: recoverySessionId,
err,
runStartedAtMs: latest?.started_at ?? row.started_at ?? null,
})) {
await finalizeSuccessfulRun(runId, row, recoverySessionId);
return;
@@ -913,18 +943,27 @@ export function createAgentRunGateway({
limit = maxConcurrentRuns,
dryRun = true,
reason = 'stale_running_timeout',
sessionFinishedGraceMs = SESSION_FINISHED_STALE_GRACE_MS,
} = {}) {
const normalizedStaleMs = positiveInteger(staleMs, runTimeoutMs);
const normalizedLimit = positiveInteger(limit, maxConcurrentRuns);
const cutoff = nowMs() - normalizedStaleMs;
const normalizedSessionFinishedGraceMs = positiveInteger(
sessionFinishedGraceMs,
SESSION_FINISHED_STALE_GRACE_MS,
);
const startedCutoff = nowMs() - normalizedStaleMs;
const sessionFinishedCutoff = nowMs() - normalizedSessionFinishedGraceMs;
const [rows] = await pool.query(
`SELECT
r.id,
r.user_id,
r.agent_session_id,
r.request_id,
r.started_at,
r.updated_at,
r.attempts,
h.latest_heartbeat_at
h.latest_heartbeat_at,
sf.session_finished_at
FROM h5_agent_runs r
LEFT JOIN (
SELECT run_id, MAX(created_at) AS latest_heartbeat_at
@@ -932,18 +971,31 @@ export function createAgentRunGateway({
WHERE event_type = 'worker_heartbeat'
GROUP BY run_id
) h ON h.run_id = r.id
LEFT JOIN (
SELECT run_id, MAX(created_at) AS session_finished_at
FROM h5_agent_run_events
WHERE event_type = 'session_finished'
GROUP BY run_id
) sf ON sf.run_id = r.id
WHERE r.status = 'running'
AND r.started_at IS NOT NULL
AND COALESCE(h.latest_heartbeat_at, r.started_at) <= ?
ORDER BY COALESCE(h.latest_heartbeat_at, r.started_at) ASC
AND (
r.started_at <= ?
OR (
sf.session_finished_at IS NOT NULL
AND sf.session_finished_at <= ?
)
)
ORDER BY COALESCE(sf.session_finished_at, r.started_at) ASC
LIMIT ?`,
[cutoff, normalizedLimit],
[startedCutoff, sessionFinishedCutoff, normalizedLimit],
);
const recovered = [];
for (const row of rows) {
const ageMs = Math.max(0, nowMs() - Number(row.started_at ?? 0));
const heartbeatAt = row.latest_heartbeat_at == null ? null : Number(row.latest_heartbeat_at);
const heartbeatAgeMs = Math.max(0, nowMs() - Number(row.latest_heartbeat_at ?? row.started_at ?? 0));
const sessionFinishedAt = row.session_finished_at == null ? null : Number(row.session_finished_at);
const item = {
id: row.id,
requestId: row.request_id,
@@ -952,13 +1004,40 @@ export function createAgentRunGateway({
attempts: Number(row.attempts ?? 0),
heartbeatAt,
heartbeatAgeMs,
sessionFinishedAt,
ageMs,
reason,
};
if (!dryRun) {
const message = heartbeatAt == null
? `agent run recovered from stale running state after ${ageMs}ms without heartbeat`
: `agent run recovered from stale running state after heartbeat was stale for ${heartbeatAgeMs}ms`;
const staleError = Object.assign(
new Error(`agent run stale after ${ageMs}ms`),
{ code: 'AGENT_RUN_STALE_RECOVERY' },
);
const deliverableRecovered = await recoverRunFromDeliverables({
runId: row.id,
userId: row.user_id,
sessionId: row.agent_session_id ?? null,
err: staleError,
runStartedAtMs: row.started_at ?? null,
requireRecoverableError: false,
});
if (deliverableRecovered) {
await markRun(row.id, 'succeeded', {
agent_session_id: row.agent_session_id ?? null,
completed_at: nowMs(),
error_message: null,
});
item.status = 'succeeded';
item.recoveredAs = 'deliverables';
recovered.push(item);
continue;
}
const message = sessionFinishedAt != null
? `agent run recovered from stale running state after session finished ${Math.max(0, nowMs() - sessionFinishedAt)}ms ago`
: heartbeatAt == null
? `agent run recovered from stale running state after ${ageMs}ms without heartbeat`
: `agent run recovered from stale running state after running for ${ageMs}ms`;
const completedAt = nowMs();
const [update] = await pool.query(
`UPDATE h5_agent_runs
@@ -966,33 +1045,41 @@ export function createAgentRunGateway({
WHERE id = ?
AND status = 'running'
AND started_at IS NOT NULL
AND COALESCE(
(SELECT MAX(created_at)
FROM h5_agent_run_events
WHERE run_id = h5_agent_runs.id AND event_type = 'worker_heartbeat'),
started_at
) <= ?`,
[message, completedAt, completedAt, row.id, cutoff],
AND (
started_at <= ?
OR EXISTS (
SELECT 1
FROM h5_agent_run_events sf
WHERE sf.run_id = h5_agent_runs.id
AND sf.event_type = 'session_finished'
AND sf.created_at <= ?
)
)`,
[message, completedAt, completedAt, row.id, startedCutoff, sessionFinishedCutoff],
);
if (Number(update?.affectedRows ?? 0) === 0) continue;
await appendEvent(row.id, 'stale_recovered', {
reason,
staleMs: normalizedStaleMs,
sessionFinishedGraceMs: normalizedSessionFinishedGraceMs,
ageMs,
heartbeatAt,
heartbeatAgeMs,
sessionFinishedAt,
status: 'failed',
error: message,
});
item.status = 'failed';
}
recovered.push(item);
}
return {
dryRun,
staleMs: normalizedStaleMs,
cutoff,
startedCutoff,
sessionFinishedCutoff,
considered: rows.length,
recovered: dryRun ? 0 : recovered.length,
recovered: dryRun ? 0 : recovered.filter((item) => item.status).length,
runs: recovered,
};
}