diff --git a/packages/db/src/migrations/0065_liveness_recovery_dedupe.sql b/packages/db/src/migrations/0069_liveness_recovery_dedupe.sql similarity index 100% rename from packages/db/src/migrations/0065_liveness_recovery_dedupe.sql rename to packages/db/src/migrations/0069_liveness_recovery_dedupe.sql diff --git a/packages/db/src/migrations/0070_active_run_output_watchdog.sql b/packages/db/src/migrations/0070_active_run_output_watchdog.sql new file mode 100644 index 00000000000..47d68fe5c2f --- /dev/null +++ b/packages/db/src/migrations/0070_active_run_output_watchdog.sql @@ -0,0 +1,70 @@ +ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "last_output_at" timestamp with time zone; +--> statement-breakpoint +ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "last_output_seq" integer DEFAULT 0 NOT NULL; +--> statement-breakpoint +ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "last_output_stream" text; +--> statement-breakpoint +ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "last_output_bytes" bigint; +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "heartbeat_runs_company_status_last_output_idx" + ON "heartbeat_runs" USING btree ("company_id","status","last_output_at"); +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "heartbeat_runs_company_status_process_started_idx" + ON "heartbeat_runs" USING btree ("company_id","status","process_started_at"); +--> statement-breakpoint +CREATE TABLE IF NOT EXISTS "heartbeat_run_watchdog_decisions" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "company_id" uuid NOT NULL, + "run_id" uuid NOT NULL, + "evaluation_issue_id" uuid, + "decision" text NOT NULL, + "snoozed_until" timestamp with time zone, + "reason" text, + "created_by_agent_id" uuid, + "created_by_user_id" text, + "created_by_run_id" uuid, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); +--> statement-breakpoint +DO $$ BEGIN + ALTER TABLE "heartbeat_run_watchdog_decisions" ADD CONSTRAINT "heartbeat_run_watchdog_decisions_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE no action ON UPDATE no action; +EXCEPTION + WHEN duplicate_object THEN null; +END $$; +--> statement-breakpoint +DO $$ BEGIN + ALTER TABLE "heartbeat_run_watchdog_decisions" ADD CONSTRAINT "heartbeat_run_watchdog_decisions_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE cascade ON UPDATE no action; +EXCEPTION + WHEN duplicate_object THEN null; +END $$; +--> statement-breakpoint +DO $$ BEGIN + ALTER TABLE "heartbeat_run_watchdog_decisions" ADD CONSTRAINT "heartbeat_run_watchdog_decisions_evaluation_issue_id_issues_id_fk" FOREIGN KEY ("evaluation_issue_id") REFERENCES "public"."issues"("id") ON DELETE set null ON UPDATE no action; +EXCEPTION + WHEN duplicate_object THEN null; +END $$; +--> statement-breakpoint +DO $$ BEGIN + ALTER TABLE "heartbeat_run_watchdog_decisions" ADD CONSTRAINT "heartbeat_run_watchdog_decisions_created_by_agent_id_agents_id_fk" FOREIGN KEY ("created_by_agent_id") REFERENCES "public"."agents"("id") ON DELETE set null ON UPDATE no action; +EXCEPTION + WHEN duplicate_object THEN null; +END $$; +--> statement-breakpoint +DO $$ BEGIN + ALTER TABLE "heartbeat_run_watchdog_decisions" ADD CONSTRAINT "heartbeat_run_watchdog_decisions_created_by_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("created_by_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE set null ON UPDATE no action; +EXCEPTION + WHEN duplicate_object THEN null; +END $$; +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "heartbeat_run_watchdog_decisions_company_run_created_idx" + ON "heartbeat_run_watchdog_decisions" USING btree ("company_id","run_id","created_at"); +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "heartbeat_run_watchdog_decisions_company_run_snooze_idx" + ON "heartbeat_run_watchdog_decisions" USING btree ("company_id","run_id","snoozed_until"); +--> statement-breakpoint +CREATE UNIQUE INDEX IF NOT EXISTS "issues_active_stale_run_evaluation_uq" + ON "issues" USING btree ("company_id","origin_kind","origin_id") + WHERE "origin_kind" = 'stale_active_run_evaluation' + AND "origin_id" IS NOT NULL + AND "hidden_at" IS NULL + AND "status" NOT IN ('done', 'cancelled'); diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index d429f9e0be5..765875c4463 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -484,6 +484,20 @@ "when": 1776959400000, "tag": "0068_environment_local_driver_unique", "breakpoints": true + }, + { + "idx": 69, + "version": "7", + "when": 1776780003000, + "tag": "0069_liveness_recovery_dedupe", + "breakpoints": true + }, + { + "idx": 70, + "version": "7", + "when": 1776780004000, + "tag": "0070_active_run_output_watchdog", + "breakpoints": true } ] } diff --git a/packages/db/src/schema/heartbeat_run_watchdog_decisions.ts b/packages/db/src/schema/heartbeat_run_watchdog_decisions.ts new file mode 100644 index 00000000000..46e17a5133d --- /dev/null +++ b/packages/db/src/schema/heartbeat_run_watchdog_decisions.ts @@ -0,0 +1,34 @@ +import { index, pgTable, text, timestamp, uuid } from "drizzle-orm/pg-core"; +import { agents } from "./agents.js"; +import { companies } from "./companies.js"; +import { heartbeatRuns } from "./heartbeat_runs.js"; +import { issues } from "./issues.js"; + +export const heartbeatRunWatchdogDecisions = pgTable( + "heartbeat_run_watchdog_decisions", + { + id: uuid("id").primaryKey().defaultRandom(), + companyId: uuid("company_id").notNull().references(() => companies.id), + runId: uuid("run_id").notNull().references(() => heartbeatRuns.id, { onDelete: "cascade" }), + evaluationIssueId: uuid("evaluation_issue_id").references(() => issues.id, { onDelete: "set null" }), + decision: text("decision").notNull(), + snoozedUntil: timestamp("snoozed_until", { withTimezone: true }), + reason: text("reason"), + createdByAgentId: uuid("created_by_agent_id").references(() => agents.id, { onDelete: "set null" }), + createdByUserId: text("created_by_user_id"), + createdByRunId: uuid("created_by_run_id").references(() => heartbeatRuns.id, { onDelete: "set null" }), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + companyRunCreatedIdx: index("heartbeat_run_watchdog_decisions_company_run_created_idx").on( + table.companyId, + table.runId, + table.createdAt, + ), + companyRunSnoozeIdx: index("heartbeat_run_watchdog_decisions_company_run_snooze_idx").on( + table.companyId, + table.runId, + table.snoozedUntil, + ), + }), +); diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 4010e2b6f47..c975892a1e0 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -34,6 +34,10 @@ export const heartbeatRuns = pgTable( processPid: integer("process_pid"), processGroupId: integer("process_group_id"), processStartedAt: timestamp("process_started_at", { withTimezone: true }), + lastOutputAt: timestamp("last_output_at", { withTimezone: true }), + lastOutputSeq: integer("last_output_seq").notNull().default(0), + lastOutputStream: text("last_output_stream"), + lastOutputBytes: bigint("last_output_bytes", { mode: "number" }), retryOfRunId: uuid("retry_of_run_id").references((): AnyPgColumn => heartbeatRuns.id, { onDelete: "set null", }), @@ -64,5 +68,15 @@ export const heartbeatRuns = pgTable( table.livenessState, table.createdAt, ), + companyStatusLastOutputIdx: index("heartbeat_runs_company_status_last_output_idx").on( + table.companyId, + table.status, + table.lastOutputAt, + ), + companyStatusProcessStartedIdx: index("heartbeat_runs_company_status_process_started_idx").on( + table.companyId, + table.status, + table.processStartedAt, + ), }), ); diff --git a/packages/db/src/schema/index.ts b/packages/db/src/schema/index.ts index fcf2ecc9471..322a326dfc0 100644 --- a/packages/db/src/schema/index.ts +++ b/packages/db/src/schema/index.ts @@ -53,6 +53,7 @@ export { documentRevisions } from "./document_revisions.js"; export { issueDocuments } from "./issue_documents.js"; export { heartbeatRuns } from "./heartbeat_runs.js"; export { heartbeatRunEvents } from "./heartbeat_run_events.js"; +export { heartbeatRunWatchdogDecisions } from "./heartbeat_run_watchdog_decisions.js"; export { costEvents } from "./cost_events.js"; export { financeEvents } from "./finance_events.js"; export { approvals } from "./approvals.js"; diff --git a/packages/db/src/schema/issues.ts b/packages/db/src/schema/issues.ts index 35ae509515f..d1e50656152 100644 --- a/packages/db/src/schema/issues.ts +++ b/packages/db/src/schema/issues.ts @@ -107,5 +107,13 @@ export const issues = pgTable( and ${table.hiddenAt} is null and ${table.status} not in ('done', 'cancelled')`, ), + activeStaleRunEvaluationIdx: uniqueIndex("issues_active_stale_run_evaluation_uq") + .on(table.companyId, table.originKind, table.originId) + .where( + sql`${table.originKind} = 'stale_active_run_evaluation' + and ${table.originId} is not null + and ${table.hiddenAt} is null + and ${table.status} not in ('done', 'cancelled')`, + ), }), ); diff --git a/packages/shared/src/constants.ts b/packages/shared/src/constants.ts index 6fe2a21e165..e997365095a 100644 --- a/packages/shared/src/constants.ts +++ b/packages/shared/src/constants.ts @@ -162,7 +162,7 @@ export const ISSUE_THREAD_INTERACTION_CONTINUATION_POLICIES = [ export type IssueThreadInteractionContinuationPolicy = (typeof ISSUE_THREAD_INTERACTION_CONTINUATION_POLICIES)[number]; -export const ISSUE_ORIGIN_KINDS = ["manual", "routine_execution"] as const; +export const ISSUE_ORIGIN_KINDS = ["manual", "routine_execution", "stale_active_run_evaluation"] as const; export type BuiltInIssueOriginKind = (typeof ISSUE_ORIGIN_KINDS)[number]; export type PluginIssueOriginKind = `plugin:${string}`; export type IssueOriginKind = BuiltInIssueOriginKind | PluginIssueOriginKind; diff --git a/packages/shared/src/types/heartbeat.ts b/packages/shared/src/types/heartbeat.ts index 98073b5e986..5aee87af5e7 100644 --- a/packages/shared/src/types/heartbeat.ts +++ b/packages/shared/src/types/heartbeat.ts @@ -37,6 +37,10 @@ export interface HeartbeatRun { processPid: number | null; processGroupId?: number | null; processStartedAt: Date | null; + lastOutputAt: Date | null; + lastOutputSeq: number; + lastOutputStream: "stdout" | "stderr" | null; + lastOutputBytes: number | null; retryOfRunId: string | null; processLossRetryCount: number; scheduledRetryAt?: Date | null; @@ -51,6 +55,28 @@ export interface HeartbeatRun { contextSnapshot: Record | null; createdAt: Date; updatedAt: Date; + outputSilence?: HeartbeatRunOutputSilence; +} + +export type HeartbeatRunOutputSilenceLevel = + | "not_applicable" + | "ok" + | "suspicious" + | "critical" + | "snoozed"; + +export interface HeartbeatRunOutputSilence { + lastOutputAt: Date | string | null; + lastOutputSeq: number; + lastOutputStream: "stdout" | "stderr" | null; + silenceStartedAt: Date | string | null; + silenceAgeMs: number | null; + level: HeartbeatRunOutputSilenceLevel; + suspicionThresholdMs: number; + criticalThresholdMs: number; + snoozedUntil: Date | string | null; + evaluationIssueId: string | null; + evaluationIssueIdentifier: string | null; } export interface AgentWakeupSkipped { diff --git a/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts b/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts new file mode 100644 index 00000000000..d8594e1dbfc --- /dev/null +++ b/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts @@ -0,0 +1,251 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, + companies, + createDb, + heartbeatRunWatchdogDecisions, + heartbeatRuns, + issueRelations, + issues, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { + ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS, + ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS, + heartbeatService, +} from "../services/heartbeat.ts"; + +const mockAdapterExecute = vi.hoisted(() => + vi.fn(async () => ({ + exitCode: 0, + signal: null, + timedOut: false, + errorMessage: null, + summary: "Acknowledged stale-run evaluation.", + provider: "test", + model: "test-model", + })), +); + +vi.mock("../telemetry.ts", () => ({ + getTelemetryClient: () => ({ track: vi.fn() }), +})); + +vi.mock("@paperclipai/shared/telemetry", async () => { + const actual = await vi.importActual( + "@paperclipai/shared/telemetry", + ); + return { + ...actual, + trackAgentFirstHeartbeat: vi.fn(), + }; +}); + +vi.mock("../adapters/index.ts", async () => { + const actual = await vi.importActual("../adapters/index.ts"); + return { + ...actual, + getServerAdapter: vi.fn(() => ({ + supportsLocalAgentJwt: false, + execute: mockAdapterExecute, + })), + }; +}); + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres active-run output watchdog tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("active-run output watchdog", () => { + let tempDb: Awaited> | null = null; + let db: ReturnType; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-active-run-output-watchdog-"); + db = createDb(tempDb.connectionString); + }, 30_000); + + afterEach(async () => { + for (let attempt = 0; attempt < 100; attempt += 1) { + const activeRuns = await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(sql`${heartbeatRuns.status} in ('queued', 'running')`); + if (activeRuns.length === 0) break; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + await db.execute(sql.raw(`TRUNCATE TABLE "companies" CASCADE`)); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seedRunningRun(opts: { now: Date; ageMs: number; withOutput?: boolean }) { + const companyId = randomUUID(); + const managerId = randomUUID(); + const coderId = randomUUID(); + const issueId = randomUUID(); + const runId = randomUUID(); + const issuePrefix = `W${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + const startedAt = new Date(opts.now.getTime() - opts.ageMs); + const lastOutputAt = opts.withOutput ? new Date(opts.now.getTime() - 5 * 60 * 1000) : null; + + await db.insert(companies).values({ + id: companyId, + name: "Watchdog Co", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values([ + { + id: managerId, + companyId, + name: "CTO", + role: "cto", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + { + id: coderId, + companyId, + name: "Coder", + role: "engineer", + status: "running", + reportsTo: managerId, + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }, + ]); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Long running implementation", + status: "in_progress", + priority: "medium", + assigneeAgentId: coderId, + issueNumber: 1, + identifier: `${issuePrefix}-1`, + updatedAt: startedAt, + createdAt: startedAt, + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: coderId, + status: "running", + invocationSource: "assignment", + triggerDetail: "system", + startedAt, + processStartedAt: startedAt, + lastOutputAt, + lastOutputSeq: opts.withOutput ? 3 : 0, + lastOutputStream: opts.withOutput ? "stdout" : null, + contextSnapshot: { issueId }, + stdoutExcerpt: "OPENAI_API_KEY=sk-test-secret-value should not leak", + logBytes: 0, + }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + return { companyId, managerId, coderId, issueId, runId, issuePrefix }; + } + + it("creates one medium-priority evaluation issue for a suspicious silent run", async () => { + const now = new Date("2026-04-22T20:00:00.000Z"); + const { companyId, managerId, runId } = await seedRunningRun({ + now, + ageMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + 60_000, + }); + const heartbeat = heartbeatService(db); + + const first = await heartbeat.scanSilentActiveRuns({ now, companyId }); + const second = await heartbeat.scanSilentActiveRuns({ now, companyId }); + + expect(first.created).toBe(1); + expect(second.created).toBe(0); + expect(second.existing).toBe(1); + + const evaluations = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stale_active_run_evaluation"))); + expect(evaluations).toHaveLength(1); + expect(["todo", "in_progress"]).toContain(evaluations[0]?.status); + expect(evaluations[0]).toMatchObject({ + priority: "medium", + assigneeAgentId: managerId, + originId: runId, + originFingerprint: `stale_active_run:${companyId}:${runId}`, + }); + expect(evaluations[0]?.description).toContain("Decision Checklist"); + expect(evaluations[0]?.description).not.toContain("sk-test-secret-value"); + }); + + it("raises critical stale-run evaluations and blocks the source issue", async () => { + const now = new Date("2026-04-22T20:00:00.000Z"); + const { companyId, issueId } = await seedRunningRun({ + now, + ageMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + 60_000, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.scanSilentActiveRuns({ now, companyId }); + + expect(result.created).toBe(1); + const [evaluation] = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stale_active_run_evaluation"))); + expect(evaluation?.priority).toBe("high"); + + const [blocker] = await db + .select() + .from(issueRelations) + .where(and(eq(issueRelations.companyId, companyId), eq(issueRelations.relatedIssueId, issueId))); + expect(blocker?.issueId).toBe(evaluation?.id); + + const [source] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(source?.status).toBe("blocked"); + }); + + it("skips snoozed runs and healthy noisy runs", async () => { + const now = new Date("2026-04-22T20:00:00.000Z"); + const stale = await seedRunningRun({ + now, + ageMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + 60_000, + }); + const noisy = await seedRunningRun({ + now, + ageMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + 60_000, + withOutput: true, + }); + await db.insert(heartbeatRunWatchdogDecisions).values({ + companyId: stale.companyId, + runId: stale.runId, + decision: "snooze", + snoozedUntil: new Date(now.getTime() + 60 * 60 * 1000), + reason: "Intentional quiet run", + }); + const heartbeat = heartbeatService(db); + + const staleResult = await heartbeat.scanSilentActiveRuns({ now, companyId: stale.companyId }); + const noisyResult = await heartbeat.scanSilentActiveRuns({ now, companyId: noisy.companyId }); + + expect(staleResult).toMatchObject({ created: 0, snoozed: 1 }); + expect(noisyResult).toMatchObject({ scanned: 0, created: 0 }); + }); +}); diff --git a/server/src/index.ts b/server/src/index.ts index a63ca877faa..e656217f265 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -685,6 +685,12 @@ export async function startServer(): Promise { logger.warn({ ...reconciled }, "startup issue-graph liveness reconciliation created escalations"); } }) + .then(async () => { + const scanned = await heartbeat.scanSilentActiveRuns(); + if (scanned.created > 0 || scanned.escalated > 0) { + logger.warn({ ...scanned }, "startup active-run output watchdog created review work"); + } + }) .catch((err) => { logger.error({ err }, "startup heartbeat recovery failed"); }); @@ -737,6 +743,12 @@ export async function startServer(): Promise { logger.warn({ ...reconciled }, "periodic issue-graph liveness reconciliation created escalations"); } }) + .then(async () => { + const scanned = await heartbeat.scanSilentActiveRuns(); + if (scanned.created > 0 || scanned.escalated > 0) { + logger.warn({ ...scanned }, "periodic active-run output watchdog created review work"); + } + }) .catch((err) => { logger.error({ err }, "periodic heartbeat recovery failed"); }); diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index 758182c81f7..07d4d0614fe 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -2,7 +2,7 @@ import { Router, type Request, type Response } from "express"; import { generateKeyPairSync, randomUUID } from "node:crypto"; import path from "node:path"; import type { Db } from "@paperclipai/db"; -import { agents as agentsTable, companies, heartbeatRuns, issues as issuesTable } from "@paperclipai/db"; +import { agents as agentsTable, companies, heartbeatRunWatchdogDecisions, heartbeatRuns, issues as issuesTable } from "@paperclipai/db"; import { and, desc, eq, inArray, not, sql } from "drizzle-orm"; import { agentSkillSyncSchema, @@ -2482,6 +2482,7 @@ export function agentRoutes(db: Db) { const columns = { id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, status: heartbeatRuns.status, invocationSource: heartbeatRuns.invocationSource, triggerDetail: heartbeatRuns.triggerDetail, @@ -2496,6 +2497,11 @@ export function agentRoutes(db: Db) { continuationAttempt: heartbeatRuns.continuationAttempt, lastUsefulActionAt: heartbeatRuns.lastUsefulActionAt, nextAction: heartbeatRuns.nextAction, + lastOutputAt: heartbeatRuns.lastOutputAt, + lastOutputSeq: heartbeatRuns.lastOutputSeq, + lastOutputStream: heartbeatRuns.lastOutputStream, + lastOutputBytes: heartbeatRuns.lastOutputBytes, + processStartedAt: heartbeatRuns.processStartedAt, issueId: sql`${heartbeatRuns.contextSnapshot} ->> 'issueId'`.as("issueId"), }; @@ -2527,11 +2533,18 @@ export function agentRoutes(db: Db) { .orderBy(desc(heartbeatRuns.createdAt)) .limit(minCount - liveRuns.length); - res.json([...liveRuns, ...recentRuns]); + const rows = [...liveRuns, ...recentRuns]; + res.json(await Promise.all(rows.map(async (run) => ({ + ...run, + outputSilence: await heartbeat.buildRunOutputSilence(run), + })))); return; } - res.json(liveRuns); + res.json(await Promise.all(liveRuns.map(async (run) => ({ + ...run, + outputSilence: await heartbeat.buildRunOutputSilence(run), + })))); }); router.get("/heartbeat-runs/:runId", async (req, res) => { @@ -2545,7 +2558,7 @@ export function agentRoutes(db: Db) { const retryExhaustedReason = await heartbeat.getRetryExhaustedReason(runId); res.json( redactCurrentUserValue( - { ...run, retryExhaustedReason }, + { ...run, retryExhaustedReason, outputSilence: await heartbeat.buildRunOutputSilence(run) }, await getCurrentUserRedactionOptions(), ), ); @@ -2575,6 +2588,90 @@ export function agentRoutes(db: Db) { res.json(run); }); + router.post("/heartbeat-runs/:runId/watchdog-decisions", async (req, res) => { + const runId = req.params.runId as string; + const existing = await heartbeat.getRun(runId); + if (!existing) { + res.status(404).json({ error: "Heartbeat run not found" }); + return; + } + assertCompanyAccess(req, existing.companyId); + const decision = typeof req.body?.decision === "string" ? req.body.decision : ""; + if (!["snooze", "continue", "dismissed_false_positive"].includes(decision)) { + res.status(400).json({ error: "Unsupported watchdog decision" }); + return; + } + const evaluationIssueId = typeof req.body?.evaluationIssueId === "string" ? req.body.evaluationIssueId : null; + const reason = typeof req.body?.reason === "string" ? req.body.reason.slice(0, 4000) : null; + const snoozedUntil = decision === "snooze" + ? new Date(String(req.body?.snoozedUntil ?? "")) + : null; + if (decision === "snooze" && (!snoozedUntil || Number.isNaN(snoozedUntil.getTime()) || snoozedUntil <= new Date())) { + res.status(400).json({ error: "snoozedUntil must be a future ISO datetime" }); + return; + } + + let evaluationIssue: { id: string; assigneeAgentId: string | null; companyId: string } | null = null; + if (evaluationIssueId) { + evaluationIssue = await db + .select({ + id: issuesTable.id, + assigneeAgentId: issuesTable.assigneeAgentId, + companyId: issuesTable.companyId, + }) + .from(issuesTable) + .where(and(eq(issuesTable.id, evaluationIssueId), eq(issuesTable.companyId, existing.companyId))) + .then((rows) => rows[0] ?? null); + if (!evaluationIssue) { + res.status(404).json({ error: "Evaluation issue not found" }); + return; + } + } + + const boardActor = req.actor.type === "board"; + const assignedRecoveryOwner = + req.actor.type === "agent" && + req.actor.agentId && + evaluationIssue?.assigneeAgentId === req.actor.agentId; + if (!boardActor && !assignedRecoveryOwner) { + throw forbidden("Only the board or the assigned recovery owner can record watchdog decisions"); + } + + const [row] = await db + .insert(heartbeatRunWatchdogDecisions) + .values({ + companyId: existing.companyId, + runId: existing.id, + evaluationIssueId, + decision, + snoozedUntil, + reason, + createdByAgentId: req.actor.type === "agent" ? req.actor.agentId ?? null : null, + createdByUserId: req.actor.type === "board" ? req.actor.userId ?? null : null, + createdByRunId: typeof req.body?.createdByRunId === "string" ? req.body.createdByRunId : null, + }) + .returning(); + + await logActivity(db, { + companyId: existing.companyId, + actorType: req.actor.type === "agent" ? "agent" : "user", + actorId: req.actor.type === "agent" ? req.actor.agentId ?? "agent" : req.actor.userId ?? "board", + agentId: req.actor.type === "agent" ? req.actor.agentId ?? null : null, + runId: existing.id, + action: decision === "snooze" ? "heartbeat.watchdog_snoozed" : "heartbeat.watchdog_decision_recorded", + entityType: "heartbeat_run", + entityId: existing.id, + details: { + decision, + evaluationIssueId, + snoozedUntil: snoozedUntil?.toISOString() ?? null, + reason, + }, + }); + + res.json(row); + }); + router.get("/heartbeat-runs/:runId/events", async (req, res) => { const runId = req.params.runId as string; const run = await heartbeat.getRun(runId); @@ -2680,6 +2777,11 @@ export function agentRoutes(db: Db) { continuationAttempt: heartbeatRuns.continuationAttempt, lastUsefulActionAt: heartbeatRuns.lastUsefulActionAt, nextAction: heartbeatRuns.nextAction, + lastOutputAt: heartbeatRuns.lastOutputAt, + lastOutputSeq: heartbeatRuns.lastOutputSeq, + lastOutputStream: heartbeatRuns.lastOutputStream, + lastOutputBytes: heartbeatRuns.lastOutputBytes, + processStartedAt: heartbeatRuns.processStartedAt, }) .from(heartbeatRuns) .innerJoin(agentsTable, eq(heartbeatRuns.agentId, agentsTable.id)) @@ -2692,7 +2794,10 @@ export function agentRoutes(db: Db) { ) .orderBy(desc(heartbeatRuns.createdAt)); - res.json(liveRuns); + res.json(await Promise.all(liveRuns.map(async (run) => ({ + ...run, + outputSilence: await heartbeat.buildRunOutputSilence({ ...run, companyId: issue.companyId }), + })))); }); router.get("/issues/:issueId/active-run", async (req, res) => { @@ -2740,6 +2845,7 @@ export function agentRoutes(db: Db) { agentId: agent.id, agentName: agent.name, adapterType: agent.adapterType, + outputSilence: await heartbeat.buildRunOutputSilence({ ...run, companyId: issue.companyId }), }); }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index f23ea06770d..f6f62677c6c 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -27,10 +27,12 @@ import { agentTaskSessions, agentWakeupRequests, activityLog, + companies, companySkills as companySkillsTable, documentRevisions, issueDocuments, heartbeatRunEvents, + heartbeatRunWatchdogDecisions, heartbeatRuns, issueComments, issueRelations, @@ -158,6 +160,11 @@ const CANCELLABLE_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retr const HEARTBEAT_RUN_TERMINAL_STATUSES = ["succeeded", "failed", "cancelled", "timed_out"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const; const ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_MIN_STALE_MS = 24 * 60 * 60 * 1000; +export const ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS = 60 * 60 * 1000; +export const ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000; +export const ACTIVE_RUN_OUTPUT_PROGRESS_FLUSH_INTERVAL_MS = 60 * 1000; +const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = "stale_active_run_evaluation"; +const ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES = 8 * 1024; export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS = [ 2 * 60 * 1000, 10 * 60 * 1000, @@ -565,6 +572,10 @@ const heartbeatRunListColumns = { 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, @@ -677,11 +688,16 @@ const heartbeatRunIssueSummaryColumns = { finishedAt: heartbeatRuns.finishedAt, createdAt: heartbeatRuns.createdAt, agentId: heartbeatRuns.agentId, + 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: sql`${heartbeatRuns.contextSnapshot} ->> 'issueId'`.as("issueId"), } as const; @@ -4456,6 +4472,551 @@ export function heartbeatService(db: Db) { readNonEmptyString(nestedContext.taskId); } + function staleActiveRunOriginFingerprint(companyId: string, runId: string) { + return `stale_active_run:${companyId}:${runId}`; + } + + function silenceStartedAtForRun(run: Pick) { + return run.lastOutputAt ?? run.processStartedAt ?? run.startedAt ?? run.createdAt ?? null; + } + + function silenceAgeMsForRun(run: Pick, now = new Date()) { + const startedAt = silenceStartedAtForRun(run); + return startedAt ? Math.max(0, now.getTime() - startedAt.getTime()) : null; + } + + async function latestActiveOutputSnooze(companyId: string, runId: string, now = new Date()) { + const [row] = await db + .select() + .from(heartbeatRunWatchdogDecisions) + .where( + and( + eq(heartbeatRunWatchdogDecisions.companyId, companyId), + eq(heartbeatRunWatchdogDecisions.runId, runId), + eq(heartbeatRunWatchdogDecisions.decision, "snooze"), + gt(heartbeatRunWatchdogDecisions.snoozedUntil, now), + ), + ) + .orderBy(desc(heartbeatRunWatchdogDecisions.createdAt)) + .limit(1); + return row ?? null; + } + + async function findOpenStaleRunEvaluation(companyId: string, runId: string) { + const [row] = await db + .select({ + id: issues.id, + identifier: issues.identifier, + status: issues.status, + priority: issues.priority, + assigneeAgentId: issues.assigneeAgentId, + updatedAt: issues.updatedAt, + }) + .from(issues) + .where( + and( + eq(issues.companyId, companyId), + eq(issues.originKind, STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND), + eq(issues.originId, runId), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ) + .limit(1); + return row ?? null; + } + + async function buildRunOutputSilence( + run: Pick< + typeof heartbeatRuns.$inferSelect, + "id" | "companyId" | "status" | "lastOutputAt" | "lastOutputSeq" | "lastOutputStream" | "processStartedAt" | "startedAt" | "createdAt" + >, + now = new Date(), + ) { + const [snooze, evaluation] = await Promise.all([ + latestActiveOutputSnooze(run.companyId, run.id, now), + findOpenStaleRunEvaluation(run.companyId, run.id), + ]); + const silenceStartedAt = silenceStartedAtForRun(run); + const silenceAgeMs = run.status === "running" ? silenceAgeMsForRun(run, now) : null; + const level = run.status !== "running" + ? "not_applicable" + : snooze + ? "snoozed" + : (silenceAgeMs ?? 0) >= ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + ? "critical" + : (silenceAgeMs ?? 0) >= ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + ? "suspicious" + : "ok"; + return { + lastOutputAt: run.lastOutputAt ?? null, + lastOutputSeq: run.lastOutputSeq ?? 0, + lastOutputStream: (run.lastOutputStream === "stdout" || run.lastOutputStream === "stderr") + ? run.lastOutputStream + : null, + silenceStartedAt, + silenceAgeMs, + level, + suspicionThresholdMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS, + criticalThresholdMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS, + snoozedUntil: snooze?.snoozedUntil ?? null, + evaluationIssueId: evaluation?.id ?? null, + evaluationIssueIdentifier: evaluation?.identifier ?? null, + }; + } + + function redactWatchdogEvidenceText(value: string, currentUserRedactionOptions: Awaited>) { + return redactCurrentUserText(value, currentUserRedactionOptions) + .replace(/\b(sk-[a-zA-Z0-9_-]{12,})\b/g, "[REDACTED_API_KEY]") + .replace(/\b([A-Za-z0-9_]*TOKEN[A-Za-z0-9_]*\s*=\s*)[^\s]+/gi, "$1[REDACTED]") + .replace(/\b([A-Za-z0-9_]*KEY[A-Za-z0-9_]*\s*=\s*)[^\s]+/gi, "$1[REDACTED]") + .replace(/\b([A-Za-z0-9_]*SECRET[A-Za-z0-9_]*\s*=\s*)[^\s]+/gi, "$1[REDACTED]"); + } + + function truncateEvidenceText(value: string, maxChars = 4000) { + if (value.length <= maxChars) return value; + return `${value.slice(value.length - maxChars)}\n[truncated earlier evidence]`; + } + + function formatDuration(ms: number | null) { + if (ms === null) return "unknown"; + const minutes = Math.floor(ms / 60_000); + if (minutes < 60) return `${minutes}m`; + const hours = Math.floor(minutes / 60); + const remainingMinutes = minutes % 60; + return remainingMinutes > 0 ? `${hours}h ${remainingMinutes}m` : `${hours}h`; + } + + function issueUiLink(issue: { identifier: string | null; id: string }, prefix: string) { + const label = issue.identifier ?? issue.id; + return `[${label}](/${prefix}/issues/${label})`; + } + + function runUiLink(run: { id: string; agentId: string }, prefix: string) { + return `[${run.id}](/${prefix}/agents/${run.agentId}/runs/${run.id})`; + } + + async function readRunLogTailForEvidence(run: typeof heartbeatRuns.$inferSelect) { + if (!run.logStore || !run.logRef || !run.logBytes) return ""; + try { + const offset = Math.max(0, run.logBytes - ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES); + const result = await runLogStore.read( + { store: run.logStore as "local_file", logRef: run.logRef }, + { offset, limitBytes: ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES }, + ); + return result.content; + } catch (err) { + logger.warn({ err, runId: run.id }, "failed to read stale-run watchdog evidence tail"); + return ""; + } + } + + async function resolveStaleRunSourceIssue(run: typeof heartbeatRuns.$inferSelect) { + const issueId = issueIdFromRunContext(run.contextSnapshot); + if (!issueId) return null; + const [issue] = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId), isNull(issues.hiddenAt))) + .limit(1); + return issue ?? null; + } + + function isAgentInvokable(agent: typeof agents.$inferSelect | null | undefined) { + return Boolean(agent && !["paused", "terminated", "pending_approval"].includes(agent.status)); + } + + async function resolveStaleRunOwnerAgentId(input: { + run: typeof heartbeatRuns.$inferSelect; + runningAgent: typeof agents.$inferSelect; + sourceIssue: typeof issues.$inferSelect | null; + }) { + const candidateIds: string[] = []; + if (input.sourceIssue?.assigneeAgentId) { + const sourceAssignee = await getAgent(input.sourceIssue.assigneeAgentId); + if (sourceAssignee?.reportsTo) candidateIds.push(sourceAssignee.reportsTo); + } + if (input.runningAgent.reportsTo) candidateIds.push(input.runningAgent.reportsTo); + const roleCandidates = await db + .select() + .from(agents) + .where(and(eq(agents.companyId, input.run.companyId), inArray(agents.role, ["cto", "ceo"]))) + .orderBy(sql`case when ${agents.role} = 'cto' then 0 else 1 end`, asc(agents.createdAt)); + candidateIds.push(...roleCandidates.map((agent) => agent.id)); + + const seen = new Set(); + for (const agentId of candidateIds) { + if (seen.has(agentId)) continue; + seen.add(agentId); + const candidate = await getAgent(agentId); + if (!candidate || candidate.companyId !== input.run.companyId) continue; + const budgetBlock = await budgets.getInvocationBlock(input.run.companyId, candidate.id, { + issueId: input.sourceIssue?.id ?? null, + projectId: input.sourceIssue?.projectId ?? null, + }); + if (isAgentInvokable(candidate) && !budgetBlock) return candidate.id; + } + + return null; + } + + async function collectStaleRunEvidence(input: { + run: typeof heartbeatRuns.$inferSelect; + runningAgent: typeof agents.$inferSelect; + sourceIssue: typeof issues.$inferSelect | null; + prefix: string; + now: Date; + }) { + const [tail, recentEvents, childIssues, blockers] = await Promise.all([ + readRunLogTailForEvidence(input.run), + db + .select({ + eventType: heartbeatRunEvents.eventType, + level: heartbeatRunEvents.level, + message: heartbeatRunEvents.message, + createdAt: heartbeatRunEvents.createdAt, + }) + .from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.companyId, input.run.companyId), eq(heartbeatRunEvents.runId, input.run.id))) + .orderBy(desc(heartbeatRunEvents.id)) + .limit(8), + input.sourceIssue + ? db + .select({ id: issues.id, identifier: issues.identifier, title: issues.title, status: issues.status }) + .from(issues) + .where(and(eq(issues.companyId, input.run.companyId), eq(issues.parentId, input.sourceIssue.id), isNull(issues.hiddenAt))) + .orderBy(desc(issues.updatedAt)) + .limit(8) + : Promise.resolve([]), + input.sourceIssue + ? db + .select({ id: issues.id, identifier: issues.identifier, title: issues.title, status: issues.status }) + .from(issueRelations) + .innerJoin(issues, eq(issueRelations.issueId, issues.id)) + .where( + and( + eq(issueRelations.companyId, input.run.companyId), + eq(issueRelations.relatedIssueId, input.sourceIssue.id), + eq(issueRelations.type, "blocks"), + ), + ) + .limit(8) + : Promise.resolve([]), + ]); + const currentUserRedactionOptions = await getCurrentUserRedactionOptions(); + const safeTail = truncateEvidenceText(redactWatchdogEvidenceText(tail, currentUserRedactionOptions)); + const silenceAgeMs = silenceAgeMsForRun(input.run, input.now); + return { + safeTail, + silenceAgeMs, + recentEvents: recentEvents.reverse().map((event) => ({ + eventType: event.eventType, + level: event.level, + createdAt: event.createdAt.toISOString(), + message: event.message ? truncateEvidenceText(redactWatchdogEvidenceText(event.message, currentUserRedactionOptions), 300) : null, + })), + childIssues, + blockers, + }; + } + + function buildStaleRunEvaluationDescription(input: { + run: typeof heartbeatRuns.$inferSelect; + runningAgent: typeof agents.$inferSelect; + sourceIssue: typeof issues.$inferSelect | null; + prefix: string; + evidence: Awaited>; + level: "suspicious" | "critical"; + now: Date; + }) { + const sourceIssue = input.sourceIssue + ? issueUiLink({ identifier: input.sourceIssue.identifier, id: input.sourceIssue.id }, input.prefix) + : "none"; + const recentEvents = input.evidence.recentEvents.length > 0 + ? input.evidence.recentEvents.map((event) => + `- ${event.createdAt} \`${event.eventType}\`${event.level ? ` ${event.level}` : ""}: ${event.message ?? "(no message)"}`, + ).join("\n") + : "- none"; + const childIssues = input.evidence.childIssues.length > 0 + ? input.evidence.childIssues.map((issue) => + `- ${issueUiLink({ identifier: issue.identifier, id: issue.id }, input.prefix)} \`${issue.status}\`: ${issue.title}`, + ).join("\n") + : "- none detected"; + const blockers = input.evidence.blockers.length > 0 + ? input.evidence.blockers.map((issue) => + `- ${issueUiLink({ identifier: issue.identifier, id: issue.id }, input.prefix)} \`${issue.status}\`: ${issue.title}`, + ).join("\n") + : "- none detected"; + return [ + `Paperclip detected ${input.level} output silence on an active heartbeat run.`, + "", + "## Run", + "", + `- Run: ${runUiLink(input.run, input.prefix)}`, + `- Agent: ${input.runningAgent.name} (${input.runningAgent.adapterType})`, + `- Invocation: ${input.run.invocationSource}${input.run.triggerDetail ? ` / ${input.run.triggerDetail}` : ""}`, + `- Source issue: ${sourceIssue}`, + `- Started at: ${input.run.startedAt?.toISOString() ?? "unknown"}`, + `- Process started at: ${input.run.processStartedAt?.toISOString() ?? "unknown"}`, + `- Last output at: ${input.run.lastOutputAt?.toISOString() ?? "none recorded"}`, + `- Last output sequence: ${input.run.lastOutputSeq ?? 0}`, + `- Silent for: ${formatDuration(input.evidence.silenceAgeMs)}`, + `- Thresholds: suspicious after ${formatDuration(ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS)}, critical after ${formatDuration(ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS)}`, + `- Process metadata: pid \`${input.run.processPid ?? "unknown"}\`, process group \`${input.run.processGroupId ?? "unknown"}\`, in-memory handle \`${runningProcesses.has(input.run.id) ? "yes" : "no"}\``, + "", + "## Last Output Excerpt", + "", + input.evidence.safeTail ? `\`\`\`text\n${input.evidence.safeTail}\n\`\`\`` : "_No run-log tail was available._", + "", + "## Recent Run Events", + "", + recentEvents, + "", + "## Related Work", + "", + "Active child issues:", + childIssues, + "", + "Current source blockers:", + blockers, + "", + "## Decision Checklist", + "", + "- Continue or snooze if the run is intentionally quiet.", + "- Ask the run owner for context if work may be delegated outside the transcript.", + "- Preserve artifacts, branch state, and useful output before cancellation.", + "- Cancel or recover through the explicit run recovery controls when authorized.", + "- Close this issue as a false positive only after recording the reason.", + ].join("\n"); + } + + function isUniqueStaleRunEvaluationConflict(error: unknown) { + if (!error || typeof error !== "object") return false; + const maybe = error as { code?: string; constraint?: string; message?: string }; + return maybe.code === "23505" && + ( + maybe.constraint === "issues_active_stale_run_evaluation_uq" || + typeof maybe.message === "string" && maybe.message.includes("issues_active_stale_run_evaluation_uq") + ); + } + + async function ensureSourceIssueBlockedByStaleEvaluation(input: { + sourceIssue: typeof issues.$inferSelect | null; + evaluationIssue: { id: string; identifier: string | null }; + run: typeof heartbeatRuns.$inferSelect; + }) { + if (!input.sourceIssue || ["done", "cancelled"].includes(input.sourceIssue.status)) return false; + const blockerIds = await existingBlockerIssueIds(input.sourceIssue.companyId, input.sourceIssue.id); + if (blockerIds.includes(input.evaluationIssue.id)) return false; + const nextBlockerIds = [...blockerIds, input.evaluationIssue.id]; + await issuesSvc.update(input.sourceIssue.id, { + ...(input.sourceIssue.status === "blocked" ? {} : { status: "blocked" }), + blockedByIssueIds: nextBlockerIds, + }); + await issuesSvc.addComment(input.sourceIssue.id, [ + "Paperclip detected critical output silence on this issue's active run.", + "", + `- Evaluation issue: ${input.evaluationIssue.identifier ?? input.evaluationIssue.id}`, + `- Run: \`${input.run.id}\``, + "", + "This blocks the source issue on the explicit review task without cancelling the active process.", + ].join("\n"), { runId: input.run.id }); + await logActivity(db, { + companyId: input.sourceIssue.companyId, + actorType: "system", + actorId: "system", + agentId: null, + runId: input.run.id, + action: "heartbeat.output_stale_escalated", + entityType: "issue", + entityId: input.sourceIssue.id, + details: { + source: "heartbeat.scan_silent_active_runs", + evaluationIssueId: input.evaluationIssue.id, + blockerIssueIds: nextBlockerIds, + }, + }); + return true; + } + + async function createOrUpdateStaleRunEvaluation(input: { + run: typeof heartbeatRuns.$inferSelect; + now: Date; + }) { + const runningAgent = await getAgent(input.run.agentId); + if (!runningAgent || runningAgent.companyId !== input.run.companyId) return { kind: "skipped" as const }; + const sourceIssue = await resolveStaleRunSourceIssue(input.run); + const [company] = await db + .select({ issuePrefix: companies.issuePrefix }) + .from(companies) + .where(eq(companies.id, input.run.companyId)) + .limit(1); + const prefix = company?.issuePrefix ?? "PAP"; + const evidence = await collectStaleRunEvidence({ + run: input.run, + runningAgent, + sourceIssue, + prefix, + now: input.now, + }); + const level = (evidence.silenceAgeMs ?? 0) >= ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS ? "critical" : "suspicious"; + const existing = await findOpenStaleRunEvaluation(input.run.companyId, input.run.id); + if (existing) { + if (level === "critical" && existing.priority !== "high") { + await issuesSvc.update(existing.id, { + priority: "high", + }); + await issuesSvc.addComment(existing.id, [ + "Critical output silence threshold crossed.", + "", + `- Run: \`${input.run.id}\``, + `- Silent for: ${formatDuration(evidence.silenceAgeMs)}`, + `- Last output at: ${input.run.lastOutputAt?.toISOString() ?? "none recorded"}`, + ].join("\n"), { runId: input.run.id }); + await ensureSourceIssueBlockedByStaleEvaluation({ + sourceIssue, + evaluationIssue: existing, + run: input.run, + }); + return { kind: "escalated" as const, evaluationIssueId: existing.id }; + } + if (level === "critical") { + await ensureSourceIssueBlockedByStaleEvaluation({ + sourceIssue, + evaluationIssue: existing, + run: input.run, + }); + } + return { kind: "existing" as const, evaluationIssueId: existing.id }; + } + + const ownerAgentId = await resolveStaleRunOwnerAgentId({ run: input.run, runningAgent, sourceIssue }); + const description = buildStaleRunEvaluationDescription({ + run: input.run, + runningAgent, + sourceIssue, + prefix, + evidence, + level, + now: input.now, + }); + let evaluation: Awaited>; + try { + evaluation = await issuesSvc.create(input.run.companyId, { + title: `Review silent active run for ${runningAgent.name}`, + description, + status: "todo", + priority: level === "critical" ? "high" : "medium", + parentId: sourceIssue && !["done", "cancelled"].includes(sourceIssue.status) ? sourceIssue.id : null, + projectId: sourceIssue?.projectId ?? null, + goalId: sourceIssue?.goalId ?? null, + billingCode: sourceIssue?.billingCode ?? null, + assigneeAgentId: ownerAgentId, + originKind: STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND, + originId: input.run.id, + originRunId: input.run.id, + originFingerprint: staleActiveRunOriginFingerprint(input.run.companyId, input.run.id), + }); + } catch (error) { + if (!isUniqueStaleRunEvaluationConflict(error)) throw error; + const raced = await findOpenStaleRunEvaluation(input.run.companyId, input.run.id); + if (!raced) throw error; + return { kind: "existing" as const, evaluationIssueId: raced.id }; + } + + await logActivity(db, { + companyId: input.run.companyId, + actorType: "system", + actorId: "system", + agentId: ownerAgentId, + runId: input.run.id, + action: "heartbeat.output_stale_detected", + entityType: "issue", + entityId: evaluation.id, + details: { + source: "heartbeat.scan_silent_active_runs", + level, + sourceIssueId: sourceIssue?.id ?? null, + silenceAgeMs: evidence.silenceAgeMs, + lastOutputAt: input.run.lastOutputAt?.toISOString() ?? null, + }, + }); + if (level === "critical") { + await ensureSourceIssueBlockedByStaleEvaluation({ + sourceIssue, + evaluationIssue: evaluation, + run: input.run, + }); + } + if (ownerAgentId) { + await enqueueWakeup(ownerAgentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { + issueId: evaluation.id, + staleRunId: input.run.id, + sourceIssueId: sourceIssue?.id ?? null, + }, + requestedByActorType: "system", + requestedByActorId: null, + contextSnapshot: { + issueId: evaluation.id, + taskId: evaluation.id, + wakeReason: "issue_assigned", + source: STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND, + staleRunId: input.run.id, + sourceIssueId: sourceIssue?.id ?? null, + }, + }); + } + return { kind: "created" as const, evaluationIssueId: evaluation.id }; + } + + async function scanSilentActiveRuns(opts?: { now?: Date; companyId?: string }) { + const now = opts?.now ?? new Date(); + const suspicionBefore = new Date(now.getTime() - ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS); + const candidates = await db + .select() + .from(heartbeatRuns) + .where( + and( + opts?.companyId ? eq(heartbeatRuns.companyId, opts.companyId) : undefined, + eq(heartbeatRuns.status, "running"), + sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.processStartedAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) <= ${suspicionBefore.toISOString()}::timestamptz`, + ), + ) + .orderBy(asc(heartbeatRuns.createdAt)) + .limit(100); + + const result = { + scanned: candidates.length, + created: 0, + existing: 0, + escalated: 0, + snoozed: 0, + skipped: 0, + evaluationIssueIds: [] as string[], + }; + + for (const run of candidates) { + if (await latestActiveOutputSnooze(run.companyId, run.id, now)) { + result.snoozed += 1; + continue; + } + const outcome = await createOrUpdateStaleRunEvaluation({ run, now }); + if (outcome.kind === "created") result.created += 1; + else if (outcome.kind === "existing") result.existing += 1; + else if (outcome.kind === "escalated") result.escalated += 1; + else result.skipped += 1; + if ("evaluationIssueId" in outcome && outcome.evaluationIssueId) { + result.evaluationIssueIds.push(outcome.evaluationIssueId); + } + } + + return result; + } + async function collectIssueGraphLivenessFindings() { const [issueRows, relationRows, agentRows, activeRunRows, activeIssueRunRows, wakeRows] = await Promise.all([ db @@ -5885,6 +6446,38 @@ export function heartbeatService(db: Db) { let handle: RunLogHandle | null = null; let stdoutExcerpt = ""; let stderrExcerpt = ""; + let outputSeq = Number(run.lastOutputSeq ?? 0); + let lastOutputFlushAt: Date | null = run.lastOutputAt ?? null; + const outputProgressState: { + pending: { + at: Date; + seq: number; + stream: "stdout" | "stderr"; + bytes: number; + } | null; + } = { pending: null }; + let persistedLogBytes = Number(run.logBytes ?? 0); + const flushOutputProgress = async (opts?: { force?: boolean }) => { + const pendingOutputProgress = outputProgressState.pending; + if (!pendingOutputProgress) return; + const shouldFlush = + opts?.force === true || + !lastOutputFlushAt || + pendingOutputProgress.at.getTime() - lastOutputFlushAt.getTime() >= ACTIVE_RUN_OUTPUT_PROGRESS_FLUSH_INTERVAL_MS; + if (!shouldFlush) return; + await db + .update(heartbeatRuns) + .set({ + lastOutputAt: pendingOutputProgress.at, + lastOutputSeq: pendingOutputProgress.seq, + lastOutputStream: pendingOutputProgress.stream, + lastOutputBytes: pendingOutputProgress.bytes, + updatedAt: new Date(), + }) + .where(eq(heartbeatRuns.id, run.id)); + lastOutputFlushAt = pendingOutputProgress.at; + outputProgressState.pending = null; + }; try { const startedAt = run.startedAt ?? new Date(); const runningWithSession = await db @@ -5951,13 +6544,23 @@ export function heartbeatService(db: Db) { if (stream === "stderr") stderrExcerpt = appendExcerpt(stderrExcerpt, sanitizedChunk); const ts = new Date().toISOString(); + let appendedBytes = 0; if (handle) { - await runLogStore.append(handle, { + appendedBytes = await runLogStore.append(handle, { stream, chunk: sanitizedChunk, ts, }); + persistedLogBytes += appendedBytes; } + outputSeq += 1; + outputProgressState.pending = { + at: new Date(ts), + seq: outputSeq, + stream, + bytes: persistedLogBytes, + }; + await flushOutputProgress(); const payloadChunk = sanitizedChunk.length > MAX_LIVE_LOG_CHUNK_BYTES @@ -6187,6 +6790,11 @@ export function heartbeatService(db: Db) { if (handle) { logSummary = await runLogStore.finalize(handle); } + const finalLogBytes = logSummary?.bytes; + if (outputProgressState.pending && typeof finalLogBytes === "number") { + outputProgressState.pending.bytes = finalLogBytes; + } + await flushOutputProgress({ force: true }); const status = outcome === "succeeded" @@ -6334,6 +6942,13 @@ export function heartbeatService(db: Db) { logger.warn({ err: finalizeErr, runId }, "failed to finalize run log after error"); } } + const finalLogBytes = logSummary?.bytes; + if (outputProgressState.pending && typeof finalLogBytes === "number") { + outputProgressState.pending.bytes = finalLogBytes; + } + await flushOutputProgress({ force: true }).catch((flushErr) => { + logger.warn({ err: flushErr, runId }, "failed to flush run output progress after error"); + }); const failedRun = await setRunStatus(run.id, "failed", { error: message, @@ -7981,6 +8596,10 @@ export function heartbeatService(db: Db) { reconcileIssueGraphLiveness, + scanSilentActiveRuns, + + buildRunOutputSilence, + tickTimers: async (now = new Date()) => { const allAgents = await db.select().from(agents); let checked = 0; diff --git a/server/src/services/run-log-store.ts b/server/src/services/run-log-store.ts index 6429c5f22d9..9c6a814287d 100644 --- a/server/src/services/run-log-store.ts +++ b/server/src/services/run-log-store.ts @@ -32,7 +32,7 @@ export interface RunLogStore { append( handle: RunLogHandle, event: { stream: "stdout" | "stderr" | "system"; chunk: string; ts: string }, - ): Promise; + ): Promise; finalize(handle: RunLogHandle): Promise; read(handle: RunLogHandle, opts?: RunLogReadOptions): Promise; } @@ -107,14 +107,16 @@ function createLocalFileRunLogStore(basePath: string): RunLogStore { }, async append(handle, event) { - if (handle.store !== "local_file") return; + if (handle.store !== "local_file") return 0; const absPath = resolveWithin(basePath, handle.logRef); const line = JSON.stringify({ ts: event.ts, stream: event.stream, chunk: event.chunk, }); - await fs.appendFile(absPath, `${line}\n`, "utf8"); + const persisted = `${line}\n`; + await fs.appendFile(absPath, persisted, "utf8"); + return Buffer.byteLength(persisted, "utf8"); }, async finalize(handle) { @@ -153,4 +155,3 @@ export function getRunLogStore() { cachedStore = createLocalFileRunLogStore(basePath); return cachedStore; } -