From 4fe45bf8fe46c88f5e3130424fdae097f8573906 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 8 Oct 2026 14:53:48 -0500 Subject: [PATCH] refactor(heartbeat): extract run retrieval and session state (#15601) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Heartbeat runs retain task sessions, usage totals, and bounded run data. > - After the preparation extraction, the main service still has over 24,000 lines. > - Run reads and session operations share one database and have no dispatch dependencies. > - This pull request moves that group into one module and preserves its callers. > - The benefit is a smaller run engine and one place to maintain saved run and session state. ## Linked Issues or Issue Description **What existing behavior does this improve?** It improves the structure and test coverage of heartbeat run retrieval and session state. **Current behavior** `server/src/services/heartbeat.ts` has 24,033 lines on the merged base. It mixes run projections, session queries, resume rules, compaction, and usage helpers with execution and recovery. **Proposed behavior** Move that group into `server/src/services/heartbeat/run-state.ts`. The main file loses 1,770 lines and ends at 22,263 lines. Keep current behavior, public exports, and service methods. This continues #15568, #15573, #15578, and #15591. **Reason and benefit** One domain extraction makes useful progress toward a few manageable files. Run reads and saved session state can be reviewed together without searching the run engine. A search found no duplicate open extraction. Related session changes include #15142, #14661, and #14670; this PR does not apply their proposed behavior changes. **Breaking changes** None. Existing public helpers retain their identity. Service method names and results stay the same. ## What Changed - Move bounded run projections, encoding checks, and run/event reads into `heartbeat/run-state.ts`. - Move task session persistence, explicit resumes, reset rules, compaction, and usage/billing helpers into the same module. - Bind database operations through `createHeartbeatRunState(db)`. Construction does no database work. The encoding cache stays local to each service instance. - Keep execution order, status transitions, session-goal recovery, cost accounting writes, dispatch, and cancellation in `heartbeat.ts`. - Preserve all 140 exports. A syntax-tree comparison confirms that 43 moved function bodies, 22 declarations, and nine service method bodies are unchanged. The copied private string helper is unchanged. Surrounding orchestration is unchanged apart from the factory binding. - Add eight tests for legacy export identity, database binding, encoding cache isolation, scoped resumes, cumulative usage, compaction, and stale or cancelled conversation session writes. - Document the module boundary in `doc/DEVELOPING.md`. ## Verification - Before and after extraction, eight existing suites pass: 235 tests. They cover workspace/session helpers, runtime state, cost accounting, ledger attribution, task reset, timer reset, run lists, and run privacy. - The new module suite passes: eight tests, including five real PostgreSQL cases. Total focused coverage: 243 tests across nine suites. - Commands: `pnpm exec vitest run server/src/services/heartbeat/run-state.test.ts` and `pnpm exec vitest run server/src/__tests__/heartbeat-runtime-state.test.ts server/src/__tests__/heartbeat-cost-accounting.test.ts server/src/__tests__/heartbeat-ledger-billing-code.test.ts server/src/__tests__/heartbeat-task-session-reset.test.ts server/src/__tests__/heartbeat-workspace-session.test.ts server/src/__tests__/heartbeat-timer-wake-session-reset-pf4.test.ts server/src/__tests__/heartbeat-list.test.ts server/src/__tests__/heartbeat-run-privacy-routes.test.ts`. - `pnpm --filter @paperclipai/server typecheck` passes. - `pnpm -r typecheck` and `pnpm build` pass. - The full local `pnpm test:run` did not complete and was stopped with SIGINT after it reported an OAuth scope health-check failure and a workspace-runtime timeout. All six OAuth variants and the workspace case pass in isolation. A single rerun of the 413-case tool-access suite passed the OAuth case but hit a different Notion callback timeout; that Notion case also passes in isolation. The full local suite is not claimed as passing. - Greptile is 5/5 with no findings or inline comments on head `2ca0908cfc112dd6ae26e2565cb3611b5bec52ba`. Its exact-head check completed successfully. - CI remains blocked on hosted-runner assignment after more than ten minutes. [`ci / Select trusted runner`](https://github.com/paperclipai/paperclip/actions/runs/37828454507/job/113487287178) is queued for `ubuntu-latest` with no runner assigned. All seven completed review/security checks pass; two optional Storybook checks are skipped. The full CI matrix has not started. It must complete before merge. ## Risks - Wiring errors could bind reads to the wrong database or share the encoding cache. Direct factory tests cover database and cache isolation. Real PostgreSQL suites cover query and transaction behavior. - Session writes must retain their issue row locks and generation fences. The original function bodies are unchanged. New database tests reject stale and cancelled conversation writes and clears. - Existing SQL_ASCII safeguards and bounded projections remain in place. No schema, API contract, migration, lockfile, or workflow changes are included. ## Model Used OpenAI GPT-6 via Codex. The exact serving model ID and context window were not exposed in this session. Used reasoning, repository inspection, shell tools, and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- doc/DEVELOPING.md | 9 + server/src/services/heartbeat.ts | 1918 +--------------- .../src/services/heartbeat/run-state.test.ts | 222 ++ server/src/services/heartbeat/run-state.ts | 1923 +++++++++++++++++ 4 files changed, 2228 insertions(+), 1844 deletions(-) create mode 100644 server/src/services/heartbeat/run-state.test.ts create mode 100644 server/src/services/heartbeat/run-state.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 24fbd4ce96e..33b50a28b12 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1245,6 +1245,15 @@ dispatch, retries, cancellation, and execution order, and re-exports the existin public helpers and configuration-incomplete error class. Keep preparation policy changes in this module and its tests. +Run retrieval and session state are in `server/src/services/heartbeat/run-state.ts`. +It owns bounded run projections, database encoding checks, task session reads and +writes, explicit resumes, session compaction, and usage/billing helpers. +`createHeartbeatRunState(db)` binds these operations without doing database work +during construction. The encoding-check cache belongs to each factory instance. +`heartbeat.ts` keeps run execution, session-goal recovery, cost accounting writes, +and status transitions, and re-exports the existing public helpers. Keep session +policy changes separate from run orchestration changes. + ## Wake Context Delivery Built-in adapters deliver wake context through the run prompt, including structured diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 7c62ed73347..a29bf8b4454 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,41 @@ +import { + type RunSessionOutcome, + EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON, + deriveTaskKeyWithHeartbeatFallback, + type UsageTotals, + getAdapterSessionCodec, + shouldResetTaskSessionForWake, + normalizeSessionParams, + normalizeResumeParamsForAdapter, + truncateDisplayId, + resolveLedgerScopeForRun, + describeSessionResetReason, + isCanonicalSessionIdForAdapter, + requiresCanonicalSessionIds, + resolveNextSessionState, + normalizeUsageTotals, + resolveCacheAdjustedCostUsd, + resolveLedgerBiller, + resolveLedgerCostStatus, + normalizeLedgerBillingType, + createHeartbeatRunState, +} from "./heartbeat/run-state.js"; +export { + summarizeHeartbeatRunContextSnapshot, + summarizeHeartbeatRunListResultJson, + normalizeBilledCostCents, + resolveLedgerCostStatus, + resolveCacheAdjustedCostUsd, + resolveLedgerScopeForRun, + buildExplicitResumeSessionOverride, + normalizeAdapterRunUsage, + parseSessionCompactionPolicy, + deriveTaskKeyWithHeartbeatFallback, + shouldResetTaskSessionForWake, + describeSessionResetReason, + normalizeSessionParams, + resolveNextSessionState, +} from "./heartbeat/run-state.js"; import { readGitConnectionFailure } from "./git-connection-failure.js"; import { CONFIGURATION_INCOMPLETE_FAILURE_CODE, @@ -158,7 +196,7 @@ import { activeIssueInteractionCondition } from "./issue-question-context.js"; import { createAgentIdentityRedactor } from "./agent-identity-redaction.js"; import { agentIdentityService, supportsManagedAgentIdentity } from "./agent-identity.js"; import { buildAgentIdentityEnv } from "@paperclipai/adapter-utils/server-utils"; -import { retryIdempotentDatabaseOperation } from "../database-retry.js"; + import { prepareConnectionInstructionDelivery } from "./connection-instructions.js"; import { resolveAssignedConnectionInstructionsForRun } from "./native-runtime/assigned-mcp-tools.js"; import { externalObjectService } from "./external-objects.js"; @@ -295,8 +333,6 @@ import { ISSUE_DISPOSITION_REPAIR_RETRY_REASON, PROVIDER_QUOTA_MONITOR_SERVICE_NAME, isEnvironmentDriverSupportedForAdapter, - type BillingType, - type CostStatus, type EnvironmentLeaseStatus, type ExecutionWorkspace, type HeartbeatRunStatusPhase, @@ -446,8 +482,6 @@ import type { AdapterExecutionResult, AdapterInvocationMeta, AdapterRuntimeEvent, - AdapterSessionCodec, - UsageSummary, } from "../adapters/index.js"; import { createLocalAgentJwt } from "../agent-auth-jwt.js"; import { createRuntimeToolsToken } from "../runtime-tools-token.js"; @@ -498,16 +532,12 @@ import { buildHeartbeatRunIssueComment, CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON, findHeartbeatRunCompletionComment, - HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS, - HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS, - HEARTBEAT_RUN_SAFE_RESULT_JSON_MAX_BYTES, hasAcceptedSemanticResult, isExternalChatPresentationContext, mergeHeartbeatRunResultJson, readCompletedAssistantMessageCandidate, resolveHeartbeatRunResponse, selectHeartbeatRunFinalAgentMessage, - summarizeRunErrorForModel, type RunPresentationDecision, } from "./heartbeat-run-summary.js"; import { @@ -723,11 +753,8 @@ import { import { redactEventPayload, redactSensitiveText } from "../redaction.js"; import { createRunSecretRedactionRegistry } from "./run-secret-redaction.js"; import { - hasSessionCompactionThresholds, resolvePaperclipRunnerIdleTimeoutMs, - resolveSessionCompactionPolicy, type RuntimeStatusUpdate, - type SessionCompactionPolicy, } from "@paperclipai/adapter-utils"; import { readPaperclipSkillSyncPreference, @@ -748,7 +775,7 @@ import { } from "./environment-runtime.js"; import { skillVersionSelectionMap } from "./runtime-skill-selections.js"; import { environmentRunOrchestrator } from "./environment-run-orchestrator.js"; -import { isUnsafeSessionWorkspaceCwd } from "./session-workspace-cwd.js"; + import { clearHeartbeatRunRuntimeStatus, getHeartbeatRunRuntimeStatus, @@ -964,8 +991,7 @@ const WORKSPACE_VALIDATION_RECOVERY_CAUSE = "workspace_validation_failed"; const CONFIGURATION_INCOMPLETE_RECOVERY_CAUSE = "configuration_incomplete"; const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_RETRY_REASON = "execution_review_participant_recovery"; -const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON = - "execution_review_participant_recovery"; + const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE = "execution_review_participant_recovery"; @@ -1540,322 +1566,7 @@ export function isConfigurationIncompleteFailedRun( ); } -const heartbeatRunProcessGroupIdColumn = - heartbeatRuns.processGroupId ?? sql`NULL`.as("processGroupId"); -const heartbeatRunListColumns = { - id: heartbeatRuns.id, - responsibleUserId: heartbeatRuns.responsibleUserId, - companyId: heartbeatRuns.companyId, - agentId: heartbeatRuns.agentId, - scopeKind: heartbeatRuns.scopeKind, - issueId: heartbeatRuns.issueId, - invocationSource: heartbeatRuns.invocationSource, - triggerDetail: heartbeatRuns.triggerDetail, - status: heartbeatRuns.status, - inputTokens: sql`(${heartbeatRuns.usageJson} ->> 'inputTokens')::numeric`.as("inputTokens"), - cachedInputTokens: sql`(${heartbeatRuns.usageJson} ->> 'cachedInputTokens')::numeric`.as("cachedInputTokens"), - outputTokens: sql`(${heartbeatRuns.usageJson} ->> 'outputTokens')::numeric`.as("outputTokens"), - totalTokens: sql`(${heartbeatRuns.usageJson} ->> 'totalTokens')::numeric`.as("totalTokens"), - costUsd: sql`coalesce( - (${heartbeatRuns.resultJson} ->> 'costUsd')::numeric, - (${heartbeatRuns.resultJson} ->> 'cost_usd')::numeric, - (${heartbeatRuns.resultJson} ->> 'total_cost_usd')::numeric - )`.as("costUsd"), - startedAt: heartbeatRuns.startedAt, - finishedAt: heartbeatRuns.finishedAt, - error: heartbeatRuns.error, - wakeupRequestId: heartbeatRuns.wakeupRequestId, - exitCode: heartbeatRuns.exitCode, - signal: heartbeatRuns.signal, - usageJson: heartbeatRuns.usageJson, - sessionIdBefore: heartbeatRuns.sessionIdBefore, - sessionIdAfter: heartbeatRuns.sessionIdAfter, - logStore: heartbeatRuns.logStore, - logRef: heartbeatRuns.logRef, - logBytes: heartbeatRuns.logBytes, - logSha256: heartbeatRuns.logSha256, - logCompressed: heartbeatRuns.logCompressed, - stdoutExcerpt: sql`NULL`.as("stdoutExcerpt"), - stderrExcerpt: sql`NULL`.as("stderrExcerpt"), - errorCode: heartbeatRuns.errorCode, - externalRunId: heartbeatRuns.externalRunId, - processPid: heartbeatRuns.processPid, - processGroupId: heartbeatRunProcessGroupIdColumn, - processStartedAt: heartbeatRuns.processStartedAt, - lastOutputAt: heartbeatRuns.lastOutputAt, - lastOutputSeq: heartbeatRuns.lastOutputSeq, - lastOutputStream: heartbeatRuns.lastOutputStream, - lastOutputBytes: heartbeatRuns.lastOutputBytes, - retryOfRunId: heartbeatRuns.retryOfRunId, - processLossRetryCount: heartbeatRuns.processLossRetryCount, - scheduledRetryAt: heartbeatRuns.scheduledRetryAt, - scheduledRetryAttempt: heartbeatRuns.scheduledRetryAttempt, - scheduledRetryReason: heartbeatRuns.scheduledRetryReason, - livenessState: heartbeatRuns.livenessState, - livenessReason: heartbeatRuns.livenessReason, - continuationAttempt: heartbeatRuns.continuationAttempt, - lastUsefulActionAt: heartbeatRuns.lastUsefulActionAt, - nextAction: heartbeatRuns.nextAction, - createdAt: heartbeatRuns.createdAt, - updatedAt: heartbeatRuns.updatedAt, -} as const; - -const heartbeatRunSummaryListColumns = { - ...heartbeatRunListColumns, - usageJson: sql | null>`NULL`.as("usageJson"), - sessionIdBefore: sql`NULL`.as("sessionIdBefore"), - sessionIdAfter: sql`NULL`.as("sessionIdAfter"), - logStore: sql`NULL`.as("logStore"), - logRef: sql`NULL`.as("logRef"), - logSha256: sql`NULL`.as("logSha256"), - externalRunId: sql`NULL`.as("externalRunId"), - processPid: sql`NULL`.as("processPid"), - processGroupId: sql`NULL`.as("processGroupId"), - resultJson: sql | null>`NULL`.as("resultJson"), -} as const; - -const heartbeatRunListContextColumns = { - contextIssueId: sql`coalesce( - ${heartbeatRuns.issueId}::text, - ${heartbeatRuns.contextSnapshot} ->> 'issueId' - )`.as("contextIssueId"), - contextTaskId: sql`${heartbeatRuns.contextSnapshot} ->> 'taskId'`.as("contextTaskId"), - contextTaskKey: sql`${heartbeatRuns.contextSnapshot} ->> 'taskKey'`.as("contextTaskKey"), - contextCommentId: sql`${heartbeatRuns.contextSnapshot} ->> 'commentId'`.as("contextCommentId"), - contextWakeCommentId: sql`${heartbeatRuns.contextSnapshot} ->> 'wakeCommentId'`.as("contextWakeCommentId"), - contextWakeReason: sql`${heartbeatRuns.contextSnapshot} ->> 'wakeReason'`.as("contextWakeReason"), - contextWakeSource: sql`${heartbeatRuns.contextSnapshot} ->> 'wakeSource'`.as("contextWakeSource"), - contextWakeTriggerDetail: sql`${heartbeatRuns.contextSnapshot} ->> 'wakeTriggerDetail'`.as("contextWakeTriggerDetail"), -} as const; - -const heartbeatRunListResultColumns = { - resultSummary: sql< - string | null - >`left(${heartbeatRuns.resultJson} ->> 'summary', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS})`.as( - "resultSummary", - ), - resultResult: sql< - string | null - >`left(${heartbeatRuns.resultJson} ->> 'result', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS})`.as( - "resultResult", - ), - resultMessage: sql< - string | null - >`left(${heartbeatRuns.resultJson} ->> 'message', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS})`.as( - "resultMessage", - ), - resultError: sql< - string | null - >`left(${heartbeatRuns.resultJson} ->> 'error', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS})`.as( - "resultError", - ), - resultTotalCostUsd: sql< - string | null - >`${heartbeatRuns.resultJson} ->> 'total_cost_usd'`.as("resultTotalCostUsd"), - resultCostUsd: sql< - string | null - >`${heartbeatRuns.resultJson} ->> 'cost_usd'`.as("resultCostUsd"), - resultCostUsdCamel: sql< - string | null - >`${heartbeatRuns.resultJson} ->> 'costUsd'`.as("resultCostUsdCamel"), -} as const; - -// Reserve at most 9 KiB for diagnostics in the reduced result. An oversized -// multibyte field uses a conservative four-byte-per-character prefix, with a -// visible pointer to the full (adapter-bounded) run error and transcript. -const diagnosticRetrievalTitleBytes = 1024; -const diagnosticRetrievalDetailsBytes = 8192; -function boundedRunDiagnosticText(field: "title" | "details", maxBytes: number) { - const value = sql`${heartbeatRuns.resultJson} #>> ARRAY['terminalSessionFailure', ${field}]`; - return sql`case when octet_length(${value}) <= ${maxBytes} then ${value} - else left(${value}, ${Math.floor((maxBytes - 100) / 4)}) - || E'\\n[truncated for run retrieval; full text in run error/transcript]' end`; -} - -const heartbeatRunSafeResultJsonColumn = sql | null>` - case - when ${heartbeatRuns.resultJson} is null then null - when pg_column_size(${heartbeatRuns.resultJson}) <= ${HEARTBEAT_RUN_SAFE_RESULT_JSON_MAX_BYTES} - then ${heartbeatRuns.resultJson} - else jsonb_strip_nulls( - jsonb_build_object( - 'summary', left(${heartbeatRuns.resultJson} ->> 'summary', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS}), - 'result', left(${heartbeatRuns.resultJson} ->> 'result', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS}), - 'message', left(${heartbeatRuns.resultJson} ->> 'message', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS}), - 'error', left(${heartbeatRuns.resultJson} ->> 'error', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS}), - 'stdout', left(${heartbeatRuns.resultJson} ->> 'stdout', ${HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS}), - 'stderr', left(${heartbeatRuns.resultJson} ->> 'stderr', ${HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS}), - 'terminalSessionFailure', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'terminalSessionFailure') = 'object' - then jsonb_strip_nulls(jsonb_build_object( - 'category', left(${heartbeatRuns.resultJson} #>> '{terminalSessionFailure,category}', 32), - 'title', ${boundedRunDiagnosticText("title", diagnosticRetrievalTitleBytes)}, - 'details', ${boundedRunDiagnosticText("details", diagnosticRetrievalDetailsBytes)}, - 'retrievalTruncated', case when - octet_length(${heartbeatRuns.resultJson} #>> '{terminalSessionFailure,title}') > ${diagnosticRetrievalTitleBytes} - or octet_length(${heartbeatRuns.resultJson} #>> '{terminalSessionFailure,details}') > ${diagnosticRetrievalDetailsBytes} - then to_jsonb(true) end, - 'truncatedFields', case when ${heartbeatRuns.resultJson} #> '{terminalSessionFailure,truncatedFields}' - in ('["title"]'::jsonb, '["details"]'::jsonb, '["title","details"]'::jsonb) - then ${heartbeatRuns.resultJson} #> '{terminalSessionFailure,truncatedFields}' end - )) end, - 'instructionSave', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'instructionSave') = 'object' - then jsonb_strip_nulls(jsonb_build_object( - 'state', left(${heartbeatRuns.resultJson} #>> '{instructionSave,state}', 32), - 'contract', left(${heartbeatRuns.resultJson} #>> '{instructionSave,contract}', 32), - 'entryFile', left(${heartbeatRuns.resultJson} #>> '{instructionSave,entryFile}', 512), - 'errorCode', left(${heartbeatRuns.resultJson} #>> '{instructionSave,errorCode}', 128), - 'errorMessage', left(${heartbeatRuns.resultJson} #>> '{instructionSave,errorMessage}', 1024), - 'storageWarning', left(${heartbeatRuns.resultJson} #>> '{instructionSave,storageWarning}', 1024) - )) end, - 'workspaceRestoreRecovery', case when ${heartbeatRuns.resultJson} #>> '{workspaceRestoreRecovery,schema}' = 'paperclip.workspace-restore-recovery.v1' - then jsonb_build_object('schema', 'paperclip.workspace-restore-recovery.v1') end, - 'workspaceRestoreFailure', case when ${heartbeatRuns.resultJson} ->> 'workspaceRestoreFailure' - in ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed') - then ${heartbeatRuns.resultJson} -> 'workspaceRestoreFailure' end, - 'cancellation', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'cancellation') = 'object' - then jsonb_strip_nulls(jsonb_build_object( - 'source', case when ${heartbeatRuns.resultJson} #>> '{cancellation,source}' - in ('operator', 'queued_message', 'shutdown', 'provider', 'transport', 'control_plane', 'unknown') - then ${heartbeatRuns.resultJson} #> '{cancellation,source}' end, - 'expected', case when jsonb_typeof(${heartbeatRuns.resultJson} #> '{cancellation,expected}') = 'boolean' - then ${heartbeatRuns.resultJson} #> '{cancellation,expected}' end, - 'initiator', jsonb_strip_nulls(jsonb_build_object( - 'type', case when ${heartbeatRuns.resultJson} #>> '{cancellation,initiator,type}' in ('user', 'agent', 'system', 'provider') - then ${heartbeatRuns.resultJson} #> '{cancellation,initiator,type}' end, - 'id', left(${heartbeatRuns.resultJson} #>> '{cancellation,initiator,id}', 128) - )), - 'reason', left(${heartbeatRuns.resultJson} #>> '{cancellation,reason}', 512), - 'recordedAt', left(${heartbeatRuns.resultJson} #>> '{cancellation,recordedAt}', 64) - )) end, - 'acpToolInventoryComplete', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'acpToolInventoryComplete') = 'boolean' - then ${heartbeatRuns.resultJson} -> 'acpToolInventoryComplete' end, - 'acpPendingToolCount', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'acpPendingToolCount') = 'number' - and length(${heartbeatRuns.resultJson} ->> 'acpPendingToolCount') < 16 - then ${heartbeatRuns.resultJson} -> 'acpPendingToolCount' end, - 'errorFamily', left(${heartbeatRuns.resultJson} ->> 'errorFamily', 32), - 'finalResponseRecorded', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'finalResponseRecorded') = 'boolean' - then ${heartbeatRuns.resultJson} -> 'finalResponseRecorded' end, - 'executionBeforeRestore', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'executionBeforeRestore') = 'object' - then jsonb_strip_nulls(jsonb_build_object( - 'errorCode', left(${heartbeatRuns.resultJson} #>> '{executionBeforeRestore,errorCode}', 128), - 'exitCode', case when jsonb_typeof(${heartbeatRuns.resultJson} #> '{executionBeforeRestore,exitCode}') = 'number' - and length(${heartbeatRuns.resultJson} #>> '{executionBeforeRestore,exitCode}') < 16 - then ${heartbeatRuns.resultJson} #> '{executionBeforeRestore,exitCode}' end, - 'signal', left(${heartbeatRuns.resultJson} #>> '{executionBeforeRestore,signal}', 50), - 'timedOut', case when jsonb_typeof(${heartbeatRuns.resultJson} #> '{executionBeforeRestore,timedOut}') = 'boolean' - then ${heartbeatRuns.resultJson} #> '{executionBeforeRestore,timedOut}' end - )) end, - 'stdoutTruncated', case - when length(${heartbeatRuns.resultJson} ->> 'stdout') > ${HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS} - then to_jsonb(true) - else null - end, - 'stderrTruncated', case - when length(${heartbeatRuns.resultJson} ->> 'stderr') > ${HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS} - then to_jsonb(true) - else null - end, - 'costUsd', coalesce( - ${heartbeatRuns.resultJson} -> 'costUsd', - ${heartbeatRuns.resultJson} -> 'cost_usd', - ${heartbeatRuns.resultJson} -> 'total_cost_usd' - ), - 'cost_usd', coalesce( - ${heartbeatRuns.resultJson} -> 'cost_usd', - ${heartbeatRuns.resultJson} -> 'costUsd', - ${heartbeatRuns.resultJson} -> 'total_cost_usd' - ), - 'total_cost_usd', coalesce( - ${heartbeatRuns.resultJson} -> 'total_cost_usd', - ${heartbeatRuns.resultJson} -> 'cost_usd', - ${heartbeatRuns.resultJson} -> 'costUsd' - ), - 'truncated', true, - 'truncationReason', 'oversized_result_json', - 'originalSizeBytes', pg_column_size(${heartbeatRuns.resultJson}) - ) - ) - end -`.as("resultJson"); - -// Execution admission needs retained server receipts even when presentation -// projection omits resultJson (SQL_ASCII or oversized provider output). Select -// only the evidence used by eligibility; never retrieve provider diagnostics. -const heartbeatRunExecutionEvidenceColumn = sql | null>` - case when ${heartbeatRuns.resultJson} is null then null else jsonb_build_object( - 'startupCancellation', ${heartbeatRuns.resultJson} -> 'startupCancellation', - 'startupPreparationSettledAt', ${heartbeatRuns.resultJson} -> 'startupPreparationSettledAt', - 'stopReason', ${heartbeatRuns.resultJson} -> 'stopReason', - 'timeoutSource', ${heartbeatRuns.resultJson} -> 'timeoutSource', - 'workspaceRestoreFailure', ${heartbeatRuns.resultJson} -> 'workspaceRestoreFailure', - 'workspaceRestoreRecovery', ${heartbeatRuns.resultJson} -> 'workspaceRestoreRecovery', - 'executionCancellation', ${heartbeatRuns.resultJson} -> 'executionCancellation', - 'nativeCancellation', ${heartbeatRuns.resultJson} -> 'nativeCancellation', - 'cancelledByActorType', ${heartbeatRuns.resultJson} -> 'cancelledByActorType', - 'cancelledByUserId', ${heartbeatRuns.resultJson} -> 'cancelledByUserId', - 'conversationContinuation', ${heartbeatRuns.resultJson} -> 'conversationContinuation', - 'cancellation', ${heartbeatRuns.resultJson} -> 'cancellation', - 'acpToolInventoryComplete', ${heartbeatRuns.resultJson} -> 'acpToolInventoryComplete', - 'acpPendingToolCount', ${heartbeatRuns.resultJson} -> 'acpPendingToolCount' - ) end -`.as("resultJson"); - -const heartbeatRunSafeColumns = { - ...getTableColumns(heartbeatRuns), - processGroupId: heartbeatRunProcessGroupIdColumn, - resultJson: heartbeatRunSafeResultJsonColumn, -} as const; - -const heartbeatRunSqlAsciiSafeColumns = { - ...getTableColumns(heartbeatRuns), - processGroupId: heartbeatRunProcessGroupIdColumn, - error: sql`NULL`.as("error"), - resultJson: sql | null>`NULL`.as("resultJson"), - stdoutExcerpt: sql`NULL`.as("stdoutExcerpt"), - stderrExcerpt: sql`NULL`.as("stderrExcerpt"), -} as const; - -const heartbeatRunLogAccessColumns = { - id: heartbeatRuns.id, - companyId: heartbeatRuns.companyId, - scopeKind: heartbeatRuns.scopeKind, - issueId: heartbeatRuns.issueId, - logStore: heartbeatRuns.logStore, - logRef: heartbeatRuns.logRef, -} as const; - -const heartbeatRunIssueSummaryColumns = { - id: heartbeatRuns.id, - runtimeMode: heartbeatRuns.runtimeMode, - status: heartbeatRuns.status, - invocationSource: heartbeatRuns.invocationSource, - triggerDetail: heartbeatRuns.triggerDetail, - contextCommentId: sql< - string | null - >`${heartbeatRuns.contextSnapshot} ->> 'commentId'`.as("contextCommentId"), - contextWakeCommentId: sql< - string | null - >`${heartbeatRuns.contextSnapshot} ->> 'wakeCommentId'`.as( - "contextWakeCommentId", - ), - startedAt: heartbeatRuns.startedAt, - finishedAt: heartbeatRuns.finishedAt, - createdAt: heartbeatRuns.createdAt, - agentId: heartbeatRuns.agentId, - logBytes: heartbeatRuns.logBytes, - processStartedAt: heartbeatRuns.processStartedAt, - livenessState: heartbeatRuns.livenessState, - livenessReason: heartbeatRuns.livenessReason, - continuationAttempt: heartbeatRuns.continuationAttempt, - lastUsefulActionAt: heartbeatRuns.lastUsefulActionAt, - nextAction: heartbeatRuns.nextAction, - lastOutputAt: heartbeatRuns.lastOutputAt, - lastOutputSeq: heartbeatRuns.lastOutputSeq, - lastOutputStream: heartbeatRuns.lastOutputStream, - lastOutputBytes: heartbeatRuns.lastOutputBytes, - issueId: heartbeatRuns.issueId, -} as const; function normalizeMaxConcurrentRuns(value: unknown) { const parsed = Math.floor( @@ -1868,19 +1579,6 @@ function normalizeMaxConcurrentRuns(value: unknown) { ); } -type UsageTotals = { - inputTokens: number; - cachedInputTokens: number; - outputTokens: number; -}; - -type SessionCompactionDecision = { - rotate: boolean; - reason: string | null; - handoffMarkdown: string | null; - previousRunId: string | null; -}; - interface ParsedIssueAssigneeAdapterOverrides { adapterConfig: Record | null; useProjectWorkspace: boolean | null; @@ -1920,353 +1618,6 @@ function parseNativeSessionGoalControl( } -export function summarizeHeartbeatRunContextSnapshot( - contextSnapshot: Record | null | undefined, -): Record | null { - const summary: Record = {}; - const allowedKeys = [ - "issueId", - "taskId", - "taskKey", - "commentId", - "wakeCommentId", - "wakeReason", - "wakeSource", - "wakeTriggerDetail", - ] as const; - - for (const key of allowedKeys) { - const value = readNonEmptyString(contextSnapshot?.[key]); - if (value) summary[key] = value; - } - - return Object.keys(summary).length > 0 ? summary : null; -} - -export function summarizeHeartbeatRunListResultJson(input: { - summary?: string | null; - result?: string | null; - message?: string | null; - error?: string | null; - totalCostUsd?: string | null; - costUsd?: string | null; - costUsdCamel?: string | null; -}): Record | null { - const summary: Record = {}; - for (const [key, value] of [ - ["summary", input.summary], - ["result", input.result], - ["message", input.message], - ["error", input.error], - ] as const) { - const normalized = readNonEmptyString(value); - if (normalized) summary[key] = normalized; - } - - for (const [key, value] of [ - ["total_cost_usd", input.totalCostUsd], - ["cost_usd", input.costUsd], - ["costUsd", input.costUsdCamel], - ] as const) { - const normalized = readNonEmptyString(value); - if (!normalized) continue; - const parsed = Number(normalized); - if (Number.isFinite(parsed)) summary[key] = parsed; - } - - return Object.keys(summary).length > 0 ? summary : null; -} - -function normalizeLedgerBillingType(value: unknown): BillingType { - const raw = readNonEmptyString(value); - switch (raw) { - case "api": - case "metered_api": - return "metered_api"; - case "subscription": - case "subscription_included": - return "subscription_included"; - case "subscription_overage": - return "subscription_overage"; - case "credits": - return "credits"; - case "fixed": - return "fixed"; - default: - return "unknown"; - } -} - -function resolveLedgerBiller(result: AdapterExecutionResult): string { - return ( - readNonEmptyString(result.biller) ?? - readNonEmptyString(result.provider) ?? - "unknown" - ); -} - -export function normalizeBilledCostCents( - costUsd: number | null | undefined, - billingType: BillingType, -): number { - if (billingType === "subscription_included") return 0; - if (typeof costUsd !== "number" || !Number.isFinite(costUsd)) return 0; - return Math.max(0, Number((costUsd * 100).toFixed(7))); -} - -export function resolveLedgerCostStatus(input: { - costUsd: number | null | undefined; - billingType?: BillingType; - inputTokens: number; - cachedInputTokens: number; - outputTokens: number; -}): CostStatus { - if (input.billingType === "subscription_included") return "reported"; - // A paused turn can have neither a token receipt nor a cost receipt. Zero - // normalized counters do not establish that its billed cost was zero. - return typeof input.costUsd === "number" && - Number.isFinite(input.costUsd) && - input.costUsd >= 0 - ? "reported" - : "unpriced"; -} - -export function resolveCacheAdjustedCostUsd(input: { - costUsd?: number | null; - cacheAdjustedCostUsd?: number | null; -}) { - const explicit = input.cacheAdjustedCostUsd; - if ( - typeof explicit === "number" && - Number.isFinite(explicit) && - explicit >= 0 - ) { - return explicit; - } - const reported = input.costUsd; - if ( - typeof reported === "number" && - Number.isFinite(reported) && - reported >= 0 - ) { - return reported; - } - return null; -} - -export async function resolveLedgerScopeForRun( - db: Db, - companyId: string, - run: typeof heartbeatRuns.$inferSelect, -) { - const context = parseObject(run.contextSnapshot); - const contextIssueId = readNonEmptyString(context.issueId); - const contextProjectId = readNonEmptyString(context.projectId); - - if (!contextIssueId) { - return { - issueId: null, - projectId: contextProjectId, - billingCode: null, - }; - } - - const issue = await db - .select({ - id: issues.id, - projectId: issues.projectId, - billingCode: issues.billingCode, - }) - .from(issues) - .where(and(eq(issues.id, contextIssueId), eq(issues.companyId, companyId))) - .then((rows) => rows[0] ?? null); - - return { - issueId: issue?.id ?? null, - projectId: issue?.projectId ?? contextProjectId, - billingCode: issue?.billingCode ?? null, - }; -} - -type ResumeSessionRow = { - sessionParamsJson: Record | null; - sessionDisplayId: string | null; - lastRunId: string | null; -}; - -export function buildExplicitResumeSessionOverride(input: { - adapterType?: string | null; - resumeFromRunId: string; - resumeRunSessionIdBefore: string | null; - resumeRunSessionIdAfter: string | null; - resumeRunSessionParams?: Record | null; - taskSession: ResumeSessionRow | null; - sessionCodec: AdapterSessionCodec; -}) { - const resumeRunSessionIdAfter = truncateDisplayId( - input.resumeRunSessionIdAfter, - ); - const resumeRunSessionIdBefore = truncateDisplayId( - input.resumeRunSessionIdBefore, - ); - const desiredDisplayId = requiresCanonicalSessionIds(input.adapterType) - ? isCanonicalSessionIdForAdapter(input.adapterType, resumeRunSessionIdAfter) - ? resumeRunSessionIdAfter - : isCanonicalSessionIdForAdapter( - input.adapterType, - resumeRunSessionIdBefore, - ) - ? resumeRunSessionIdBefore - : null - : (resumeRunSessionIdAfter ?? resumeRunSessionIdBefore); - const runSessionParams = requiresCanonicalSessionIds(input.adapterType) - ? normalizeResumeParamsForAdapter( - input.adapterType, - input.sessionCodec.deserialize(input.resumeRunSessionParams ?? null), - ) - : null; - const runSessionDisplayId = truncateDisplayId( - readNonEmptyString(runSessionParams?.sessionId), - ); - const taskSessionParams = normalizeResumeParamsForAdapter( - input.adapterType, - input.sessionCodec.deserialize( - input.taskSession?.sessionParamsJson ?? null, - ), - ); - const taskSessionRawDisplayId = input.taskSession?.sessionDisplayId ?? null; - const taskSessionDisplayId = truncateDisplayId( - requiresCanonicalSessionIds(input.adapterType) - ? (readNonEmptyString(taskSessionParams?.sessionId) ?? - (isCanonicalSessionIdForAdapter( - input.adapterType, - taskSessionRawDisplayId, - ) - ? taskSessionRawDisplayId - : null)) - : (taskSessionRawDisplayId ?? - (input.sessionCodec.getDisplayId - ? input.sessionCodec.getDisplayId(taskSessionParams) - : null) ?? - readNonEmptyString(taskSessionParams?.sessionId)), - ); - const canReuseTaskSessionParams = - input.taskSession != null && - (!requiresCanonicalSessionIds(input.adapterType) || - taskSessionParams != null) && - (input.taskSession.lastRunId === input.resumeFromRunId || - (!!desiredDisplayId && taskSessionDisplayId === desiredDisplayId)); - const sessionParams = canReuseTaskSessionParams - ? taskSessionParams - : runSessionParams - ? runSessionParams - : desiredDisplayId - ? { sessionId: desiredDisplayId } - : null; - const sessionDisplayId = canReuseTaskSessionParams - ? taskSessionDisplayId - : runSessionParams - ? runSessionDisplayId - : desiredDisplayId; - - if (!sessionDisplayId && !sessionParams) return null; - return { - sessionDisplayId, - sessionParams, - }; -} - -function normalizeUsageTotals( - usage: UsageSummary | null | undefined, -): UsageTotals | null { - if (!usage) return null; - return { - inputTokens: Math.max(0, Math.floor(asNumber(usage.inputTokens, 0))), - cachedInputTokens: Math.max( - 0, - Math.floor(asNumber(usage.cachedInputTokens, 0)), - ), - outputTokens: Math.max(0, Math.floor(asNumber(usage.outputTokens, 0))), - }; -} - -function readRawUsageTotals(usageJson: unknown): UsageTotals | null { - const parsed = parseObject(usageJson); - if (Object.keys(parsed).length === 0) return null; - - const inputTokens = Math.max( - 0, - Math.floor( - asNumber(parsed.rawInputTokens, asNumber(parsed.inputTokens, 0)), - ), - ); - const cachedInputTokens = Math.max( - 0, - Math.floor( - asNumber( - parsed.rawCachedInputTokens, - asNumber(parsed.cachedInputTokens, 0), - ), - ), - ); - const outputTokens = Math.max( - 0, - Math.floor( - asNumber(parsed.rawOutputTokens, asNumber(parsed.outputTokens, 0)), - ), - ); - - if (inputTokens <= 0 && cachedInputTokens <= 0 && outputTokens <= 0) { - return null; - } - - return { - inputTokens, - cachedInputTokens, - outputTokens, - }; -} - -export function normalizeAdapterRunUsage( - current: UsageTotals | null, - previous: UsageTotals | null, - usageBasis?: "per_run" | "session_cumulative" | null, -): UsageTotals | null { - if (!current) return null; - if (!previous || usageBasis !== "session_cumulative") return { ...current }; - - const inputTokens = - current.inputTokens >= previous.inputTokens - ? current.inputTokens - previous.inputTokens - : current.inputTokens; - const cachedInputTokens = - current.cachedInputTokens >= previous.cachedInputTokens - ? current.cachedInputTokens - previous.cachedInputTokens - : current.cachedInputTokens; - const outputTokens = - current.outputTokens >= previous.outputTokens - ? current.outputTokens - previous.outputTokens - : current.outputTokens; - - return { - inputTokens: Math.max(0, inputTokens), - cachedInputTokens: Math.max(0, cachedInputTokens), - outputTokens: Math.max(0, outputTokens), - }; -} - -function formatCount(value: number | null | undefined) { - if (typeof value !== "number" || !Number.isFinite(value)) return "0"; - return value.toLocaleString("en-US"); -} - -export function parseSessionCompactionPolicy( - agent: typeof agents.$inferSelect, -): SessionCompactionPolicy { - return resolveSessionCompactionPolicy(agent.adapterType, agent.runtimeConfig) - .policy; -} - function parseIssueAssigneeAdapterOverrides( raw: unknown, ): ParsedIssueAssigneeAdapterOverrides | null { @@ -2285,60 +1636,6 @@ function parseIssueAssigneeAdapterOverrides( }; } -/** - * Synthetic task key for timer/heartbeat wakes that have no issue context. - * This allows timer wakes to participate in the `agentTaskSessions` system - * and benefit from robust session resume, instead of relying solely on the - * simpler `agentRuntimeState.sessionId` fallback. - */ -const HEARTBEAT_TASK_KEY = "__heartbeat__"; - -/** - * Extended task key derivation that falls back to a stable synthetic key - * for timer/heartbeat wakes. The synthetic key keeps the - * `agentTaskSessions` row addressable across heartbeats so the row can be - * cleared and re-keyed deterministically. Unscoped exploratory timer wakes - * still start fresh to avoid accumulating low-value inbox scans, while timer - * wakes scoped to a real issue reuse that issue's task session. - * - * The synthetic key is only used when: - * - No explicit task/issue key exists in the context - * - The wake source is "timer" (scheduled heartbeat) - */ -export function deriveTaskKeyWithHeartbeatFallback( - contextSnapshot: Record | null | undefined, - payload: Record | null | undefined, -) { - const explicit = deriveTaskKey(contextSnapshot, payload); - if (explicit) return explicit; - - const wakeSource = readNonEmptyString(contextSnapshot?.wakeSource); - if (wakeSource === "timer") return HEARTBEAT_TASK_KEY; - - return null; -} - -export function shouldResetTaskSessionForWake( - contextSnapshot: Record | null | undefined, -) { - if (contextSnapshot?.forceFreshSession === true) return true; - - const wakeReason = readNonEmptyString(contextSnapshot?.wakeReason); - if ( - wakeReason === "issue_assigned" || - wakeReason === EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON || - wakeReason === "execution_approval_requested" || - // PF-4: unscoped timer wakes are exploratory ("any new work?") and should - // not accumulate low-value inbox scans. Issue-scoped timer wakes are - // continuation work, so reuse their task session to avoid paying the full - // session-start and re-orientation cost on every heartbeat. - (wakeReason === "heartbeat_timer" && !deriveTaskKey(contextSnapshot, null)) - ) { - return true; - } - return false; -} - function shouldRequireIssueCommentForWake( contextSnapshot: Record | null | undefined, ) { @@ -2419,30 +1716,6 @@ export function filterZombieCoalesceTarget< return target && isZombieRun(target, tracked) ? null : target; } -export function describeSessionResetReason( - contextSnapshot: Record | null | undefined, -) { - if (contextSnapshot?.forceFreshSession === true) - return "forceFreshSession was requested"; - - const wakeReason = readNonEmptyString(contextSnapshot?.wakeReason); - if (wakeReason === "issue_assigned") return "wake reason is issue_assigned"; - if (wakeReason === EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON) { - return `wake reason is ${EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON}`; - } - if (wakeReason === "execution_approval_requested") - return "wake reason is execution_approval_requested"; - // PF-4: paired with shouldResetTaskSessionForWake — keep the reason wording - // explicit so run logs make session reuse/reset behavior legible. - if ( - wakeReason === "heartbeat_timer" && - !deriveTaskKey(contextSnapshot, null) - ) { - return "wake reason is heartbeat_timer (unscoped timer wake starts fresh)"; - } - return null; -} - /** * Failure signatures from sandbox→host git workspace reconciliation. These * describe the state of the SHARED workspace (divergent histories written by @@ -2982,51 +2255,12 @@ function mergeHotRestartAdoptionResultJson( }; } -function truncateDisplayId(value: string | null | undefined, max = 128) { - if (!value) return null; - return value.length > max ? value.slice(0, max) : value; -} - function normalizeAgentNameKey(value: string | null | undefined) { if (typeof value !== "string") return null; const normalized = value.trim().toLowerCase(); return normalized.length > 0 ? normalized : null; } -const defaultSessionCodec: AdapterSessionCodec = { - deserialize(raw: unknown) { - const asObj = parseObject(raw); - if (Object.keys(asObj).length > 0) return asObj; - const sessionId = readNonEmptyString( - (raw as Record | null)?.sessionId, - ); - if (sessionId) return { sessionId }; - return null; - }, - serialize(params: Record | null) { - if (!params || Object.keys(params).length === 0) return null; - return params; - }, - getDisplayId(params: Record | null) { - return readNonEmptyString(params?.sessionId); - }, -}; - -function getAdapterSessionCodec(adapterType: string) { - const adapter = getServerAdapter(adapterType); - return adapter.sessionCodec ?? defaultSessionCodec; -} - -export function normalizeSessionParams( - params: Record | null | undefined, -) { - if (!params) return null; - return Object.keys(params).length > 0 ? params : null; -} - -type RunSessionOutcome = - "succeeded" | "interrupted" | "failed" | "cancelled" | "timed_out"; - type SkillTestHeartbeatCompletion = { outcome: "failed" | "cancelled"; error: string | null; @@ -3061,196 +2295,6 @@ export function resolveSkillTestRunCompletionForHeartbeatOutcome( return null; } -const HERMES_ADAPTER_TYPE = "hermes_local"; -const HERMES_SESSION_ID_REGEX = - /^(?:\d{8}_\d{6}_[A-Za-z0-9_-]{4,}|[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})$/; - -function requiresCanonicalSessionIds(adapterType: string | null | undefined) { - return adapterType === HERMES_ADAPTER_TYPE; -} - -function isCanonicalSessionIdForAdapter( - adapterType: string | null | undefined, - sessionId: string | null | undefined, -) { - if (!sessionId) return false; - if (!requiresCanonicalSessionIds(adapterType)) return true; - return HERMES_SESSION_ID_REGEX.test(sessionId); -} - -function normalizeResumeParamsForAdapter( - adapterType: string | null | undefined, - params: Record | null | undefined, -) { - const normalized = normalizeSessionParams(params); - if (!normalized) return null; - if (!requiresCanonicalSessionIds(adapterType)) return normalized; - const sessionId = readNonEmptyString(normalized.sessionId); - return isCanonicalSessionIdForAdapter(adapterType, sessionId) - ? normalized - : null; -} - -export function resolveNextSessionState(input: { - adapterType?: string | null; - codec: AdapterSessionCodec; - adapterResult: AdapterExecutionResult; - outcome: RunSessionOutcome; - previousParams: Record | null; - previousDisplayId: string | null; - previousLegacySessionId: string | null; -}) { - const { - adapterType, - codec, - adapterResult, - previousParams, - previousDisplayId, - previousLegacySessionId, - } = input; - - if (adapterResult.clearSession) { - return { - params: null as Record | null, - displayId: null as string | null, - legacySessionId: null as string | null, - }; - } - - if (!requiresCanonicalSessionIds(adapterType)) { - const explicitParams = adapterResult.sessionParams; - const hasExplicitParams = adapterResult.sessionParams !== undefined; - const hasExplicitSessionId = adapterResult.sessionId !== undefined; - const explicitSessionId = readNonEmptyString(adapterResult.sessionId); - const hasExplicitDisplay = adapterResult.sessionDisplayId !== undefined; - const explicitDisplayId = readNonEmptyString( - adapterResult.sessionDisplayId, - ); - const shouldUsePrevious = - !hasExplicitParams && !hasExplicitSessionId && !hasExplicitDisplay; - - const candidateParams = hasExplicitParams - ? explicitParams - : hasExplicitSessionId - ? explicitSessionId - ? { sessionId: explicitSessionId } - : null - : previousParams; - - const serialized = normalizeSessionParams( - codec.serialize(normalizeSessionParams(candidateParams) ?? null), - ); - const deserialized = normalizeSessionParams(codec.deserialize(serialized)); - - const displayId = truncateDisplayId( - explicitDisplayId ?? - (codec.getDisplayId ? codec.getDisplayId(deserialized) : null) ?? - readNonEmptyString(deserialized?.sessionId) ?? - (shouldUsePrevious ? previousDisplayId : null) ?? - explicitSessionId ?? - (shouldUsePrevious ? previousLegacySessionId : null), - ); - - const legacySessionId = - explicitSessionId ?? - readNonEmptyString(deserialized?.sessionId) ?? - displayId ?? - (shouldUsePrevious ? previousLegacySessionId : null); - - return { - params: serialized, - displayId, - legacySessionId, - }; - } - - const previousSerializedParams = normalizeResumeParamsForAdapter( - adapterType, - codec.serialize( - normalizeResumeParamsForAdapter(adapterType, previousParams), - ), - ); - const validPreviousDisplayId = isCanonicalSessionIdForAdapter( - adapterType, - previousDisplayId, - ) - ? previousDisplayId - : null; - const validPreviousLegacySessionId = isCanonicalSessionIdForAdapter( - adapterType, - previousLegacySessionId, - ) - ? previousLegacySessionId - : null; - const previousState = () => { - const displayId = truncateDisplayId( - readNonEmptyString(previousSerializedParams?.sessionId) ?? - validPreviousDisplayId ?? - validPreviousLegacySessionId, - ); - return { - params: previousSerializedParams, - displayId, - legacySessionId: - readNonEmptyString(previousSerializedParams?.sessionId) ?? - displayId ?? - validPreviousLegacySessionId, - }; - }; - - if (input.outcome !== "succeeded") { - return previousState(); - } - - const explicitParams = adapterResult.sessionParams; - const hasExplicitParams = adapterResult.sessionParams !== undefined; - const explicitSessionId = readNonEmptyString(adapterResult.sessionId); - const validExplicitSessionId = isCanonicalSessionIdForAdapter( - adapterType, - explicitSessionId, - ) - ? explicitSessionId - : null; - const explicitDisplayId = readNonEmptyString(adapterResult.sessionDisplayId); - const validExplicitDisplayId = isCanonicalSessionIdForAdapter( - adapterType, - explicitDisplayId, - ) - ? explicitDisplayId - : null; - const explicitSerializedParams = hasExplicitParams - ? normalizeResumeParamsForAdapter( - adapterType, - codec.serialize(normalizeSessionParams(explicitParams) ?? null), - ) - : null; - const explicitCanonicalSessionId = - readNonEmptyString(explicitSerializedParams?.sessionId) ?? - validExplicitSessionId ?? - validExplicitDisplayId; - - if (!explicitCanonicalSessionId) { - return previousState(); - } - - const serialized = normalizeResumeParamsForAdapter( - adapterType, - codec.serialize({ sessionId: explicitCanonicalSessionId }), - ); - const displayId = truncateDisplayId( - readNonEmptyString(serialized?.sessionId) ?? - (codec.getDisplayId ? codec.getDisplayId(serialized) : null) ?? - explicitCanonicalSessionId, - ); - const legacySessionId = - readNonEmptyString(serialized?.sessionId) ?? explicitCanonicalSessionId; - - return { - params: serialized, - displayId, - legacySessionId, - }; -} export type HeartbeatEnvironmentRuntime = ReturnType< typeof environmentRuntimeService @@ -3446,6 +2490,30 @@ export function heartbeatService( db: Db, options: HeartbeatServiceOptions = {}, ) { + const { + getAgent, + resolveSessionBeforeForWakeup, + getRun, + ensureRuntimeState, + getTaskSession, + getLatestAgentConfigRevision, + evaluateSessionCompaction, + upsertTaskSession, + resolveNormalizedUsageForSession, + clearTaskSessions, + resolveExplicitResumeSessionOverride, + hasResolvablePriorSessionWorkspaceForWake, + getRunLogAccess, + listRuns, + getRuntimeStateWithSessions, + listTaskSessions, + resetRuntimeSession, + listEvents, + getRetryExhaustedReason, + getRunIssueSummary, + getActiveRunForAgent, + getActiveRunIssueSummaryForAgent, + } = createHeartbeatRunState(db); const { toAgentOrgRow, listCompanyAgentOrgRows, @@ -4077,8 +3145,6 @@ export function heartbeatService( } const taskWatchdogs = taskWatchdogService(db, { enqueueWakeup }); - let unsafeTextProjectionPromise: Promise | null = null; - async function completeSkillTestRunForHeartbeatOutcome(input: { run: typeof heartbeatRuns.$inferSelect; issueId: string | null; @@ -4578,66 +3644,12 @@ export function heartbeatService( } } - async function hasUnsafeTextProjectionDatabase() { - if (!unsafeTextProjectionPromise) { - unsafeTextProjectionPromise = db - .execute( - sql`select current_setting('server_encoding') as server_encoding`, - ) - .then((rows) => { - const first = Array.isArray(rows) ? rows[0] : null; - const serverEncoding = - typeof first === "object" && first !== null - ? (first as Record).server_encoding - : null; - return ( - typeof serverEncoding === "string" && - serverEncoding.toUpperCase() === "SQL_ASCII" - ); - }) - .catch((err) => { - logger.warn( - { err }, - "failed to inspect database server encoding; using conservative heartbeat result projection", - ); - return true; - }); - } - return unsafeTextProjectionPromise; - } - - async function getAgent(agentId: string) { - return db - .select() - .from(agents) - .where(eq(agents.id, agentId)) - .then((rows) => rows[0] ?? null); - } - async function getAgentInvokability( agent: typeof agents.$inferSelect | null | undefined, ) { return evaluateAgentInvokabilityFromDb(db, agent); } - async function getRun( - runId: string, - opts?: { unsafeFullResultJson?: boolean; includeExecutionEvidence?: boolean }, - ) { - const safeForLegacyEncoding = - !opts?.unsafeFullResultJson && (await hasUnsafeTextProjectionDatabase()); - const columns = opts?.unsafeFullResultJson - ? getTableColumns(heartbeatRuns) - : safeForLegacyEncoding ? heartbeatRunSqlAsciiSafeColumns : heartbeatRunSafeColumns; - return db - .select(opts?.includeExecutionEvidence - ? { ...columns, resultJson: heartbeatRunExecutionEvidenceColumn } - : columns) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .then((rows) => rows[0] ?? null); - } - async function recordCurrentHeartbeatRunRuntimeProgress( run: Pick< typeof heartbeatRuns.$inferSelect, @@ -4660,91 +3672,6 @@ export function heartbeatService( return recordHeartbeatRunRuntimeProgress(currentRun, update, issueId); } - async function getRunLogAccess(runId: string) { - return db - .select(heartbeatRunLogAccessColumns) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .then((rows) => rows[0] ?? null); - } - - async function getRuntimeState(agentId: string) { - return db - .select() - .from(agentRuntimeState) - .where(eq(agentRuntimeState.agentId, agentId)) - .then((rows) => rows[0] ?? null); - } - - async function getLatestAgentConfigRevision( - companyId: string, - agentId: string, - ) { - return db - .select({ - id: agentConfigRevisions.id, - changedKeys: agentConfigRevisions.changedKeys, - createdAt: agentConfigRevisions.createdAt, - }) - .from(agentConfigRevisions) - .where( - and( - eq(agentConfigRevisions.companyId, companyId), - eq(agentConfigRevisions.agentId, agentId), - ), - ) - .orderBy( - desc(agentConfigRevisions.createdAt), - desc(agentConfigRevisions.id), - ) - .limit(1) - .then((rows) => rows[0] ?? null); - } - - async function getTaskSession( - companyId: string, - agentId: string, - adapterType: string, - taskKey: string, - ) { - return db - .select() - .from(agentTaskSessions) - .where( - and( - eq(agentTaskSessions.companyId, companyId), - eq(agentTaskSessions.agentId, agentId), - eq(agentTaskSessions.adapterType, adapterType), - eq(agentTaskSessions.taskKey, taskKey), - ), - ) - .then((rows) => rows[0] ?? null); - } - - async function getLatestRunForSession( - agentId: string, - sessionId: string, - opts?: { excludeRunId?: string | null }, - ) { - const conditions = [ - eq(heartbeatRuns.agentId, agentId), - eq(heartbeatRuns.sessionIdAfter, sessionId), - ]; - if (opts?.excludeRunId) { - conditions.push(sql`${heartbeatRuns.id} <> ${opts.excludeRunId}`); - } - return db - .select({ - id: heartbeatRuns.id, - usageJson: heartbeatRuns.usageJson, - }) - .from(heartbeatRuns) - .where(and(...conditions)) - .orderBy(desc(heartbeatRuns.createdAt)) - .limit(1) - .then((rows) => rows[0] ?? null); - } - const issueMonitorDispatchColumns = { id: issues.id, companyId: issues.companyId, @@ -5611,470 +4538,11 @@ export function heartbeatService( }; } - async function getOldestRunForSession(agentId: string, sessionId: string) { - return db - .select({ - id: heartbeatRuns.id, - createdAt: heartbeatRuns.createdAt, - }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.agentId, agentId), - eq(heartbeatRuns.sessionIdAfter, sessionId), - ), - ) - .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) - .limit(1) - .then((rows) => rows[0] ?? null); - } - - async function resolveNormalizedUsageForSession(input: { - agentId: string; - runId: string; - sessionId: string | null; - rawUsage: UsageTotals | null; - usageBasis?: "per_run" | "session_cumulative" | null; - }) { - const { agentId, runId, sessionId, rawUsage, usageBasis } = input; - // Adapters that declare per-run usage (e.g. the ACPX lane reports each - // turn's tokens, not session totals) must not be session-delta'd, or - // consecutive runs would be undercounted. - if (!sessionId || !rawUsage || usageBasis !== "session_cumulative") { - return { - normalizedUsage: rawUsage, - previousRawUsage: null as UsageTotals | null, - derivedFromSessionTotals: false, - }; - } - - const previousRun = await getLatestRunForSession(agentId, sessionId, { - excludeRunId: runId, - }); - const previousRawUsage = readRawUsageTotals(previousRun?.usageJson); - return { - normalizedUsage: normalizeAdapterRunUsage(rawUsage, previousRawUsage, usageBasis), - previousRawUsage, - derivedFromSessionTotals: previousRawUsage !== null, - }; - } - - async function evaluateSessionCompaction(input: { - agent: typeof agents.$inferSelect; - sessionId: string | null; - issueId: string | null; - continuationSummaryBody?: string | null; - }): Promise { - const { agent, sessionId, issueId } = input; - if (!sessionId) { - return { - rotate: false, - reason: null, - handoffMarkdown: null, - previousRunId: null, - }; - } - - const policy = parseSessionCompactionPolicy(agent); - if (!policy.enabled || !hasSessionCompactionThresholds(policy)) { - return { - rotate: false, - reason: null, - handoffMarkdown: null, - previousRunId: null, - }; - } - - const fetchLimit = Math.max( - policy.maxSessionRuns > 0 ? policy.maxSessionRuns + 1 : 0, - 4, - ); - const runs = await db - .select({ - id: heartbeatRuns.id, - createdAt: heartbeatRuns.createdAt, - usageJson: heartbeatRuns.usageJson, - error: heartbeatRuns.error, - terminalFailureCategory: sql`case - when jsonb_typeof(${heartbeatRuns.resultJson} -> 'terminalSessionFailure') = 'object' - then coalesce(left(${heartbeatRuns.resultJson} #>> '{terminalSessionFailure,category}', 32), 'unknown') end`, - ...heartbeatRunListResultColumns, - }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.agentId, agent.id), - eq(heartbeatRuns.sessionIdAfter, sessionId), - ), - ) - .orderBy(desc(heartbeatRuns.createdAt)) - .limit(fetchLimit); - - if (runs.length === 0) { - return { - rotate: false, - reason: null, - handoffMarkdown: null, - previousRunId: null, - }; - } - - const latestRun = runs[0] ?? null; - const oldestRun = - policy.maxSessionAgeHours > 0 - ? await getOldestRunForSession(agent.id, sessionId) - : (runs[runs.length - 1] ?? latestRun); - const latestRawUsage = readRawUsageTotals(latestRun?.usageJson); - // Historical Codex/Gemini raw input includes cache reads. Only add the - // separate cache counter when the writer explicitly saved exclusive input. - const latestRawInputTokens = (latestRawUsage?.inputTokens ?? 0) + - (latestRun?.usageJson?.rawInputIncludesCached === false ? latestRawUsage?.cachedInputTokens ?? 0 : 0); - const sessionAgeHours = - latestRun && oldestRun - ? Math.max( - 0, - (new Date(latestRun.createdAt).getTime() - - new Date(oldestRun.createdAt).getTime()) / - (1000 * 60 * 60), - ) - : 0; - - let reason: string | null = null; - if (policy.maxSessionRuns > 0 && runs.length > policy.maxSessionRuns) { - reason = `session exceeded ${policy.maxSessionRuns} runs`; - } else if ( - policy.maxRawInputTokens > 0 && - latestRawUsage && - latestRawInputTokens >= policy.maxRawInputTokens - ) { - reason = - `session raw input reached ${formatCount(latestRawInputTokens)} tokens ` + - `(threshold ${formatCount(policy.maxRawInputTokens)})`; - } else if ( - policy.maxSessionAgeHours > 0 && - sessionAgeHours >= policy.maxSessionAgeHours - ) { - reason = `session age reached ${Math.floor(sessionAgeHours)} hours`; - } - - if (!reason || !latestRun) { - return { - rotate: false, - reason: null, - handoffMarkdown: null, - previousRunId: latestRun?.id ?? null, - }; - } - - const latestSummary = summarizeHeartbeatRunListResultJson({ - summary: latestRun?.resultSummary, - result: latestRun?.resultResult, - message: latestRun?.resultMessage, - error: latestRun?.resultError, - totalCostUsd: latestRun?.resultTotalCostUsd, - costUsd: latestRun?.resultCostUsd, - costUsdCamel: latestRun?.resultCostUsdCamel, - }); - const latestTextSummary = - readNonEmptyString(latestSummary?.summary) ?? - readNonEmptyString(latestSummary?.result) ?? - readNonEmptyString(latestSummary?.message) ?? - readNonEmptyString(summarizeRunErrorForModel(latestRun.error, latestRun.terminalFailureCategory)); - - const handoffMarkdown = [ - "Paperclip session handoff:", - `- Previous session: ${sessionId}`, - issueId ? `- Issue: ${issueId}` : "", - `- Rotation reason: ${reason}`, - latestTextSummary ? `- Last run summary: ${latestTextSummary}` : "", - input.continuationSummaryBody - ? `- Issue continuation summary: ${input.continuationSummaryBody.slice(0, 1_500)}` - : "", - "Continue from the current task state. Rebuild only the minimum context you need.", - ] - .filter(Boolean) - .join("\n"); - - return { - rotate: true, - reason, - handoffMarkdown, - previousRunId: latestRun.id, - }; - } - - async function resolveSessionBeforeForWakeup( - agent: typeof agents.$inferSelect, - taskKey: string | null, - ) { - if (taskKey) { - const codec = getAdapterSessionCodec(agent.adapterType); - const existingTaskSession = await getTaskSession( - agent.companyId, - agent.id, - agent.adapterType, - taskKey, - ); - const parsedParams = normalizeSessionParams( - codec.deserialize(existingTaskSession?.sessionParamsJson ?? null), - ); - return truncateDisplayId( - existingTaskSession?.sessionDisplayId ?? - (codec.getDisplayId ? codec.getDisplayId(parsedParams) : null) ?? - readNonEmptyString(parsedParams?.sessionId), - ); - } - - const runtimeForRun = await getRuntimeState(agent.id); - return runtimeForRun?.sessionId ?? null; - } - - async function hasResolvableSessionWorkspaceCwd( - sessionParams: Record | null | undefined, - ) { - const cwd = readNonEmptyString(sessionParams?.cwd); - if (!cwd || isUnsafeSessionWorkspaceCwd(cwd)) return false; - return fs - .stat(cwd) - .then((stats) => stats.isDirectory()) - .catch(() => false); - } - - async function hasResolvablePriorSessionWorkspaceForWake(input: { - agent: typeof agents.$inferSelect; - contextSnapshot: Record; - taskKey: string | null; - explicitResumeSession: Awaited< - ReturnType - > | null; - }) { - if ( - await hasResolvableSessionWorkspaceCwd( - input.explicitResumeSession?.sessionParams, - ) - ) - return true; - if (shouldResetTaskSessionForWake(input.contextSnapshot)) return false; - if (!input.taskKey) return false; - - const codec = getAdapterSessionCodec(input.agent.adapterType); - const taskSession = await getTaskSession( - input.agent.companyId, - input.agent.id, - input.agent.adapterType, - input.taskKey, - ); - const taskSessionParams = normalizeResumeParamsForAdapter( - input.agent.adapterType, - codec.deserialize(taskSession?.sessionParamsJson ?? null), - ); - return hasResolvableSessionWorkspaceCwd(taskSessionParams); - } - - async function resolveExplicitResumeSessionOverride( - agent: typeof agents.$inferSelect, - payload: Record | null, - taskKey: string | null, - ) { - const resumeFromRunId = readNonEmptyString(payload?.resumeFromRunId); - if (!resumeFromRunId) return null; - - const resumeRun = await db - .select({ - id: heartbeatRuns.id, - contextSnapshot: heartbeatRuns.contextSnapshot, - resultJson: heartbeatRuns.resultJson, - sessionIdBefore: heartbeatRuns.sessionIdBefore, - sessionIdAfter: heartbeatRuns.sessionIdAfter, - }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.id, resumeFromRunId), - eq(heartbeatRuns.companyId, agent.companyId), - eq(heartbeatRuns.agentId, agent.id), - ), - ) - .then((rows) => rows[0] ?? null); - if (!resumeRun) return null; - - const resumeContext = parseObject(resumeRun.contextSnapshot); - const resumeTaskKey = deriveTaskKey(resumeContext, null) ?? taskKey; - const resumeTaskSession = resumeTaskKey - ? await getTaskSession( - agent.companyId, - agent.id, - agent.adapterType, - resumeTaskKey, - ) - : null; - const sessionCodec = getAdapterSessionCodec(agent.adapterType); - const resumeRunResult = parseObject(resumeRun.resultJson); - const resumeRunSessionId = requiresCanonicalSessionIds(agent.adapterType) - ? (readNonEmptyString(resumeRunResult.sessionId) ?? - readNonEmptyString(resumeRunResult.session_id)) - : null; - const sessionOverride = buildExplicitResumeSessionOverride({ - adapterType: agent.adapterType, - resumeFromRunId, - resumeRunSessionIdBefore: resumeRun.sessionIdBefore, - resumeRunSessionIdAfter: resumeRun.sessionIdAfter, - resumeRunSessionParams: resumeRunSessionId - ? { sessionId: resumeRunSessionId } - : null, - taskSession: resumeTaskSession, - sessionCodec, - }); - if (!sessionOverride) return null; - - return { - resumeFromRunId, - taskKey: resumeTaskKey, - issueId: readNonEmptyString(resumeContext.issueId), - taskId: - readNonEmptyString(resumeContext.taskId) ?? - readNonEmptyString(resumeContext.issueId), - sessionDisplayId: sessionOverride.sessionDisplayId, - sessionParams: sessionOverride.sessionParams, - }; - } - const { resolveReusedGitWorkspaceAnchor, resolveWorkspaceForRun, } = createHeartbeatWorkspaceResolver(db); - async function upsertTaskSession(input: { - companyId: string; - agentId: string; - adapterType: string; - taskKey: string; - sessionParamsJson: Record | null; - sessionDisplayId: string | null; - lastRunId: string | null; - lastError: string | null; - }) { - return db.transaction(async (tx) => { - const [issue] = await tx.select().from(issues).where(and(sql`${issues.id}::text = ${input.taskKey}`, eq(issues.companyId, input.companyId))).for("update"); - if (isConversation(issue)) { - const [run] = input.lastRunId ? await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.lastRunId)) : []; - if (run?.status === "cancelled" || run?.contextSnapshot?.conversationSessionGeneration !== issue.conversationSessionGeneration) return null; - } - const existing = await tx.select().from(agentTaskSessions).where(and(eq(agentTaskSessions.companyId, input.companyId), eq(agentTaskSessions.agentId, input.agentId), eq(agentTaskSessions.adapterType, input.adapterType), eq(agentTaskSessions.taskKey, input.taskKey))).then((rows) => rows[0] ?? null); - if (existing) { - return tx - .update(agentTaskSessions) - .set({ - sessionParamsJson: input.sessionParamsJson, - sessionDisplayId: input.sessionDisplayId, - lastRunId: input.lastRunId, - lastError: input.lastError, - updatedAt: new Date(), - }) - .where(eq(agentTaskSessions.id, existing.id)) - .returning() - .then((rows) => rows[0] ?? null); - } - - return tx - .insert(agentTaskSessions) - .values({ - companyId: input.companyId, - agentId: input.agentId, - adapterType: input.adapterType, - taskKey: input.taskKey, - sessionParamsJson: input.sessionParamsJson, - sessionDisplayId: input.sessionDisplayId, - lastRunId: input.lastRunId, - lastError: input.lastError, - }) - .returning() - .then((rows) => rows[0] ?? null); - }); - } - - async function clearTaskSessions( - companyId: string, - agentId: string, - opts?: { - taskKey?: string | null; - adapterType?: string | null; - expectedRunId?: string; - includeIssueAliases?: boolean; - }, - ) { - const conditions = [ - eq(agentTaskSessions.companyId, companyId), - eq(agentTaskSessions.agentId, agentId), - ]; - if (opts?.taskKey) { - const exactTaskKey = eq(agentTaskSessions.taskKey, opts.taskKey); - if (opts.includeIssueAliases) { - const selectedIssue = isUuidLike(opts.taskKey) - ? eq(issues.id, opts.taskKey) - : eq(issues.identifier, opts.taskKey.toUpperCase()); - // Operator task resets accept the UUID sent by run detail and the - // identifier used by some saved sessions. Resolve only from the current - // same-company issue row, in this DELETE's snapshot; arbitrary custom - // keys retain exact-match behavior and run/model context grants no alias. - conditions.push( - or( - exactTaskKey, - sql`exists ( - select 1 from ${issues} - where ${issues.companyId} = ${companyId} - and ${selectedIssue} - and (${agentTaskSessions.taskKey} = ${issues.id}::text - or ${agentTaskSessions.taskKey} = ${issues.identifier}) - )`, - )!, - ); - } else { - conditions.push(exactTaskKey); - } - } - if (opts?.adapterType) { - conditions.push(eq(agentTaskSessions.adapterType, opts.adapterType)); - } - - return db.transaction(async (tx) => { - if (opts?.taskKey && opts.expectedRunId) { - const [issue] = await tx.select().from(issues).where(sql`${issues.id}::text = ${opts.taskKey}`).for("update"); - if (isConversation(issue)) { - const [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, opts.expectedRunId)); - if (run?.status === "cancelled" || run?.contextSnapshot?.conversationSessionGeneration !== issue.conversationSessionGeneration) return 0; - } - } - return tx.delete(agentTaskSessions).where(and(...conditions)).returning().then((rows) => rows.length); - }); - } - - async function ensureRuntimeState(agent: typeof agents.$inferSelect) { - const existing = await getRuntimeState(agent.id); - if (existing) return existing; - - const inserted = await db - .insert(agentRuntimeState) - .values({ - agentId: agent.id, - companyId: agent.companyId, - adapterType: agent.adapterType, - stateJson: {}, - }) - .onConflictDoNothing({ - target: agentRuntimeState.agentId, - }) - .returning() - .then((rows) => rows[0] ?? null); - if (inserted) return inserted; - - const ensured = await getRuntimeState(agent.id); - if (!ensured) { - throw new Error(`Failed to ensure runtime state for agent ${agent.id}`); - } - return ensured; - } - // Emits agent.task_run for a run write that just reached a terminal // status, unless the write only re-set a status the run already had (a // status-preserving patch, such as a livenessReason update on a run that @@ -23534,102 +22002,7 @@ export function heartbeatService( await new Promise((resolve) => setTimeout(resolve, intervalMs)); } }, - list: async ( - companyId: string, - agentId?: string, - limit?: number, - options: { summary?: boolean } = {}, - ) => { - const safeForLegacyEncoding = await hasUnsafeTextProjectionDatabase(); - const summary = options.summary === true; - const rows = await retryIdempotentDatabaseOperation(async () => { - const query = db - .select( - summary - ? { - ...heartbeatRunSummaryListColumns, - ...heartbeatRunListContextColumns, - } - : safeForLegacyEncoding - ? { - ...heartbeatRunListColumns, - error: sql`NULL`.as("error"), - ...heartbeatRunListContextColumns, - } - : { - ...heartbeatRunListColumns, - ...heartbeatRunListContextColumns, - ...heartbeatRunListResultColumns, - }, - ) - .from(heartbeatRuns) - .where( - agentId - ? and( - eq(heartbeatRuns.companyId, companyId), - eq(heartbeatRuns.agentId, agentId), - ) - : eq(heartbeatRuns.companyId, companyId), - ) - .orderBy(desc(heartbeatRuns.createdAt)); - - return limit ? await query.limit(limit) : await query; - }); - return rows.map((row) => { - const { - contextIssueId, - contextTaskId, - contextTaskKey, - contextCommentId, - contextWakeCommentId, - contextWakeReason, - contextWakeSource, - contextWakeTriggerDetail, - resultSummary, - resultResult, - resultMessage, - resultError, - resultTotalCostUsd, - resultCostUsd, - resultCostUsdCamel, - ...rest - } = row as typeof row & { - resultSummary?: string | null; - resultResult?: string | null; - resultMessage?: string | null; - resultError?: string | null; - resultTotalCostUsd?: string | null; - resultCostUsd?: string | null; - resultCostUsdCamel?: string | null; - }; - - return { - ...rest, - contextSnapshot: summarizeHeartbeatRunContextSnapshot({ - issueId: contextIssueId, - taskId: contextTaskId, - taskKey: contextTaskKey, - commentId: contextCommentId, - wakeCommentId: contextWakeCommentId, - wakeReason: contextWakeReason, - wakeSource: contextWakeSource, - wakeTriggerDetail: contextWakeTriggerDetail, - }), - resultJson: - safeForLegacyEncoding || summary - ? null - : summarizeHeartbeatRunListResultJson({ - summary: resultSummary, - result: resultResult, - message: resultMessage, - error: resultError, - totalCostUsd: resultTotalCostUsd, - costUsd: resultCostUsd, - costUsdCamel: resultCostUsdCamel, - }), - }; - }); - }, + list: listRuns, getRun, @@ -23639,125 +22012,15 @@ export function heartbeatService( getRunLogAccess, - getRuntimeState: async (agentId: string) => { - const state = await getRuntimeState(agentId); - const agent = await getAgent(agentId); - if (!agent) return null; - const ensured = state ?? (await ensureRuntimeState(agent)); - const latestTaskSession = await db - .select() - .from(agentTaskSessions) - .where( - and( - eq(agentTaskSessions.companyId, agent.companyId), - eq(agentTaskSessions.agentId, agent.id), - ), - ) - .orderBy(desc(agentTaskSessions.updatedAt)) - .limit(1) - .then((rows) => rows[0] ?? null); - return { - ...ensured, - sessionDisplayId: - latestTaskSession?.sessionDisplayId ?? ensured.sessionId, - sessionParamsJson: latestTaskSession?.sessionParamsJson ?? null, - }; - }, + getRuntimeState: getRuntimeStateWithSessions, - listTaskSessions: async (agentId: string) => { - const agent = await getAgent(agentId); - if (!agent) throw notFound("Agent not found"); + listTaskSessions, - return db - .select() - .from(agentTaskSessions) - .where( - and( - eq(agentTaskSessions.companyId, agent.companyId), - eq(agentTaskSessions.agentId, agentId), - ), - ) - .orderBy( - desc(agentTaskSessions.updatedAt), - desc(agentTaskSessions.createdAt), - ); - }, + resetRuntimeSession, - resetRuntimeSession: async ( - agentId: string, - opts?: { taskKey?: string | null }, - ) => { - const agent = await getAgent(agentId); - if (!agent) throw notFound("Agent not found"); - await ensureRuntimeState(agent); - const taskKey = readNonEmptyString(opts?.taskKey); - const clearedTaskSessions = await clearTaskSessions( - agent.companyId, - agent.id, - taskKey - ? { - taskKey, - adapterType: agent.adapterType, - includeIssueAliases: true, - } - : undefined, - ); - const runtimePatch: Partial = { - sessionId: null, - lastError: null, - updatedAt: new Date(), - }; - if (!taskKey) { - runtimePatch.stateJson = {}; - } + listEvents, - const updated = await db - .update(agentRuntimeState) - .set(runtimePatch) - .where(eq(agentRuntimeState.agentId, agentId)) - .returning() - .then((rows) => rows[0] ?? null); - - if (!updated) return null; - return { - ...updated, - sessionDisplayId: null, - sessionParamsJson: null, - clearedTaskSessions, - }; - }, - - listEvents: (runId: string, afterSeq = 0, limit = 200) => - db - .select() - .from(heartbeatRunEvents) - .where( - and( - eq(heartbeatRunEvents.runId, runId), - gt(heartbeatRunEvents.seq, afterSeq), - ), - ) - .orderBy(asc(heartbeatRunEvents.seq)) - .limit(Math.max(1, Math.min(limit, 1000))), - - getRetryExhaustedReason: async (runId: string) => { - const row = await db - .select({ - message: heartbeatRunEvents.message, - }) - .from(heartbeatRunEvents) - .where( - and( - eq(heartbeatRunEvents.runId, runId), - eq(heartbeatRunEvents.eventType, "lifecycle"), - sql`${heartbeatRunEvents.message} like 'Bounded retry exhausted%'`, - ), - ) - .orderBy(desc(heartbeatRunEvents.id)) - .limit(1) - .then((rows) => rows[0] ?? null); - return row?.message ?? null; - }, + getRetryExhaustedReason, readLog: async ( runOrLookup: @@ -23991,43 +22254,10 @@ export function heartbeatService( cancelBudgetScopeWork, - getRunIssueSummary: async (runId: string) => { - const [run] = await db - .select(heartbeatRunIssueSummaryColumns) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .limit(1); - return run ?? null; - }, + getRunIssueSummary, - getActiveRunForAgent: async (agentId: string) => { - const [run] = await db - .select() - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.agentId, agentId), - eq(heartbeatRuns.status, "running"), - ), - ) - .orderBy(desc(heartbeatRuns.startedAt)) - .limit(1); - return run ?? null; - }, + getActiveRunForAgent, - getActiveRunIssueSummaryForAgent: async (agentId: string) => { - const [run] = await db - .select(heartbeatRunIssueSummaryColumns) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.agentId, agentId), - eq(heartbeatRuns.status, "running"), - ), - ) - .orderBy(desc(heartbeatRuns.startedAt)) - .limit(1); - return run ?? null; - }, + getActiveRunIssueSummaryForAgent, }; } diff --git a/server/src/services/heartbeat/run-state.test.ts b/server/src/services/heartbeat/run-state.test.ts new file mode 100644 index 00000000000..c9bb5e50011 --- /dev/null +++ b/server/src/services/heartbeat/run-state.test.ts @@ -0,0 +1,222 @@ +import { randomUUID } from "node:crypto"; +import type { Db } from "@paperclipai/db"; +import { + agents, + agentTaskSessions, + companies, + createDb, + heartbeatRuns, + issues, +} from "@paperclipai/db"; +import { eq, type SQL } from "drizzle-orm"; +import { PgDialect } from "drizzle-orm/pg-core"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "../../__tests__/helpers/embedded-postgres.js"; +import * as legacy from "../heartbeat.js"; +import * as extracted from "./run-state.js"; + +// SQL_ASCII needs a separate database cluster. Inspect its selected SQL here; +// the PostgreSQL tests below exercise the session queries and transactions. +function readingDatabase(serverEncoding: string, ...results: unknown[][]) { + const scopes: unknown[][] = []; + const columns: Record[] = []; + const execute = vi.fn(async () => [{ server_encoding: serverEncoding }]); + const select = vi.fn((projection: Record = {}) => { + columns.push(projection); + const rows = results.shift() ?? []; + const query = { + from: () => query, + where: (condition: SQL) => { + scopes.push(new PgDialect().sqlToQuery(condition).params); + return query; + }, + orderBy: () => query, + limit: () => query, + then: (resolve: (rows: unknown[]) => unknown) => Promise.resolve(rows).then(resolve), + }; + return query; + }); + return { db: { execute, select } as unknown as Db, execute, select, columns, scopes }; +} + +describe("heartbeat run state module", () => { + it("preserves the legacy public helper identities", () => { + expect(legacy.resolveNextSessionState).toBe(extracted.resolveNextSessionState); + expect(legacy.buildExplicitResumeSessionOverride).toBe(extracted.buildExplicitResumeSessionOverride); + expect(legacy.normalizeAdapterRunUsage).toBe(extracted.normalizeAdapterRunUsage); + expect(legacy.resolveLedgerScopeForRun).toBe(extracted.resolveLedgerScopeForRun); + expect(legacy.summarizeHeartbeatRunListResultJson).toBe(extracted.summarizeHeartbeatRunListResultJson); + }); + + it("constructs without queries and binds session lookups to each database", async () => { + const first = readingDatabase("UTF8", [{ sessionDisplayId: "first" }]); + const second = readingDatabase("UTF8", [{ sessionDisplayId: "second" }]); + const firstState = extracted.createHeartbeatRunState(first.db); + const secondState = extracted.createHeartbeatRunState(second.db); + expect(first.select).not.toHaveBeenCalled(); + expect(first.execute).not.toHaveBeenCalled(); + expect(second.select).not.toHaveBeenCalled(); + expect(second.execute).not.toHaveBeenCalled(); + const [one, two] = await Promise.all([ + firstState.getTaskSession("company-one", "agent-one", "codex_local", "task-one"), + secondState.getTaskSession("company-two", "agent-two", "claude_local", "task-two"), + ]); + expect(one?.sessionDisplayId).toBe("first"); + expect(two?.sessionDisplayId).toBe("second"); + expect(first.scopes[0]).toEqual(["company-one", "agent-one", "codex_local", "task-one"]); + expect(second.scopes[0]).toEqual(["company-two", "agent-two", "claude_local", "task-two"]); + }); + + it("keeps encoding caches local to each factory and preserves full-result reads", async () => { + const ascii = readingDatabase("SQL_ASCII"); + const utf8 = readingDatabase("UTF8"); + const asciiState = extracted.createHeartbeatRunState(ascii.db); + const utf8State = extracted.createHeartbeatRunState(utf8.db); + await Promise.all([asciiState.getRun("ascii-one"), asciiState.getRun("ascii-two"), utf8State.getRun("utf8")]); + const projectedSql = (column: unknown) => new PgDialect().sqlToQuery((column as SQL.Aliased).sql).sql; + expect(ascii.execute).toHaveBeenCalledTimes(1); + expect(utf8.execute).toHaveBeenCalledTimes(1); + expect(projectedSql(ascii.columns[0].resultJson)).toBe("NULL"); + expect(projectedSql(utf8.columns[0].resultJson)).toContain("pg_column_size"); + await asciiState.getRun("evidence", { includeExecutionEvidence: true }); + expect(projectedSql(ascii.columns[2].resultJson)).toContain("startupPreparationSettledAt"); + const raw = readingDatabase("SQL_ASCII"); + await extracted.createHeartbeatRunState(raw.db).getRun("raw", { unsafeFullResultJson: true }); + expect(raw.execute).not.toHaveBeenCalled(); + expect(raw.columns[0].resultJson).toBe(heartbeatRuns.resultJson); + }); +}); + +const postgresSupport = await getEmbeddedPostgresTestSupport(); +const describePostgres = postgresSupport.supported ? describe : describe.skip; +if (!postgresSupport.supported) { + console.warn(`Skipping run-state PostgreSQL tests: ${postgresSupport.reason ?? "unsupported host"}`); +} + +describePostgres("heartbeat run state database wiring", () => { + let database: Awaited>; + let db: Db; + + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("heartbeat-run-state-"); + db = createDb(database.connectionString); + }, 20_000); + + afterAll(async () => { + await database?.cleanup(); + }); + + async function fixture() { + const companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Run state", issuePrefix: `S${companyId.slice(0, 8)}` }); + const [agent] = await db.insert(agents).values({ + companyId, name: "Session agent", adapterType: "codex_local", + }).returning(); + const [issue] = await db.insert(issues).values({ companyId, title: "Session task", assigneeAgentId: agent.id }).returning(); + return { companyId, agent, issue, state: extracted.createHeartbeatRunState(db) }; + } + + it("resumes only the selected agent's run and task session", async () => { + const { companyId, agent, issue, state } = await fixture(); + const foreign = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ + companyId, agentId: agent.id, invocationSource: "assignment", status: "failed", + contextSnapshot: { issueId: issue.id }, sessionIdAfter: "our-session", + }).returning(); + const [foreignRun] = await db.insert(heartbeatRuns).values({ + companyId: foreign.companyId, agentId: foreign.agent.id, invocationSource: "assignment", + contextSnapshot: { issueId: foreign.issue.id }, sessionIdAfter: "foreign-session", + }).returning(); + await state.upsertTaskSession({ + companyId, agentId: agent.id, adapterType: agent.adapterType, taskKey: issue.id, + sessionParamsJson: { sessionId: "our-session", cwd: "/saved-workspace" }, + sessionDisplayId: "our-session", lastRunId: run.id, lastError: null, + }); + expect(await state.resolveExplicitResumeSessionOverride(agent, { resumeFromRunId: foreignRun.id }, issue.id)).toBeNull(); + expect(await state.resolveExplicitResumeSessionOverride(agent, { resumeFromRunId: run.id }, issue.id)).toMatchObject({ + resumeFromRunId: run.id, issueId: issue.id, taskKey: issue.id, + sessionDisplayId: "our-session", sessionParams: { sessionId: "our-session", cwd: "/saved-workspace" }, + }); + expect(await state.resolveSessionBeforeForWakeup(agent, issue.id)).toBe("our-session"); + expect(await state.getRuntimeStateWithSessions(agent.id)).toMatchObject({ + companyId, sessionDisplayId: "our-session", sessionParamsJson: { sessionId: "our-session" }, + }); + expect((await state.listRuns(companyId)).map((row) => row.id)).toEqual([run.id]); + }); + + it("subtracts prior cumulative usage while excluding the current run", async () => { + const { companyId, agent, state } = await fixture(); + await db.insert(heartbeatRuns).values({ + companyId, agentId: agent.id, invocationSource: "assignment", sessionIdAfter: "cumulative", + createdAt: new Date("2026-01-01T00:00:00Z"), + usageJson: { rawInputTokens: 800, rawCachedInputTokens: 100, rawOutputTokens: 80 }, + }); + const [current] = await db.insert(heartbeatRuns).values({ + companyId, agentId: agent.id, invocationSource: "assignment", sessionIdAfter: "cumulative", + createdAt: new Date("2026-01-02T00:00:00Z"), + usageJson: { rawInputTokens: 1000, rawCachedInputTokens: 125, rawOutputTokens: 100 }, + }).returning(); + const rawUsage = { inputTokens: 1000, cachedInputTokens: 125, outputTokens: 100 }; + const input = { agentId: agent.id, runId: current.id, sessionId: "cumulative", rawUsage }; + expect(await state.resolveNormalizedUsageForSession({ ...input, usageBasis: "session_cumulative" })).toEqual({ + normalizedUsage: { inputTokens: 200, cachedInputTokens: 25, outputTokens: 20 }, + previousRawUsage: { inputTokens: 800, cachedInputTokens: 100, outputTokens: 80 }, + derivedFromSessionTotals: true, + }); + expect(await state.resolveNormalizedUsageForSession({ ...input, usageBasis: "per_run" })).toEqual({ + normalizedUsage: rawUsage, previousRawUsage: null, derivedFromSessionTotals: false, + }); + }); + + it("uses the selected session's usage and summary for compaction", async () => { + const { companyId, agent, issue, state } = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ + companyId, agentId: agent.id, invocationSource: "assignment", sessionIdAfter: "compact", + usageJson: { rawInputTokens: 90, rawCachedInputTokens: 20, rawInputIncludesCached: false }, + resultJson: { summary: "Continue our task", providerPrivateData: "omit this" }, + }).returning(); + const foreign = await fixture(); + await db.insert(heartbeatRuns).values({ + companyId: foreign.companyId, agentId: foreign.agent.id, invocationSource: "assignment", + sessionIdAfter: "compact", resultJson: { summary: "Foreign task" }, + }); + const decision = await state.evaluateSessionCompaction({ + agent: { ...agent, runtimeConfig: { heartbeat: { sessionCompaction: { maxRawInputTokens: 100 } } } }, + sessionId: "compact", issueId: issue.id, + }); + expect(decision).toMatchObject({ rotate: true, reason: "session raw input reached 110 tokens (threshold 100)", previousRunId: run.id }); + expect(decision.handoffMarkdown).toContain("Continue our task"); + expect(decision.handoffMarkdown).not.toContain("Foreign task"); + expect(decision.handoffMarkdown).not.toContain("omit this"); + }); + + it.each(["superseded", "cancelled"] as const)("fences %s conversation session writes and clears", async (scenario) => { + const { companyId, agent, issue, state } = await fixture(); + await db.update(issues).set({ + conversationAgentId: agent.id, conversationUserId: "test-user", conversationState: "active", + conversationSessionGeneration: 2, + }).where(eq(issues.id, issue.id)); + const [run] = await db.insert(heartbeatRuns).values({ + companyId, agentId: agent.id, invocationSource: "assignment", status: "succeeded", + contextSnapshot: { conversationSessionGeneration: 2 }, + }).returning(); + const input = { + companyId, agentId: agent.id, adapterType: agent.adapterType, taskKey: issue.id, + sessionParamsJson: { sessionId: "retained" }, sessionDisplayId: "retained", lastRunId: run.id, lastError: null, + }; + const saved = await state.upsertTaskSession(input); + expect(saved?.sessionDisplayId).toBe("retained"); + if (scenario === "superseded") { + await db.update(issues).set({ conversationSessionGeneration: 3 }).where(eq(issues.id, issue.id)); + } else { + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, run.id)); + } + expect(await state.upsertTaskSession({ ...input, sessionDisplayId: "stale" })).toBeNull(); + expect(await state.clearTaskSessions(companyId, agent.id, { taskKey: issue.id, expectedRunId: run.id })).toBe(0); + const [retained] = await db.select().from(agentTaskSessions).where(eq(agentTaskSessions.id, saved!.id)); + expect(retained.sessionDisplayId).toBe("retained"); + }); +}); diff --git a/server/src/services/heartbeat/run-state.ts b/server/src/services/heartbeat/run-state.ts new file mode 100644 index 00000000000..75961481cad --- /dev/null +++ b/server/src/services/heartbeat/run-state.ts @@ -0,0 +1,1923 @@ +import { deriveTaskKey } from "./run-preparation.js"; +import { retryIdempotentDatabaseOperation } from "../../database-retry.js"; +import { isConversation } from "../agent-conversations.js"; +import fs from "node:fs/promises"; +import { + and, + asc, + desc, + eq, + getTableColumns, + gt, + or, + sql, +} from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { + type BillingType, + type CostStatus, +} from "@paperclipai/shared"; +import { + agents, + agentConfigRevisions, + agentRuntimeState, + agentTaskSessions, + heartbeatRunEvents, + heartbeatRuns, + issues, +} from "@paperclipai/db"; +import { notFound } from "../../errors.js"; +import { logger } from "../../middleware/logger.js"; +import { getServerAdapter } from "../../adapters/index.js"; +import type { + AdapterExecutionResult, + AdapterSessionCodec, + UsageSummary, +} from "../../adapters/index.js"; +import { + parseObject, + asNumber, +} from "../../adapters/utils.js"; +import { + HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS, + HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS, + HEARTBEAT_RUN_SAFE_RESULT_JSON_MAX_BYTES, + summarizeRunErrorForModel, +} from "../heartbeat-run-summary.js"; +import { + hasSessionCompactionThresholds, + resolveSessionCompactionPolicy, + type SessionCompactionPolicy, +} from "@paperclipai/adapter-utils"; +import { isUuidLike } from "@paperclipai/shared"; +import { isUnsafeSessionWorkspaceCwd } from "../session-workspace-cwd.js"; + +export const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON = + "execution_review_participant_recovery"; + +const heartbeatRunProcessGroupIdColumn = + heartbeatRuns.processGroupId ?? sql`NULL`.as("processGroupId"); + +const heartbeatRunListColumns = { + id: heartbeatRuns.id, + responsibleUserId: heartbeatRuns.responsibleUserId, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + scopeKind: heartbeatRuns.scopeKind, + issueId: heartbeatRuns.issueId, + invocationSource: heartbeatRuns.invocationSource, + triggerDetail: heartbeatRuns.triggerDetail, + status: heartbeatRuns.status, + inputTokens: sql`(${heartbeatRuns.usageJson} ->> 'inputTokens')::numeric`.as("inputTokens"), + cachedInputTokens: sql`(${heartbeatRuns.usageJson} ->> 'cachedInputTokens')::numeric`.as("cachedInputTokens"), + outputTokens: sql`(${heartbeatRuns.usageJson} ->> 'outputTokens')::numeric`.as("outputTokens"), + totalTokens: sql`(${heartbeatRuns.usageJson} ->> 'totalTokens')::numeric`.as("totalTokens"), + costUsd: sql`coalesce( + (${heartbeatRuns.resultJson} ->> 'costUsd')::numeric, + (${heartbeatRuns.resultJson} ->> 'cost_usd')::numeric, + (${heartbeatRuns.resultJson} ->> 'total_cost_usd')::numeric + )`.as("costUsd"), + startedAt: heartbeatRuns.startedAt, + finishedAt: heartbeatRuns.finishedAt, + error: heartbeatRuns.error, + wakeupRequestId: heartbeatRuns.wakeupRequestId, + exitCode: heartbeatRuns.exitCode, + signal: heartbeatRuns.signal, + usageJson: heartbeatRuns.usageJson, + sessionIdBefore: heartbeatRuns.sessionIdBefore, + sessionIdAfter: heartbeatRuns.sessionIdAfter, + logStore: heartbeatRuns.logStore, + logRef: heartbeatRuns.logRef, + logBytes: heartbeatRuns.logBytes, + logSha256: heartbeatRuns.logSha256, + logCompressed: heartbeatRuns.logCompressed, + stdoutExcerpt: sql`NULL`.as("stdoutExcerpt"), + stderrExcerpt: sql`NULL`.as("stderrExcerpt"), + errorCode: heartbeatRuns.errorCode, + externalRunId: heartbeatRuns.externalRunId, + processPid: heartbeatRuns.processPid, + processGroupId: heartbeatRunProcessGroupIdColumn, + processStartedAt: heartbeatRuns.processStartedAt, + lastOutputAt: heartbeatRuns.lastOutputAt, + lastOutputSeq: heartbeatRuns.lastOutputSeq, + lastOutputStream: heartbeatRuns.lastOutputStream, + lastOutputBytes: heartbeatRuns.lastOutputBytes, + retryOfRunId: heartbeatRuns.retryOfRunId, + processLossRetryCount: heartbeatRuns.processLossRetryCount, + scheduledRetryAt: heartbeatRuns.scheduledRetryAt, + scheduledRetryAttempt: heartbeatRuns.scheduledRetryAttempt, + scheduledRetryReason: heartbeatRuns.scheduledRetryReason, + livenessState: heartbeatRuns.livenessState, + livenessReason: heartbeatRuns.livenessReason, + continuationAttempt: heartbeatRuns.continuationAttempt, + lastUsefulActionAt: heartbeatRuns.lastUsefulActionAt, + nextAction: heartbeatRuns.nextAction, + createdAt: heartbeatRuns.createdAt, + updatedAt: heartbeatRuns.updatedAt, +} as const; + +const heartbeatRunSummaryListColumns = { + ...heartbeatRunListColumns, + usageJson: sql | null>`NULL`.as("usageJson"), + sessionIdBefore: sql`NULL`.as("sessionIdBefore"), + sessionIdAfter: sql`NULL`.as("sessionIdAfter"), + logStore: sql`NULL`.as("logStore"), + logRef: sql`NULL`.as("logRef"), + logSha256: sql`NULL`.as("logSha256"), + externalRunId: sql`NULL`.as("externalRunId"), + processPid: sql`NULL`.as("processPid"), + processGroupId: sql`NULL`.as("processGroupId"), + resultJson: sql | null>`NULL`.as("resultJson"), +} as const; + +const heartbeatRunListContextColumns = { + contextIssueId: sql`coalesce( + ${heartbeatRuns.issueId}::text, + ${heartbeatRuns.contextSnapshot} ->> 'issueId' + )`.as("contextIssueId"), + contextTaskId: sql`${heartbeatRuns.contextSnapshot} ->> 'taskId'`.as("contextTaskId"), + contextTaskKey: sql`${heartbeatRuns.contextSnapshot} ->> 'taskKey'`.as("contextTaskKey"), + contextCommentId: sql`${heartbeatRuns.contextSnapshot} ->> 'commentId'`.as("contextCommentId"), + contextWakeCommentId: sql`${heartbeatRuns.contextSnapshot} ->> 'wakeCommentId'`.as("contextWakeCommentId"), + contextWakeReason: sql`${heartbeatRuns.contextSnapshot} ->> 'wakeReason'`.as("contextWakeReason"), + contextWakeSource: sql`${heartbeatRuns.contextSnapshot} ->> 'wakeSource'`.as("contextWakeSource"), + contextWakeTriggerDetail: sql`${heartbeatRuns.contextSnapshot} ->> 'wakeTriggerDetail'`.as("contextWakeTriggerDetail"), +} as const; + +const heartbeatRunListResultColumns = { + resultSummary: sql< + string | null + >`left(${heartbeatRuns.resultJson} ->> 'summary', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS})`.as( + "resultSummary", + ), + resultResult: sql< + string | null + >`left(${heartbeatRuns.resultJson} ->> 'result', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS})`.as( + "resultResult", + ), + resultMessage: sql< + string | null + >`left(${heartbeatRuns.resultJson} ->> 'message', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS})`.as( + "resultMessage", + ), + resultError: sql< + string | null + >`left(${heartbeatRuns.resultJson} ->> 'error', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS})`.as( + "resultError", + ), + resultTotalCostUsd: sql< + string | null + >`${heartbeatRuns.resultJson} ->> 'total_cost_usd'`.as("resultTotalCostUsd"), + resultCostUsd: sql< + string | null + >`${heartbeatRuns.resultJson} ->> 'cost_usd'`.as("resultCostUsd"), + resultCostUsdCamel: sql< + string | null + >`${heartbeatRuns.resultJson} ->> 'costUsd'`.as("resultCostUsdCamel"), +} as const; + +// Reserve at most 9 KiB for diagnostics in the reduced result. An oversized +// multibyte field uses a conservative four-byte-per-character prefix, with a +// visible pointer to the full (adapter-bounded) run error and transcript. +const diagnosticRetrievalTitleBytes = 1024; + +const diagnosticRetrievalDetailsBytes = 8192; + +function boundedRunDiagnosticText(field: "title" | "details", maxBytes: number) { + const value = sql`${heartbeatRuns.resultJson} #>> ARRAY['terminalSessionFailure', ${field}]`; + return sql`case when octet_length(${value}) <= ${maxBytes} then ${value} + else left(${value}, ${Math.floor((maxBytes - 100) / 4)}) + || E'\\n[truncated for run retrieval; full text in run error/transcript]' end`; +} + +const heartbeatRunSafeResultJsonColumn = sql | null>` + case + when ${heartbeatRuns.resultJson} is null then null + when pg_column_size(${heartbeatRuns.resultJson}) <= ${HEARTBEAT_RUN_SAFE_RESULT_JSON_MAX_BYTES} + then ${heartbeatRuns.resultJson} + else jsonb_strip_nulls( + jsonb_build_object( + 'summary', left(${heartbeatRuns.resultJson} ->> 'summary', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS}), + 'result', left(${heartbeatRuns.resultJson} ->> 'result', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS}), + 'message', left(${heartbeatRuns.resultJson} ->> 'message', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS}), + 'error', left(${heartbeatRuns.resultJson} ->> 'error', ${HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS}), + 'stdout', left(${heartbeatRuns.resultJson} ->> 'stdout', ${HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS}), + 'stderr', left(${heartbeatRuns.resultJson} ->> 'stderr', ${HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS}), + 'terminalSessionFailure', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'terminalSessionFailure') = 'object' + then jsonb_strip_nulls(jsonb_build_object( + 'category', left(${heartbeatRuns.resultJson} #>> '{terminalSessionFailure,category}', 32), + 'title', ${boundedRunDiagnosticText("title", diagnosticRetrievalTitleBytes)}, + 'details', ${boundedRunDiagnosticText("details", diagnosticRetrievalDetailsBytes)}, + 'retrievalTruncated', case when + octet_length(${heartbeatRuns.resultJson} #>> '{terminalSessionFailure,title}') > ${diagnosticRetrievalTitleBytes} + or octet_length(${heartbeatRuns.resultJson} #>> '{terminalSessionFailure,details}') > ${diagnosticRetrievalDetailsBytes} + then to_jsonb(true) end, + 'truncatedFields', case when ${heartbeatRuns.resultJson} #> '{terminalSessionFailure,truncatedFields}' + in ('["title"]'::jsonb, '["details"]'::jsonb, '["title","details"]'::jsonb) + then ${heartbeatRuns.resultJson} #> '{terminalSessionFailure,truncatedFields}' end + )) end, + 'instructionSave', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'instructionSave') = 'object' + then jsonb_strip_nulls(jsonb_build_object( + 'state', left(${heartbeatRuns.resultJson} #>> '{instructionSave,state}', 32), + 'contract', left(${heartbeatRuns.resultJson} #>> '{instructionSave,contract}', 32), + 'entryFile', left(${heartbeatRuns.resultJson} #>> '{instructionSave,entryFile}', 512), + 'errorCode', left(${heartbeatRuns.resultJson} #>> '{instructionSave,errorCode}', 128), + 'errorMessage', left(${heartbeatRuns.resultJson} #>> '{instructionSave,errorMessage}', 1024), + 'storageWarning', left(${heartbeatRuns.resultJson} #>> '{instructionSave,storageWarning}', 1024) + )) end, + 'workspaceRestoreRecovery', case when ${heartbeatRuns.resultJson} #>> '{workspaceRestoreRecovery,schema}' = 'paperclip.workspace-restore-recovery.v1' + then jsonb_build_object('schema', 'paperclip.workspace-restore-recovery.v1') end, + 'workspaceRestoreFailure', case when ${heartbeatRuns.resultJson} ->> 'workspaceRestoreFailure' + in ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed') + then ${heartbeatRuns.resultJson} -> 'workspaceRestoreFailure' end, + 'cancellation', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'cancellation') = 'object' + then jsonb_strip_nulls(jsonb_build_object( + 'source', case when ${heartbeatRuns.resultJson} #>> '{cancellation,source}' + in ('operator', 'queued_message', 'shutdown', 'provider', 'transport', 'control_plane', 'unknown') + then ${heartbeatRuns.resultJson} #> '{cancellation,source}' end, + 'expected', case when jsonb_typeof(${heartbeatRuns.resultJson} #> '{cancellation,expected}') = 'boolean' + then ${heartbeatRuns.resultJson} #> '{cancellation,expected}' end, + 'initiator', jsonb_strip_nulls(jsonb_build_object( + 'type', case when ${heartbeatRuns.resultJson} #>> '{cancellation,initiator,type}' in ('user', 'agent', 'system', 'provider') + then ${heartbeatRuns.resultJson} #> '{cancellation,initiator,type}' end, + 'id', left(${heartbeatRuns.resultJson} #>> '{cancellation,initiator,id}', 128) + )), + 'reason', left(${heartbeatRuns.resultJson} #>> '{cancellation,reason}', 512), + 'recordedAt', left(${heartbeatRuns.resultJson} #>> '{cancellation,recordedAt}', 64) + )) end, + 'acpToolInventoryComplete', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'acpToolInventoryComplete') = 'boolean' + then ${heartbeatRuns.resultJson} -> 'acpToolInventoryComplete' end, + 'acpPendingToolCount', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'acpPendingToolCount') = 'number' + and length(${heartbeatRuns.resultJson} ->> 'acpPendingToolCount') < 16 + then ${heartbeatRuns.resultJson} -> 'acpPendingToolCount' end, + 'errorFamily', left(${heartbeatRuns.resultJson} ->> 'errorFamily', 32), + 'finalResponseRecorded', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'finalResponseRecorded') = 'boolean' + then ${heartbeatRuns.resultJson} -> 'finalResponseRecorded' end, + 'executionBeforeRestore', case when jsonb_typeof(${heartbeatRuns.resultJson} -> 'executionBeforeRestore') = 'object' + then jsonb_strip_nulls(jsonb_build_object( + 'errorCode', left(${heartbeatRuns.resultJson} #>> '{executionBeforeRestore,errorCode}', 128), + 'exitCode', case when jsonb_typeof(${heartbeatRuns.resultJson} #> '{executionBeforeRestore,exitCode}') = 'number' + and length(${heartbeatRuns.resultJson} #>> '{executionBeforeRestore,exitCode}') < 16 + then ${heartbeatRuns.resultJson} #> '{executionBeforeRestore,exitCode}' end, + 'signal', left(${heartbeatRuns.resultJson} #>> '{executionBeforeRestore,signal}', 50), + 'timedOut', case when jsonb_typeof(${heartbeatRuns.resultJson} #> '{executionBeforeRestore,timedOut}') = 'boolean' + then ${heartbeatRuns.resultJson} #> '{executionBeforeRestore,timedOut}' end + )) end, + 'stdoutTruncated', case + when length(${heartbeatRuns.resultJson} ->> 'stdout') > ${HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS} + then to_jsonb(true) + else null + end, + 'stderrTruncated', case + when length(${heartbeatRuns.resultJson} ->> 'stderr') > ${HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS} + then to_jsonb(true) + else null + end, + 'costUsd', coalesce( + ${heartbeatRuns.resultJson} -> 'costUsd', + ${heartbeatRuns.resultJson} -> 'cost_usd', + ${heartbeatRuns.resultJson} -> 'total_cost_usd' + ), + 'cost_usd', coalesce( + ${heartbeatRuns.resultJson} -> 'cost_usd', + ${heartbeatRuns.resultJson} -> 'costUsd', + ${heartbeatRuns.resultJson} -> 'total_cost_usd' + ), + 'total_cost_usd', coalesce( + ${heartbeatRuns.resultJson} -> 'total_cost_usd', + ${heartbeatRuns.resultJson} -> 'cost_usd', + ${heartbeatRuns.resultJson} -> 'costUsd' + ), + 'truncated', true, + 'truncationReason', 'oversized_result_json', + 'originalSizeBytes', pg_column_size(${heartbeatRuns.resultJson}) + ) + ) + end +`.as("resultJson"); + +// Execution admission needs retained server receipts even when presentation +// projection omits resultJson (SQL_ASCII or oversized provider output). Select +// only the evidence used by eligibility; never retrieve provider diagnostics. +const heartbeatRunExecutionEvidenceColumn = sql | null>` + case when ${heartbeatRuns.resultJson} is null then null else jsonb_build_object( + 'startupCancellation', ${heartbeatRuns.resultJson} -> 'startupCancellation', + 'startupPreparationSettledAt', ${heartbeatRuns.resultJson} -> 'startupPreparationSettledAt', + 'stopReason', ${heartbeatRuns.resultJson} -> 'stopReason', + 'timeoutSource', ${heartbeatRuns.resultJson} -> 'timeoutSource', + 'workspaceRestoreFailure', ${heartbeatRuns.resultJson} -> 'workspaceRestoreFailure', + 'workspaceRestoreRecovery', ${heartbeatRuns.resultJson} -> 'workspaceRestoreRecovery', + 'executionCancellation', ${heartbeatRuns.resultJson} -> 'executionCancellation', + 'nativeCancellation', ${heartbeatRuns.resultJson} -> 'nativeCancellation', + 'cancelledByActorType', ${heartbeatRuns.resultJson} -> 'cancelledByActorType', + 'cancelledByUserId', ${heartbeatRuns.resultJson} -> 'cancelledByUserId', + 'conversationContinuation', ${heartbeatRuns.resultJson} -> 'conversationContinuation', + 'cancellation', ${heartbeatRuns.resultJson} -> 'cancellation', + 'acpToolInventoryComplete', ${heartbeatRuns.resultJson} -> 'acpToolInventoryComplete', + 'acpPendingToolCount', ${heartbeatRuns.resultJson} -> 'acpPendingToolCount' + ) end +`.as("resultJson"); + +const heartbeatRunSafeColumns = { + ...getTableColumns(heartbeatRuns), + processGroupId: heartbeatRunProcessGroupIdColumn, + resultJson: heartbeatRunSafeResultJsonColumn, +} as const; + +const heartbeatRunSqlAsciiSafeColumns = { + ...getTableColumns(heartbeatRuns), + processGroupId: heartbeatRunProcessGroupIdColumn, + error: sql`NULL`.as("error"), + resultJson: sql | null>`NULL`.as("resultJson"), + stdoutExcerpt: sql`NULL`.as("stdoutExcerpt"), + stderrExcerpt: sql`NULL`.as("stderrExcerpt"), +} as const; + +const heartbeatRunLogAccessColumns = { + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + scopeKind: heartbeatRuns.scopeKind, + issueId: heartbeatRuns.issueId, + logStore: heartbeatRuns.logStore, + logRef: heartbeatRuns.logRef, +} as const; + +const heartbeatRunIssueSummaryColumns = { + id: heartbeatRuns.id, + runtimeMode: heartbeatRuns.runtimeMode, + status: heartbeatRuns.status, + invocationSource: heartbeatRuns.invocationSource, + triggerDetail: heartbeatRuns.triggerDetail, + contextCommentId: sql< + string | null + >`${heartbeatRuns.contextSnapshot} ->> 'commentId'`.as("contextCommentId"), + contextWakeCommentId: sql< + string | null + >`${heartbeatRuns.contextSnapshot} ->> 'wakeCommentId'`.as( + "contextWakeCommentId", + ), + startedAt: heartbeatRuns.startedAt, + finishedAt: heartbeatRuns.finishedAt, + createdAt: heartbeatRuns.createdAt, + agentId: heartbeatRuns.agentId, + logBytes: heartbeatRuns.logBytes, + processStartedAt: heartbeatRuns.processStartedAt, + livenessState: heartbeatRuns.livenessState, + livenessReason: heartbeatRuns.livenessReason, + continuationAttempt: heartbeatRuns.continuationAttempt, + lastUsefulActionAt: heartbeatRuns.lastUsefulActionAt, + nextAction: heartbeatRuns.nextAction, + lastOutputAt: heartbeatRuns.lastOutputAt, + lastOutputSeq: heartbeatRuns.lastOutputSeq, + lastOutputStream: heartbeatRuns.lastOutputStream, + lastOutputBytes: heartbeatRuns.lastOutputBytes, + issueId: heartbeatRuns.issueId, +} as const; + +export type UsageTotals = { + inputTokens: number; + cachedInputTokens: number; + outputTokens: number; +}; + +type SessionCompactionDecision = { + rotate: boolean; + reason: string | null; + handoffMarkdown: string | null; + previousRunId: string | null; +}; + +export function summarizeHeartbeatRunContextSnapshot( + contextSnapshot: Record | null | undefined, +): Record | null { + const summary: Record = {}; + const allowedKeys = [ + "issueId", + "taskId", + "taskKey", + "commentId", + "wakeCommentId", + "wakeReason", + "wakeSource", + "wakeTriggerDetail", + ] as const; + + for (const key of allowedKeys) { + const value = readNonEmptyString(contextSnapshot?.[key]); + if (value) summary[key] = value; + } + + return Object.keys(summary).length > 0 ? summary : null; +} + +export function summarizeHeartbeatRunListResultJson(input: { + summary?: string | null; + result?: string | null; + message?: string | null; + error?: string | null; + totalCostUsd?: string | null; + costUsd?: string | null; + costUsdCamel?: string | null; +}): Record | null { + const summary: Record = {}; + for (const [key, value] of [ + ["summary", input.summary], + ["result", input.result], + ["message", input.message], + ["error", input.error], + ] as const) { + const normalized = readNonEmptyString(value); + if (normalized) summary[key] = normalized; + } + + for (const [key, value] of [ + ["total_cost_usd", input.totalCostUsd], + ["cost_usd", input.costUsd], + ["costUsd", input.costUsdCamel], + ] as const) { + const normalized = readNonEmptyString(value); + if (!normalized) continue; + const parsed = Number(normalized); + if (Number.isFinite(parsed)) summary[key] = parsed; + } + + return Object.keys(summary).length > 0 ? summary : null; +} + +export function normalizeLedgerBillingType(value: unknown): BillingType { + const raw = readNonEmptyString(value); + switch (raw) { + case "api": + case "metered_api": + return "metered_api"; + case "subscription": + case "subscription_included": + return "subscription_included"; + case "subscription_overage": + return "subscription_overage"; + case "credits": + return "credits"; + case "fixed": + return "fixed"; + default: + return "unknown"; + } +} + +export function resolveLedgerBiller(result: AdapterExecutionResult): string { + return ( + readNonEmptyString(result.biller) ?? + readNonEmptyString(result.provider) ?? + "unknown" + ); +} + +export function normalizeBilledCostCents( + costUsd: number | null | undefined, + billingType: BillingType, +): number { + if (billingType === "subscription_included") return 0; + if (typeof costUsd !== "number" || !Number.isFinite(costUsd)) return 0; + return Math.max(0, Number((costUsd * 100).toFixed(7))); +} + +export function resolveLedgerCostStatus(input: { + costUsd: number | null | undefined; + billingType?: BillingType; + inputTokens: number; + cachedInputTokens: number; + outputTokens: number; +}): CostStatus { + if (input.billingType === "subscription_included") return "reported"; + // A paused turn can have neither a token receipt nor a cost receipt. Zero + // normalized counters do not establish that its billed cost was zero. + return typeof input.costUsd === "number" && + Number.isFinite(input.costUsd) && + input.costUsd >= 0 + ? "reported" + : "unpriced"; +} + +export function resolveCacheAdjustedCostUsd(input: { + costUsd?: number | null; + cacheAdjustedCostUsd?: number | null; +}) { + const explicit = input.cacheAdjustedCostUsd; + if ( + typeof explicit === "number" && + Number.isFinite(explicit) && + explicit >= 0 + ) { + return explicit; + } + const reported = input.costUsd; + if ( + typeof reported === "number" && + Number.isFinite(reported) && + reported >= 0 + ) { + return reported; + } + return null; +} + +export async function resolveLedgerScopeForRun( + db: Db, + companyId: string, + run: typeof heartbeatRuns.$inferSelect, +) { + const context = parseObject(run.contextSnapshot); + const contextIssueId = readNonEmptyString(context.issueId); + const contextProjectId = readNonEmptyString(context.projectId); + + if (!contextIssueId) { + return { + issueId: null, + projectId: contextProjectId, + billingCode: null, + }; + } + + const issue = await db + .select({ + id: issues.id, + projectId: issues.projectId, + billingCode: issues.billingCode, + }) + .from(issues) + .where(and(eq(issues.id, contextIssueId), eq(issues.companyId, companyId))) + .then((rows) => rows[0] ?? null); + + return { + issueId: issue?.id ?? null, + projectId: issue?.projectId ?? contextProjectId, + billingCode: issue?.billingCode ?? null, + }; +} + +type ResumeSessionRow = { + sessionParamsJson: Record | null; + sessionDisplayId: string | null; + lastRunId: string | null; +}; + +export function buildExplicitResumeSessionOverride(input: { + adapterType?: string | null; + resumeFromRunId: string; + resumeRunSessionIdBefore: string | null; + resumeRunSessionIdAfter: string | null; + resumeRunSessionParams?: Record | null; + taskSession: ResumeSessionRow | null; + sessionCodec: AdapterSessionCodec; +}) { + const resumeRunSessionIdAfter = truncateDisplayId( + input.resumeRunSessionIdAfter, + ); + const resumeRunSessionIdBefore = truncateDisplayId( + input.resumeRunSessionIdBefore, + ); + const desiredDisplayId = requiresCanonicalSessionIds(input.adapterType) + ? isCanonicalSessionIdForAdapter(input.adapterType, resumeRunSessionIdAfter) + ? resumeRunSessionIdAfter + : isCanonicalSessionIdForAdapter( + input.adapterType, + resumeRunSessionIdBefore, + ) + ? resumeRunSessionIdBefore + : null + : (resumeRunSessionIdAfter ?? resumeRunSessionIdBefore); + const runSessionParams = requiresCanonicalSessionIds(input.adapterType) + ? normalizeResumeParamsForAdapter( + input.adapterType, + input.sessionCodec.deserialize(input.resumeRunSessionParams ?? null), + ) + : null; + const runSessionDisplayId = truncateDisplayId( + readNonEmptyString(runSessionParams?.sessionId), + ); + const taskSessionParams = normalizeResumeParamsForAdapter( + input.adapterType, + input.sessionCodec.deserialize( + input.taskSession?.sessionParamsJson ?? null, + ), + ); + const taskSessionRawDisplayId = input.taskSession?.sessionDisplayId ?? null; + const taskSessionDisplayId = truncateDisplayId( + requiresCanonicalSessionIds(input.adapterType) + ? (readNonEmptyString(taskSessionParams?.sessionId) ?? + (isCanonicalSessionIdForAdapter( + input.adapterType, + taskSessionRawDisplayId, + ) + ? taskSessionRawDisplayId + : null)) + : (taskSessionRawDisplayId ?? + (input.sessionCodec.getDisplayId + ? input.sessionCodec.getDisplayId(taskSessionParams) + : null) ?? + readNonEmptyString(taskSessionParams?.sessionId)), + ); + const canReuseTaskSessionParams = + input.taskSession != null && + (!requiresCanonicalSessionIds(input.adapterType) || + taskSessionParams != null) && + (input.taskSession.lastRunId === input.resumeFromRunId || + (!!desiredDisplayId && taskSessionDisplayId === desiredDisplayId)); + const sessionParams = canReuseTaskSessionParams + ? taskSessionParams + : runSessionParams + ? runSessionParams + : desiredDisplayId + ? { sessionId: desiredDisplayId } + : null; + const sessionDisplayId = canReuseTaskSessionParams + ? taskSessionDisplayId + : runSessionParams + ? runSessionDisplayId + : desiredDisplayId; + + if (!sessionDisplayId && !sessionParams) return null; + return { + sessionDisplayId, + sessionParams, + }; +} + +export function normalizeUsageTotals( + usage: UsageSummary | null | undefined, +): UsageTotals | null { + if (!usage) return null; + return { + inputTokens: Math.max(0, Math.floor(asNumber(usage.inputTokens, 0))), + cachedInputTokens: Math.max( + 0, + Math.floor(asNumber(usage.cachedInputTokens, 0)), + ), + outputTokens: Math.max(0, Math.floor(asNumber(usage.outputTokens, 0))), + }; +} + +function readRawUsageTotals(usageJson: unknown): UsageTotals | null { + const parsed = parseObject(usageJson); + if (Object.keys(parsed).length === 0) return null; + + const inputTokens = Math.max( + 0, + Math.floor( + asNumber(parsed.rawInputTokens, asNumber(parsed.inputTokens, 0)), + ), + ); + const cachedInputTokens = Math.max( + 0, + Math.floor( + asNumber( + parsed.rawCachedInputTokens, + asNumber(parsed.cachedInputTokens, 0), + ), + ), + ); + const outputTokens = Math.max( + 0, + Math.floor( + asNumber(parsed.rawOutputTokens, asNumber(parsed.outputTokens, 0)), + ), + ); + + if (inputTokens <= 0 && cachedInputTokens <= 0 && outputTokens <= 0) { + return null; + } + + return { + inputTokens, + cachedInputTokens, + outputTokens, + }; +} + +export function normalizeAdapterRunUsage( + current: UsageTotals | null, + previous: UsageTotals | null, + usageBasis?: "per_run" | "session_cumulative" | null, +): UsageTotals | null { + if (!current) return null; + if (!previous || usageBasis !== "session_cumulative") return { ...current }; + + const inputTokens = + current.inputTokens >= previous.inputTokens + ? current.inputTokens - previous.inputTokens + : current.inputTokens; + const cachedInputTokens = + current.cachedInputTokens >= previous.cachedInputTokens + ? current.cachedInputTokens - previous.cachedInputTokens + : current.cachedInputTokens; + const outputTokens = + current.outputTokens >= previous.outputTokens + ? current.outputTokens - previous.outputTokens + : current.outputTokens; + + return { + inputTokens: Math.max(0, inputTokens), + cachedInputTokens: Math.max(0, cachedInputTokens), + outputTokens: Math.max(0, outputTokens), + }; +} + +function formatCount(value: number | null | undefined) { + if (typeof value !== "number" || !Number.isFinite(value)) return "0"; + return value.toLocaleString("en-US"); +} + +export function parseSessionCompactionPolicy( + agent: typeof agents.$inferSelect, +): SessionCompactionPolicy { + return resolveSessionCompactionPolicy(agent.adapterType, agent.runtimeConfig) + .policy; +} + +/** + * Synthetic task key for timer/heartbeat wakes that have no issue context. + * This allows timer wakes to participate in the `agentTaskSessions` system + * and benefit from robust session resume, instead of relying solely on the + * simpler `agentRuntimeState.sessionId` fallback. + */ +const HEARTBEAT_TASK_KEY = "__heartbeat__"; + +/** + * Extended task key derivation that falls back to a stable synthetic key + * for timer/heartbeat wakes. The synthetic key keeps the + * `agentTaskSessions` row addressable across heartbeats so the row can be + * cleared and re-keyed deterministically. Unscoped exploratory timer wakes + * still start fresh to avoid accumulating low-value inbox scans, while timer + * wakes scoped to a real issue reuse that issue's task session. + * + * The synthetic key is only used when: + * - No explicit task/issue key exists in the context + * - The wake source is "timer" (scheduled heartbeat) + */ +export function deriveTaskKeyWithHeartbeatFallback( + contextSnapshot: Record | null | undefined, + payload: Record | null | undefined, +) { + const explicit = deriveTaskKey(contextSnapshot, payload); + if (explicit) return explicit; + + const wakeSource = readNonEmptyString(contextSnapshot?.wakeSource); + if (wakeSource === "timer") return HEARTBEAT_TASK_KEY; + + return null; +} + +export function shouldResetTaskSessionForWake( + contextSnapshot: Record | null | undefined, +) { + if (contextSnapshot?.forceFreshSession === true) return true; + + const wakeReason = readNonEmptyString(contextSnapshot?.wakeReason); + if ( + wakeReason === "issue_assigned" || + wakeReason === EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON || + wakeReason === "execution_approval_requested" || + // PF-4: unscoped timer wakes are exploratory ("any new work?") and should + // not accumulate low-value inbox scans. Issue-scoped timer wakes are + // continuation work, so reuse their task session to avoid paying the full + // session-start and re-orientation cost on every heartbeat. + (wakeReason === "heartbeat_timer" && !deriveTaskKey(contextSnapshot, null)) + ) { + return true; + } + return false; +} + +export function describeSessionResetReason( + contextSnapshot: Record | null | undefined, +) { + if (contextSnapshot?.forceFreshSession === true) + return "forceFreshSession was requested"; + + const wakeReason = readNonEmptyString(contextSnapshot?.wakeReason); + if (wakeReason === "issue_assigned") return "wake reason is issue_assigned"; + if (wakeReason === EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON) { + return `wake reason is ${EXECUTION_REVIEW_PARTICIPANT_RECOVERY_WAKE_REASON}`; + } + if (wakeReason === "execution_approval_requested") + return "wake reason is execution_approval_requested"; + // PF-4: paired with shouldResetTaskSessionForWake — keep the reason wording + // explicit so run logs make session reuse/reset behavior legible. + if ( + wakeReason === "heartbeat_timer" && + !deriveTaskKey(contextSnapshot, null) + ) { + return "wake reason is heartbeat_timer (unscoped timer wake starts fresh)"; + } + return null; +} + +export function truncateDisplayId(value: string | null | undefined, max = 128) { + if (!value) return null; + return value.length > max ? value.slice(0, max) : value; +} + +const defaultSessionCodec: AdapterSessionCodec = { + deserialize(raw: unknown) { + const asObj = parseObject(raw); + if (Object.keys(asObj).length > 0) return asObj; + const sessionId = readNonEmptyString( + (raw as Record | null)?.sessionId, + ); + if (sessionId) return { sessionId }; + return null; + }, + serialize(params: Record | null) { + if (!params || Object.keys(params).length === 0) return null; + return params; + }, + getDisplayId(params: Record | null) { + return readNonEmptyString(params?.sessionId); + }, +}; + +export function getAdapterSessionCodec(adapterType: string) { + const adapter = getServerAdapter(adapterType); + return adapter.sessionCodec ?? defaultSessionCodec; +} + +export function normalizeSessionParams( + params: Record | null | undefined, +) { + if (!params) return null; + return Object.keys(params).length > 0 ? params : null; +} + +export type RunSessionOutcome = + "succeeded" | "interrupted" | "failed" | "cancelled" | "timed_out"; + +const HERMES_ADAPTER_TYPE = "hermes_local"; + +const HERMES_SESSION_ID_REGEX = + /^(?:\d{8}_\d{6}_[A-Za-z0-9_-]{4,}|[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})$/; + +export function requiresCanonicalSessionIds(adapterType: string | null | undefined) { + return adapterType === HERMES_ADAPTER_TYPE; +} + +export function isCanonicalSessionIdForAdapter( + adapterType: string | null | undefined, + sessionId: string | null | undefined, +) { + if (!sessionId) return false; + if (!requiresCanonicalSessionIds(adapterType)) return true; + return HERMES_SESSION_ID_REGEX.test(sessionId); +} + +export function normalizeResumeParamsForAdapter( + adapterType: string | null | undefined, + params: Record | null | undefined, +) { + const normalized = normalizeSessionParams(params); + if (!normalized) return null; + if (!requiresCanonicalSessionIds(adapterType)) return normalized; + const sessionId = readNonEmptyString(normalized.sessionId); + return isCanonicalSessionIdForAdapter(adapterType, sessionId) + ? normalized + : null; +} + +export function resolveNextSessionState(input: { + adapterType?: string | null; + codec: AdapterSessionCodec; + adapterResult: AdapterExecutionResult; + outcome: RunSessionOutcome; + previousParams: Record | null; + previousDisplayId: string | null; + previousLegacySessionId: string | null; +}) { + const { + adapterType, + codec, + adapterResult, + previousParams, + previousDisplayId, + previousLegacySessionId, + } = input; + + if (adapterResult.clearSession) { + return { + params: null as Record | null, + displayId: null as string | null, + legacySessionId: null as string | null, + }; + } + + if (!requiresCanonicalSessionIds(adapterType)) { + const explicitParams = adapterResult.sessionParams; + const hasExplicitParams = adapterResult.sessionParams !== undefined; + const hasExplicitSessionId = adapterResult.sessionId !== undefined; + const explicitSessionId = readNonEmptyString(adapterResult.sessionId); + const hasExplicitDisplay = adapterResult.sessionDisplayId !== undefined; + const explicitDisplayId = readNonEmptyString( + adapterResult.sessionDisplayId, + ); + const shouldUsePrevious = + !hasExplicitParams && !hasExplicitSessionId && !hasExplicitDisplay; + + const candidateParams = hasExplicitParams + ? explicitParams + : hasExplicitSessionId + ? explicitSessionId + ? { sessionId: explicitSessionId } + : null + : previousParams; + + const serialized = normalizeSessionParams( + codec.serialize(normalizeSessionParams(candidateParams) ?? null), + ); + const deserialized = normalizeSessionParams(codec.deserialize(serialized)); + + const displayId = truncateDisplayId( + explicitDisplayId ?? + (codec.getDisplayId ? codec.getDisplayId(deserialized) : null) ?? + readNonEmptyString(deserialized?.sessionId) ?? + (shouldUsePrevious ? previousDisplayId : null) ?? + explicitSessionId ?? + (shouldUsePrevious ? previousLegacySessionId : null), + ); + + const legacySessionId = + explicitSessionId ?? + readNonEmptyString(deserialized?.sessionId) ?? + displayId ?? + (shouldUsePrevious ? previousLegacySessionId : null); + + return { + params: serialized, + displayId, + legacySessionId, + }; + } + + const previousSerializedParams = normalizeResumeParamsForAdapter( + adapterType, + codec.serialize( + normalizeResumeParamsForAdapter(adapterType, previousParams), + ), + ); + const validPreviousDisplayId = isCanonicalSessionIdForAdapter( + adapterType, + previousDisplayId, + ) + ? previousDisplayId + : null; + const validPreviousLegacySessionId = isCanonicalSessionIdForAdapter( + adapterType, + previousLegacySessionId, + ) + ? previousLegacySessionId + : null; + const previousState = () => { + const displayId = truncateDisplayId( + readNonEmptyString(previousSerializedParams?.sessionId) ?? + validPreviousDisplayId ?? + validPreviousLegacySessionId, + ); + return { + params: previousSerializedParams, + displayId, + legacySessionId: + readNonEmptyString(previousSerializedParams?.sessionId) ?? + displayId ?? + validPreviousLegacySessionId, + }; + }; + + if (input.outcome !== "succeeded") { + return previousState(); + } + + const explicitParams = adapterResult.sessionParams; + const hasExplicitParams = adapterResult.sessionParams !== undefined; + const explicitSessionId = readNonEmptyString(adapterResult.sessionId); + const validExplicitSessionId = isCanonicalSessionIdForAdapter( + adapterType, + explicitSessionId, + ) + ? explicitSessionId + : null; + const explicitDisplayId = readNonEmptyString(adapterResult.sessionDisplayId); + const validExplicitDisplayId = isCanonicalSessionIdForAdapter( + adapterType, + explicitDisplayId, + ) + ? explicitDisplayId + : null; + const explicitSerializedParams = hasExplicitParams + ? normalizeResumeParamsForAdapter( + adapterType, + codec.serialize(normalizeSessionParams(explicitParams) ?? null), + ) + : null; + const explicitCanonicalSessionId = + readNonEmptyString(explicitSerializedParams?.sessionId) ?? + validExplicitSessionId ?? + validExplicitDisplayId; + + if (!explicitCanonicalSessionId) { + return previousState(); + } + + const serialized = normalizeResumeParamsForAdapter( + adapterType, + codec.serialize({ sessionId: explicitCanonicalSessionId }), + ); + const displayId = truncateDisplayId( + readNonEmptyString(serialized?.sessionId) ?? + (codec.getDisplayId ? codec.getDisplayId(serialized) : null) ?? + explicitCanonicalSessionId, + ); + const legacySessionId = + readNonEmptyString(serialized?.sessionId) ?? explicitCanonicalSessionId; + + return { + params: serialized, + displayId, + legacySessionId, + }; +} + +function readNonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value : null; +} + +/** Run retrieval and session persistence bound to one service database. */ +export function createHeartbeatRunState(db: Db) { + let unsafeTextProjectionPromise: Promise | null = null; + + async function hasUnsafeTextProjectionDatabase() { + if (!unsafeTextProjectionPromise) { + unsafeTextProjectionPromise = db + .execute( + sql`select current_setting('server_encoding') as server_encoding`, + ) + .then((rows) => { + const first = Array.isArray(rows) ? rows[0] : null; + const serverEncoding = + typeof first === "object" && first !== null + ? (first as Record).server_encoding + : null; + return ( + typeof serverEncoding === "string" && + serverEncoding.toUpperCase() === "SQL_ASCII" + ); + }) + .catch((err) => { + logger.warn( + { err }, + "failed to inspect database server encoding; using conservative heartbeat result projection", + ); + return true; + }); + } + return unsafeTextProjectionPromise; + } + + async function getAgent(agentId: string) { + return db + .select() + .from(agents) + .where(eq(agents.id, agentId)) + .then((rows) => rows[0] ?? null); + } + + async function getRun( + runId: string, + opts?: { unsafeFullResultJson?: boolean; includeExecutionEvidence?: boolean }, + ) { + const safeForLegacyEncoding = + !opts?.unsafeFullResultJson && (await hasUnsafeTextProjectionDatabase()); + const columns = opts?.unsafeFullResultJson + ? getTableColumns(heartbeatRuns) + : safeForLegacyEncoding ? heartbeatRunSqlAsciiSafeColumns : heartbeatRunSafeColumns; + return db + .select(opts?.includeExecutionEvidence + ? { ...columns, resultJson: heartbeatRunExecutionEvidenceColumn } + : columns) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + } + + async function getRunLogAccess(runId: string) { + return db + .select(heartbeatRunLogAccessColumns) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + } + + async function getRuntimeState(agentId: string) { + return db + .select() + .from(agentRuntimeState) + .where(eq(agentRuntimeState.agentId, agentId)) + .then((rows) => rows[0] ?? null); + } + + async function getLatestAgentConfigRevision( + companyId: string, + agentId: string, + ) { + return db + .select({ + id: agentConfigRevisions.id, + changedKeys: agentConfigRevisions.changedKeys, + createdAt: agentConfigRevisions.createdAt, + }) + .from(agentConfigRevisions) + .where( + and( + eq(agentConfigRevisions.companyId, companyId), + eq(agentConfigRevisions.agentId, agentId), + ), + ) + .orderBy( + desc(agentConfigRevisions.createdAt), + desc(agentConfigRevisions.id), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function getTaskSession( + companyId: string, + agentId: string, + adapterType: string, + taskKey: string, + ) { + return db + .select() + .from(agentTaskSessions) + .where( + and( + eq(agentTaskSessions.companyId, companyId), + eq(agentTaskSessions.agentId, agentId), + eq(agentTaskSessions.adapterType, adapterType), + eq(agentTaskSessions.taskKey, taskKey), + ), + ) + .then((rows) => rows[0] ?? null); + } + + async function getLatestRunForSession( + agentId: string, + sessionId: string, + opts?: { excludeRunId?: string | null }, + ) { + const conditions = [ + eq(heartbeatRuns.agentId, agentId), + eq(heartbeatRuns.sessionIdAfter, sessionId), + ]; + if (opts?.excludeRunId) { + conditions.push(sql`${heartbeatRuns.id} <> ${opts.excludeRunId}`); + } + return db + .select({ + id: heartbeatRuns.id, + usageJson: heartbeatRuns.usageJson, + }) + .from(heartbeatRuns) + .where(and(...conditions)) + .orderBy(desc(heartbeatRuns.createdAt)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function getOldestRunForSession(agentId: string, sessionId: string) { + return db + .select({ + id: heartbeatRuns.id, + createdAt: heartbeatRuns.createdAt, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.agentId, agentId), + eq(heartbeatRuns.sessionIdAfter, sessionId), + ), + ) + .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function resolveNormalizedUsageForSession(input: { + agentId: string; + runId: string; + sessionId: string | null; + rawUsage: UsageTotals | null; + usageBasis?: "per_run" | "session_cumulative" | null; + }) { + const { agentId, runId, sessionId, rawUsage, usageBasis } = input; + // Adapters that declare per-run usage (e.g. the ACPX lane reports each + // turn's tokens, not session totals) must not be session-delta'd, or + // consecutive runs would be undercounted. + if (!sessionId || !rawUsage || usageBasis !== "session_cumulative") { + return { + normalizedUsage: rawUsage, + previousRawUsage: null as UsageTotals | null, + derivedFromSessionTotals: false, + }; + } + + const previousRun = await getLatestRunForSession(agentId, sessionId, { + excludeRunId: runId, + }); + const previousRawUsage = readRawUsageTotals(previousRun?.usageJson); + return { + normalizedUsage: normalizeAdapterRunUsage(rawUsage, previousRawUsage, usageBasis), + previousRawUsage, + derivedFromSessionTotals: previousRawUsage !== null, + }; + } + + async function evaluateSessionCompaction(input: { + agent: typeof agents.$inferSelect; + sessionId: string | null; + issueId: string | null; + continuationSummaryBody?: string | null; + }): Promise { + const { agent, sessionId, issueId } = input; + if (!sessionId) { + return { + rotate: false, + reason: null, + handoffMarkdown: null, + previousRunId: null, + }; + } + + const policy = parseSessionCompactionPolicy(agent); + if (!policy.enabled || !hasSessionCompactionThresholds(policy)) { + return { + rotate: false, + reason: null, + handoffMarkdown: null, + previousRunId: null, + }; + } + + const fetchLimit = Math.max( + policy.maxSessionRuns > 0 ? policy.maxSessionRuns + 1 : 0, + 4, + ); + const runs = await db + .select({ + id: heartbeatRuns.id, + createdAt: heartbeatRuns.createdAt, + usageJson: heartbeatRuns.usageJson, + error: heartbeatRuns.error, + terminalFailureCategory: sql`case + when jsonb_typeof(${heartbeatRuns.resultJson} -> 'terminalSessionFailure') = 'object' + then coalesce(left(${heartbeatRuns.resultJson} #>> '{terminalSessionFailure,category}', 32), 'unknown') end`, + ...heartbeatRunListResultColumns, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.agentId, agent.id), + eq(heartbeatRuns.sessionIdAfter, sessionId), + ), + ) + .orderBy(desc(heartbeatRuns.createdAt)) + .limit(fetchLimit); + + if (runs.length === 0) { + return { + rotate: false, + reason: null, + handoffMarkdown: null, + previousRunId: null, + }; + } + + const latestRun = runs[0] ?? null; + const oldestRun = + policy.maxSessionAgeHours > 0 + ? await getOldestRunForSession(agent.id, sessionId) + : (runs[runs.length - 1] ?? latestRun); + const latestRawUsage = readRawUsageTotals(latestRun?.usageJson); + // Historical Codex/Gemini raw input includes cache reads. Only add the + // separate cache counter when the writer explicitly saved exclusive input. + const latestRawInputTokens = (latestRawUsage?.inputTokens ?? 0) + + (latestRun?.usageJson?.rawInputIncludesCached === false ? latestRawUsage?.cachedInputTokens ?? 0 : 0); + const sessionAgeHours = + latestRun && oldestRun + ? Math.max( + 0, + (new Date(latestRun.createdAt).getTime() - + new Date(oldestRun.createdAt).getTime()) / + (1000 * 60 * 60), + ) + : 0; + + let reason: string | null = null; + if (policy.maxSessionRuns > 0 && runs.length > policy.maxSessionRuns) { + reason = `session exceeded ${policy.maxSessionRuns} runs`; + } else if ( + policy.maxRawInputTokens > 0 && + latestRawUsage && + latestRawInputTokens >= policy.maxRawInputTokens + ) { + reason = + `session raw input reached ${formatCount(latestRawInputTokens)} tokens ` + + `(threshold ${formatCount(policy.maxRawInputTokens)})`; + } else if ( + policy.maxSessionAgeHours > 0 && + sessionAgeHours >= policy.maxSessionAgeHours + ) { + reason = `session age reached ${Math.floor(sessionAgeHours)} hours`; + } + + if (!reason || !latestRun) { + return { + rotate: false, + reason: null, + handoffMarkdown: null, + previousRunId: latestRun?.id ?? null, + }; + } + + const latestSummary = summarizeHeartbeatRunListResultJson({ + summary: latestRun?.resultSummary, + result: latestRun?.resultResult, + message: latestRun?.resultMessage, + error: latestRun?.resultError, + totalCostUsd: latestRun?.resultTotalCostUsd, + costUsd: latestRun?.resultCostUsd, + costUsdCamel: latestRun?.resultCostUsdCamel, + }); + const latestTextSummary = + readNonEmptyString(latestSummary?.summary) ?? + readNonEmptyString(latestSummary?.result) ?? + readNonEmptyString(latestSummary?.message) ?? + readNonEmptyString(summarizeRunErrorForModel(latestRun.error, latestRun.terminalFailureCategory)); + + const handoffMarkdown = [ + "Paperclip session handoff:", + `- Previous session: ${sessionId}`, + issueId ? `- Issue: ${issueId}` : "", + `- Rotation reason: ${reason}`, + latestTextSummary ? `- Last run summary: ${latestTextSummary}` : "", + input.continuationSummaryBody + ? `- Issue continuation summary: ${input.continuationSummaryBody.slice(0, 1_500)}` + : "", + "Continue from the current task state. Rebuild only the minimum context you need.", + ] + .filter(Boolean) + .join("\n"); + + return { + rotate: true, + reason, + handoffMarkdown, + previousRunId: latestRun.id, + }; + } + + async function resolveSessionBeforeForWakeup( + agent: typeof agents.$inferSelect, + taskKey: string | null, + ) { + if (taskKey) { + const codec = getAdapterSessionCodec(agent.adapterType); + const existingTaskSession = await getTaskSession( + agent.companyId, + agent.id, + agent.adapterType, + taskKey, + ); + const parsedParams = normalizeSessionParams( + codec.deserialize(existingTaskSession?.sessionParamsJson ?? null), + ); + return truncateDisplayId( + existingTaskSession?.sessionDisplayId ?? + (codec.getDisplayId ? codec.getDisplayId(parsedParams) : null) ?? + readNonEmptyString(parsedParams?.sessionId), + ); + } + + const runtimeForRun = await getRuntimeState(agent.id); + return runtimeForRun?.sessionId ?? null; + } + + async function hasResolvableSessionWorkspaceCwd( + sessionParams: Record | null | undefined, + ) { + const cwd = readNonEmptyString(sessionParams?.cwd); + if (!cwd || isUnsafeSessionWorkspaceCwd(cwd)) return false; + return fs + .stat(cwd) + .then((stats) => stats.isDirectory()) + .catch(() => false); + } + + async function hasResolvablePriorSessionWorkspaceForWake(input: { + agent: typeof agents.$inferSelect; + contextSnapshot: Record; + taskKey: string | null; + explicitResumeSession: Awaited< + ReturnType + > | null; + }) { + if ( + await hasResolvableSessionWorkspaceCwd( + input.explicitResumeSession?.sessionParams, + ) + ) + return true; + if (shouldResetTaskSessionForWake(input.contextSnapshot)) return false; + if (!input.taskKey) return false; + + const codec = getAdapterSessionCodec(input.agent.adapterType); + const taskSession = await getTaskSession( + input.agent.companyId, + input.agent.id, + input.agent.adapterType, + input.taskKey, + ); + const taskSessionParams = normalizeResumeParamsForAdapter( + input.agent.adapterType, + codec.deserialize(taskSession?.sessionParamsJson ?? null), + ); + return hasResolvableSessionWorkspaceCwd(taskSessionParams); + } + + async function resolveExplicitResumeSessionOverride( + agent: typeof agents.$inferSelect, + payload: Record | null, + taskKey: string | null, + ) { + const resumeFromRunId = readNonEmptyString(payload?.resumeFromRunId); + if (!resumeFromRunId) return null; + + const resumeRun = await db + .select({ + id: heartbeatRuns.id, + contextSnapshot: heartbeatRuns.contextSnapshot, + resultJson: heartbeatRuns.resultJson, + sessionIdBefore: heartbeatRuns.sessionIdBefore, + sessionIdAfter: heartbeatRuns.sessionIdAfter, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.id, resumeFromRunId), + eq(heartbeatRuns.companyId, agent.companyId), + eq(heartbeatRuns.agentId, agent.id), + ), + ) + .then((rows) => rows[0] ?? null); + if (!resumeRun) return null; + + const resumeContext = parseObject(resumeRun.contextSnapshot); + const resumeTaskKey = deriveTaskKey(resumeContext, null) ?? taskKey; + const resumeTaskSession = resumeTaskKey + ? await getTaskSession( + agent.companyId, + agent.id, + agent.adapterType, + resumeTaskKey, + ) + : null; + const sessionCodec = getAdapterSessionCodec(agent.adapterType); + const resumeRunResult = parseObject(resumeRun.resultJson); + const resumeRunSessionId = requiresCanonicalSessionIds(agent.adapterType) + ? (readNonEmptyString(resumeRunResult.sessionId) ?? + readNonEmptyString(resumeRunResult.session_id)) + : null; + const sessionOverride = buildExplicitResumeSessionOverride({ + adapterType: agent.adapterType, + resumeFromRunId, + resumeRunSessionIdBefore: resumeRun.sessionIdBefore, + resumeRunSessionIdAfter: resumeRun.sessionIdAfter, + resumeRunSessionParams: resumeRunSessionId + ? { sessionId: resumeRunSessionId } + : null, + taskSession: resumeTaskSession, + sessionCodec, + }); + if (!sessionOverride) return null; + + return { + resumeFromRunId, + taskKey: resumeTaskKey, + issueId: readNonEmptyString(resumeContext.issueId), + taskId: + readNonEmptyString(resumeContext.taskId) ?? + readNonEmptyString(resumeContext.issueId), + sessionDisplayId: sessionOverride.sessionDisplayId, + sessionParams: sessionOverride.sessionParams, + }; + } + + async function upsertTaskSession(input: { + companyId: string; + agentId: string; + adapterType: string; + taskKey: string; + sessionParamsJson: Record | null; + sessionDisplayId: string | null; + lastRunId: string | null; + lastError: string | null; + }) { + return db.transaction(async (tx) => { + const [issue] = await tx.select().from(issues).where(and(sql`${issues.id}::text = ${input.taskKey}`, eq(issues.companyId, input.companyId))).for("update"); + if (isConversation(issue)) { + const [run] = input.lastRunId ? await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.lastRunId)) : []; + if (run?.status === "cancelled" || run?.contextSnapshot?.conversationSessionGeneration !== issue.conversationSessionGeneration) return null; + } + const existing = await tx.select().from(agentTaskSessions).where(and(eq(agentTaskSessions.companyId, input.companyId), eq(agentTaskSessions.agentId, input.agentId), eq(agentTaskSessions.adapterType, input.adapterType), eq(agentTaskSessions.taskKey, input.taskKey))).then((rows) => rows[0] ?? null); + if (existing) { + return tx + .update(agentTaskSessions) + .set({ + sessionParamsJson: input.sessionParamsJson, + sessionDisplayId: input.sessionDisplayId, + lastRunId: input.lastRunId, + lastError: input.lastError, + updatedAt: new Date(), + }) + .where(eq(agentTaskSessions.id, existing.id)) + .returning() + .then((rows) => rows[0] ?? null); + } + + return tx + .insert(agentTaskSessions) + .values({ + companyId: input.companyId, + agentId: input.agentId, + adapterType: input.adapterType, + taskKey: input.taskKey, + sessionParamsJson: input.sessionParamsJson, + sessionDisplayId: input.sessionDisplayId, + lastRunId: input.lastRunId, + lastError: input.lastError, + }) + .returning() + .then((rows) => rows[0] ?? null); + }); + } + + async function clearTaskSessions( + companyId: string, + agentId: string, + opts?: { + taskKey?: string | null; + adapterType?: string | null; + expectedRunId?: string; + includeIssueAliases?: boolean; + }, + ) { + const conditions = [ + eq(agentTaskSessions.companyId, companyId), + eq(agentTaskSessions.agentId, agentId), + ]; + if (opts?.taskKey) { + const exactTaskKey = eq(agentTaskSessions.taskKey, opts.taskKey); + if (opts.includeIssueAliases) { + const selectedIssue = isUuidLike(opts.taskKey) + ? eq(issues.id, opts.taskKey) + : eq(issues.identifier, opts.taskKey.toUpperCase()); + // Operator task resets accept the UUID sent by run detail and the + // identifier used by some saved sessions. Resolve only from the current + // same-company issue row, in this DELETE's snapshot; arbitrary custom + // keys retain exact-match behavior and run/model context grants no alias. + conditions.push( + or( + exactTaskKey, + sql`exists ( + select 1 from ${issues} + where ${issues.companyId} = ${companyId} + and ${selectedIssue} + and (${agentTaskSessions.taskKey} = ${issues.id}::text + or ${agentTaskSessions.taskKey} = ${issues.identifier}) + )`, + )!, + ); + } else { + conditions.push(exactTaskKey); + } + } + if (opts?.adapterType) { + conditions.push(eq(agentTaskSessions.adapterType, opts.adapterType)); + } + + return db.transaction(async (tx) => { + if (opts?.taskKey && opts.expectedRunId) { + const [issue] = await tx.select().from(issues).where(sql`${issues.id}::text = ${opts.taskKey}`).for("update"); + if (isConversation(issue)) { + const [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, opts.expectedRunId)); + if (run?.status === "cancelled" || run?.contextSnapshot?.conversationSessionGeneration !== issue.conversationSessionGeneration) return 0; + } + } + return tx.delete(agentTaskSessions).where(and(...conditions)).returning().then((rows) => rows.length); + }); + } + + async function ensureRuntimeState(agent: typeof agents.$inferSelect) { + const existing = await getRuntimeState(agent.id); + if (existing) return existing; + + const inserted = await db + .insert(agentRuntimeState) + .values({ + agentId: agent.id, + companyId: agent.companyId, + adapterType: agent.adapterType, + stateJson: {}, + }) + .onConflictDoNothing({ + target: agentRuntimeState.agentId, + }) + .returning() + .then((rows) => rows[0] ?? null); + if (inserted) return inserted; + + const ensured = await getRuntimeState(agent.id); + if (!ensured) { + throw new Error(`Failed to ensure runtime state for agent ${agent.id}`); + } + return ensured; + } + + return { + getAgent, + resolveSessionBeforeForWakeup, + getRun, + ensureRuntimeState, + getTaskSession, + getLatestAgentConfigRevision, + evaluateSessionCompaction, + upsertTaskSession, + resolveNormalizedUsageForSession, + clearTaskSessions, + resolveExplicitResumeSessionOverride, + hasResolvablePriorSessionWorkspaceForWake, + getRunLogAccess, + + listRuns: async ( + companyId: string, + agentId?: string, + limit?: number, + options: { summary?: boolean } = {}, + ) => { + const safeForLegacyEncoding = await hasUnsafeTextProjectionDatabase(); + const summary = options.summary === true; + const rows = await retryIdempotentDatabaseOperation(async () => { + const query = db + .select( + summary + ? { + ...heartbeatRunSummaryListColumns, + ...heartbeatRunListContextColumns, + } + : safeForLegacyEncoding + ? { + ...heartbeatRunListColumns, + error: sql`NULL`.as("error"), + ...heartbeatRunListContextColumns, + } + : { + ...heartbeatRunListColumns, + ...heartbeatRunListContextColumns, + ...heartbeatRunListResultColumns, + }, + ) + .from(heartbeatRuns) + .where( + agentId + ? and( + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.agentId, agentId), + ) + : eq(heartbeatRuns.companyId, companyId), + ) + .orderBy(desc(heartbeatRuns.createdAt)); + + return limit ? await query.limit(limit) : await query; + }); + return rows.map((row) => { + const { + contextIssueId, + contextTaskId, + contextTaskKey, + contextCommentId, + contextWakeCommentId, + contextWakeReason, + contextWakeSource, + contextWakeTriggerDetail, + resultSummary, + resultResult, + resultMessage, + resultError, + resultTotalCostUsd, + resultCostUsd, + resultCostUsdCamel, + ...rest + } = row as typeof row & { + resultSummary?: string | null; + resultResult?: string | null; + resultMessage?: string | null; + resultError?: string | null; + resultTotalCostUsd?: string | null; + resultCostUsd?: string | null; + resultCostUsdCamel?: string | null; + }; + + return { + ...rest, + contextSnapshot: summarizeHeartbeatRunContextSnapshot({ + issueId: contextIssueId, + taskId: contextTaskId, + taskKey: contextTaskKey, + commentId: contextCommentId, + wakeCommentId: contextWakeCommentId, + wakeReason: contextWakeReason, + wakeSource: contextWakeSource, + wakeTriggerDetail: contextWakeTriggerDetail, + }), + resultJson: + safeForLegacyEncoding || summary + ? null + : summarizeHeartbeatRunListResultJson({ + summary: resultSummary, + result: resultResult, + message: resultMessage, + error: resultError, + totalCostUsd: resultTotalCostUsd, + costUsd: resultCostUsd, + costUsdCamel: resultCostUsdCamel, + }), + }; + }); + }, + + getRuntimeStateWithSessions: async (agentId: string) => { + const state = await getRuntimeState(agentId); + const agent = await getAgent(agentId); + if (!agent) return null; + const ensured = state ?? (await ensureRuntimeState(agent)); + const latestTaskSession = await db + .select() + .from(agentTaskSessions) + .where( + and( + eq(agentTaskSessions.companyId, agent.companyId), + eq(agentTaskSessions.agentId, agent.id), + ), + ) + .orderBy(desc(agentTaskSessions.updatedAt)) + .limit(1) + .then((rows) => rows[0] ?? null); + return { + ...ensured, + sessionDisplayId: + latestTaskSession?.sessionDisplayId ?? ensured.sessionId, + sessionParamsJson: latestTaskSession?.sessionParamsJson ?? null, + }; + }, + + listTaskSessions: async (agentId: string) => { + const agent = await getAgent(agentId); + if (!agent) throw notFound("Agent not found"); + + return db + .select() + .from(agentTaskSessions) + .where( + and( + eq(agentTaskSessions.companyId, agent.companyId), + eq(agentTaskSessions.agentId, agentId), + ), + ) + .orderBy( + desc(agentTaskSessions.updatedAt), + desc(agentTaskSessions.createdAt), + ); + }, + + resetRuntimeSession: async ( + agentId: string, + opts?: { taskKey?: string | null }, + ) => { + const agent = await getAgent(agentId); + if (!agent) throw notFound("Agent not found"); + await ensureRuntimeState(agent); + const taskKey = readNonEmptyString(opts?.taskKey); + const clearedTaskSessions = await clearTaskSessions( + agent.companyId, + agent.id, + taskKey + ? { + taskKey, + adapterType: agent.adapterType, + includeIssueAliases: true, + } + : undefined, + ); + const runtimePatch: Partial = { + sessionId: null, + lastError: null, + updatedAt: new Date(), + }; + if (!taskKey) { + runtimePatch.stateJson = {}; + } + + const updated = await db + .update(agentRuntimeState) + .set(runtimePatch) + .where(eq(agentRuntimeState.agentId, agentId)) + .returning() + .then((rows) => rows[0] ?? null); + + if (!updated) return null; + return { + ...updated, + sessionDisplayId: null, + sessionParamsJson: null, + clearedTaskSessions, + }; + }, + + listEvents: (runId: string, afterSeq = 0, limit = 200) => + db + .select() + .from(heartbeatRunEvents) + .where( + and( + eq(heartbeatRunEvents.runId, runId), + gt(heartbeatRunEvents.seq, afterSeq), + ), + ) + .orderBy(asc(heartbeatRunEvents.seq)) + .limit(Math.max(1, Math.min(limit, 1000))), + + getRetryExhaustedReason: async (runId: string) => { + const row = await db + .select({ + message: heartbeatRunEvents.message, + }) + .from(heartbeatRunEvents) + .where( + and( + eq(heartbeatRunEvents.runId, runId), + eq(heartbeatRunEvents.eventType, "lifecycle"), + sql`${heartbeatRunEvents.message} like 'Bounded retry exhausted%'`, + ), + ) + .orderBy(desc(heartbeatRunEvents.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + return row?.message ?? null; + }, + + getRunIssueSummary: async (runId: string) => { + const [run] = await db + .select(heartbeatRunIssueSummaryColumns) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .limit(1); + return run ?? null; + }, + + getActiveRunForAgent: async (agentId: string) => { + const [run] = await db + .select() + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.agentId, agentId), + eq(heartbeatRuns.status, "running"), + ), + ) + .orderBy(desc(heartbeatRuns.startedAt)) + .limit(1); + return run ?? null; + }, + + getActiveRunIssueSummaryForAgent: async (agentId: string) => { + const [run] = await db + .select(heartbeatRunIssueSummaryColumns) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.agentId, agentId), + eq(heartbeatRuns.status, "running"), + ), + ) + .orderBy(desc(heartbeatRuns.startedAt)) + .limit(1); + return run ?? null; + }, + }; +}