feat: gate issue-bound heartbeat run data

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-01 22:54:01 +00:00
1 parent 6959c3d365
commit d18b769fe4
18 files changed
+700 -20

No files matched your search

+8
View File
@@ -340,6 +340,14 @@ No Docker or external database is required for this mode.
Grant lookups use a five-second cache by default; override it with `PAPERCLIP_ISSUE_PRIVACY_CACHE_TTL_MS` when testing revocation timing.
Issue-bound heartbeat runs inherit the same read predicate through
`heartbeat_runs.issue_id`. In enforce mode, direct run detail, transcript,
event, log, and workspace-operation reads return `404` to non-members. Company
run-history and live-run lists retain a metadata-only row with timing, status,
token usage, and cost for budget oversight; issue identifiers, summaries, and
run content are omitted. Runs without an issue binding keep company-level
maintenance-run visibility.
## Storage in Dev (Auto-Handled)
For local development, the default storage provider is `local_disk`, which persists uploaded images/attachments at:
+2
View File
@@ -288,6 +288,7 @@ An issue is readable when it is open, the actor is an implicit principal (`respo
- `id` uuid pk
- `company_id` uuid fk not null
- `agent_id` uuid fk not null
- `issue_id` uuid fk `issues.id` null; first-class binding for issue-scoped run confidentiality
- `invocation_source` enum: `scheduler | manual | callback`
- `status` enum: `queued | running | succeeded | failed | cancelled | timed_out`
- `started_at` timestamptz null
@@ -390,6 +391,7 @@ Operational policy:
- `cost_events(company_id, occurred_at)`
- `cost_events(company_id, agent_id, occurred_at)`
- `heartbeat_runs(company_id, agent_id, started_at desc)`
- `heartbeat_runs(company_id, issue_id, created_at)`
- `approvals(company_id, status, type)`
- `activity_log(company_id, created_at desc)`
- `assets(company_id, created_at desc)`
@@ -0,0 +1,96 @@
import { createHash, randomUUID } from "node:crypto";
import fs from "node:fs";
import { afterEach, describe, expect, it } from "vitest";
import postgres from "postgres";
import { applyPendingMigrations } from "./client.js";
import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
} from "./test-embedded-postgres.js";
const MIGRATION_FILE = "0199_heartbeat_run_issue_privacy.sql";
const cleanups: Array<() => Promise<void>> = [];
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
async function migrationHash() {
const content = await fs.promises.readFile(new URL(`./migrations/${MIGRATION_FILE}`, import.meta.url), "utf8");
return createHash("sha256").update(content).digest("hex");
}
afterEach(async () => {
while (cleanups.length > 0) await cleanups.pop()?.();
});
describeEmbeddedPostgres("heartbeat run issue privacy migration", () => {
it("backfills every valid context issue binding and preserves maintenance runs", async () => {
const database = await startEmbeddedPostgresTestDatabase("paperclip-run-issue-privacy-");
cleanups.push(database.cleanup);
const sql = postgres(database.connectionString, { max: 1, onnotice: () => {} });
cleanups.push(async () => sql.end());
await sql`DELETE FROM "drizzle"."__drizzle_migrations" WHERE "hash" = ${await migrationHash()}`;
const companyId = randomUUID();
const agentId = randomUUID();
const firstIssueId = randomUUID();
const secondIssueId = randomUUID();
const firstRunId = randomUUID();
const secondRunId = randomUUID();
const maintenanceRunId = randomUUID();
await sql`
INSERT INTO "companies" ("id", "name", "issue_prefix")
VALUES (${companyId}, 'Run Privacy', 'RPR')
`;
await sql`
INSERT INTO "agents" ("id", "company_id", "name", "role", "adapter_type", "adapter_config")
VALUES (${agentId}, ${companyId}, 'Runner', 'engineer', 'process', '{}'::jsonb)
`;
await sql`
INSERT INTO "issues" ("id", "company_id", "title", "identifier")
VALUES
(${firstIssueId}, ${companyId}, 'First private task', 'RPR-1'),
(${secondIssueId}, ${companyId}, 'Second private task', 'RPR-2')
`;
await sql`
INSERT INTO "heartbeat_runs" (
"id", "company_id", "agent_id", "status", "issue_id", "context_snapshot"
)
VALUES
(${firstRunId}, ${companyId}, ${agentId}, 'succeeded', NULL, ${sql.json({ issueId: firstIssueId })}),
(${secondRunId}, ${companyId}, ${agentId}, 'failed', NULL, ${sql.json({ issueId: secondIssueId })}),
(${maintenanceRunId}, ${companyId}, ${agentId}, 'succeeded', NULL, ${sql.json({ wakeReason: "heartbeat_timer" })})
`;
await applyPendingMigrations(database.connectionString);
const rows = await sql<{ id: string; issue_id: string | null }[]>`
SELECT "id", "issue_id"
FROM "heartbeat_runs"
WHERE "id" IN (${firstRunId}, ${secondRunId}, ${maintenanceRunId})
ORDER BY "id"
`;
const issueByRun = new Map(rows.map((row) => [row.id, row.issue_id]));
expect(issueByRun.get(firstRunId)).toBe(firstIssueId);
expect(issueByRun.get(secondRunId)).toBe(secondIssueId);
expect(issueByRun.get(maintenanceRunId)).toBeNull();
const [parity] = await sql<{ expected_count: number; populated_count: number }[]>`
SELECT
count(*) FILTER (WHERE "context_snapshot" ? 'issueId')::int AS "expected_count",
count(*) FILTER (WHERE "context_snapshot" ? 'issueId' AND "issue_id" IS NOT NULL)::int AS "populated_count"
FROM "heartbeat_runs"
WHERE "id" IN (${firstRunId}, ${secondRunId}, ${maintenanceRunId})
`;
expect(parity).toEqual({ expected_count: 2, populated_count: 2 });
const indexes = await sql<{ indexname: string }[]>`
SELECT "indexname"
FROM "pg_indexes"
WHERE "schemaname" = 'public'
AND "indexname" = 'heartbeat_runs_company_issue_created_idx'
`;
expect(indexes).toHaveLength(1);
}, 30_000);
});
@@ -0,0 +1,68 @@
ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "issue_id" uuid;
--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "heartbeat_runs_context_issue_backfill_idx"
ON "heartbeat_runs" USING btree ("id")
WHERE "issue_id" IS NULL AND "context_snapshot" ? 'issueId';
--> statement-breakpoint
UPDATE "heartbeat_runs" AS "run"
SET "issue_id" = "issue"."id"
FROM "issues" AS "issue"
WHERE "run"."issue_id" IS NULL
AND "run"."context_snapshot" ? 'issueId'
AND "issue"."id" = CASE
WHEN "run"."context_snapshot"->>'issueId' ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$'
THEN ("run"."context_snapshot"->>'issueId')::uuid
ELSE NULL
END
AND "issue"."company_id" = "run"."company_id";
--> statement-breakpoint
DO $$
DECLARE
"expected_count" bigint;
"populated_count" bigint;
BEGIN
SELECT count(*)
INTO "expected_count"
FROM "heartbeat_runs" AS "run"
INNER JOIN "issues" AS "issue"
ON "issue"."company_id" = "run"."company_id"
AND "issue"."id" = CASE
WHEN "run"."context_snapshot"->>'issueId' ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$'
THEN ("run"."context_snapshot"->>'issueId')::uuid
ELSE NULL
END
WHERE "run"."context_snapshot" ? 'issueId'
AND "run"."context_snapshot"->>'issueId' ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$';
SELECT count(*)
INTO "populated_count"
FROM "heartbeat_runs" AS "run"
WHERE "run"."context_snapshot" ? 'issueId'
AND "run"."issue_id" = CASE
WHEN "run"."context_snapshot"->>'issueId' ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$'
THEN ("run"."context_snapshot"->>'issueId')::uuid
ELSE NULL
END;
IF "expected_count" <> "populated_count" THEN
RAISE EXCEPTION 'heartbeat_runs.issue_id backfill parity failed: expected %, populated %',
"expected_count", "populated_count";
END IF;
END $$;
--> statement-breakpoint
DROP INDEX IF EXISTS "heartbeat_runs_context_issue_backfill_idx";
--> statement-breakpoint
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM "pg_constraint"
WHERE "conname" = 'heartbeat_runs_issue_id_issues_id_fk'
) THEN
ALTER TABLE "heartbeat_runs"
ADD CONSTRAINT "heartbeat_runs_issue_id_issues_id_fk"
FOREIGN KEY ("issue_id") REFERENCES "public"."issues"("id")
ON DELETE SET NULL ON UPDATE NO ACTION;
END IF;
END $$;
--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "heartbeat_runs_company_issue_created_idx"
ON "heartbeat_runs" USING btree ("company_id", "issue_id", "created_at");
@@ -1380,6 +1380,13 @@
"when": 1785542400000,
"tag": "0198_issue_privacy_foundation",
"breakpoints": true
},
{
"idx": 199,
"version": "7",
"when": 1785628800000,
"tag": "0199_heartbeat_run_issue_privacy",
"breakpoints": true
}
]
}
+7
View File
@@ -2,6 +2,7 @@ import { type AnyPgColumn, pgTable, uuid, text, timestamp, jsonb, index, integer
import { companies } from "./companies.js";
import { agents } from "./agents.js";
import { agentWakeupRequests } from "./agent_wakeup_requests.js";
import { issues } from "./issues.js";
export const heartbeatRuns = pgTable(
"heartbeat_runs",
@@ -9,6 +10,7 @@ export const heartbeatRuns = pgTable(
id: uuid("id").primaryKey().defaultRandom(),
companyId: uuid("company_id").notNull().references(() => companies.id),
agentId: uuid("agent_id").notNull().references(() => agents.id),
issueId: uuid("issue_id").references((): AnyPgColumn => issues.id, { onDelete: "set null" }),
invocationSource: text("invocation_source").notNull().default("on_demand"),
triggerDetail: text("trigger_detail"),
status: text("status").notNull().default("queued"),
@@ -64,6 +66,11 @@ export const heartbeatRuns = pgTable(
table.agentId,
table.startedAt,
),
companyIssueCreatedIdx: index("heartbeat_runs_company_issue_created_idx").on(
table.companyId,
table.issueId,
table.createdAt,
),
companyResponsibleUserIdx: index("heartbeat_runs_company_responsible_user_idx").on(
table.companyId,
table.responsibleUserId,
+5
View File
@@ -68,6 +68,7 @@ export interface HeartbeatRun {
id: string;
companyId: string;
agentId: string;
issueId: string | null;
invocationSource: HeartbeatInvocationSource;
triggerDetail: WakeupTriggerDetail | null;
status: HeartbeatRunStatus;
@@ -112,6 +113,10 @@ export interface HeartbeatRun {
contextSnapshot: Record<string, unknown> | null;
createdAt: Date;
updatedAt: Date;
/** Metadata-only company-list row for a run bound to an unreadable private issue. */
redacted?: boolean;
durationMs?: number | null;
costUsd?: number | null;
outputSilence?: HeartbeatRunOutputSilence;
/**
* Ephemeral, process-local current status message for an active run. Resolved
@@ -0,0 +1,281 @@
import { randomUUID } from "node:crypto";
import express from "express";
import request from "supertest";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
agents,
companies,
createDb,
heartbeatRunEvents,
heartbeatRuns,
issueAccessGrants,
issues,
workspaceOperations,
} from "@paperclipai/db";
import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
} from "./helpers/embedded-postgres.js";
import { errorHandler } from "../middleware/index.js";
import { activityRoutes } from "../routes/activity.js";
import { agentRoutes } from "../routes/agents.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
describeEmbeddedPostgres.sequential("heartbeat run privacy routes", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
const previousPrivacyMode = process.env.PAPERCLIP_ISSUE_PRIVACY_MODE;
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-run-privacy-routes-");
db = createDb(tempDb.connectionString);
process.env.PAPERCLIP_ISSUE_PRIVACY_MODE = "enforce";
}, 20_000);
afterEach(async () => {
await db.delete(workspaceOperations);
await db.delete(heartbeatRunEvents);
await db.delete(issueAccessGrants);
await db.delete(heartbeatRuns);
await db.delete(issues);
await db.delete(agents);
await db.delete(companies);
});
afterAll(async () => {
if (previousPrivacyMode === undefined) delete process.env.PAPERCLIP_ISSUE_PRIVACY_MODE;
else process.env.PAPERCLIP_ISSUE_PRIVACY_MODE = previousPrivacyMode;
await tempDb?.cleanup();
});
async function seedFixture() {
const companyId = randomUUID();
const ownerAgentId = randomUUID();
const grantedAgentId = randomUUID();
const otherAgentId = randomUUID();
const issueId = randomUUID();
const privateRunId = randomUUID();
const maintenanceRunId = randomUUID();
const operationId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Run Privacy",
issuePrefix: "RPR",
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values([
{
id: ownerAgentId,
companyId,
name: "Owner",
role: "engineer",
adapterType: "process",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
},
{
id: grantedAgentId,
companyId,
name: "Granted reviewer",
role: "security",
adapterType: "process",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
},
{
id: otherAgentId,
companyId,
name: "Other agent",
role: "engineer",
adapterType: "process",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
},
]);
await db.insert(issues).values({
id: issueId,
companyId,
identifier: "RPR-1",
title: "Confidential email triage",
description: "private@example.test",
visibility: "private",
privacyRootIssueId: issueId,
status: "in_progress",
assigneeAgentId: ownerAgentId,
});
await db.insert(issueAccessGrants).values({
issueId,
subjectType: "agent",
subjectId: grantedAgentId,
source: "explicit",
grantedByAgentId: ownerAgentId,
});
const startedAt = new Date("2026-07-31T12:00:00.000Z");
const finishedAt = new Date("2026-07-31T12:01:00.000Z");
await db.insert(heartbeatRuns).values([
{
id: privateRunId,
companyId,
agentId: ownerAgentId,
issueId,
invocationSource: "assignment",
status: "running",
startedAt,
finishedAt,
usageJson: { inputTokens: 11, cachedInputTokens: 2, outputTokens: 7 },
resultJson: { summary: "Read the confidential email", costUsd: 0.42 },
contextSnapshot: { issueId, title: "Confidential email triage" },
logStore: "local_file",
logRef: "missing-private-run-log.ndjson",
},
{
id: maintenanceRunId,
companyId,
agentId: ownerAgentId,
issueId: null,
invocationSource: "timer",
status: "succeeded",
startedAt,
finishedAt,
resultJson: { summary: "Maintenance complete", costUsd: 0.01 },
contextSnapshot: { wakeReason: "heartbeat_timer" },
},
]);
await db.insert(heartbeatRunEvents).values({
companyId,
runId: privateRunId,
agentId: ownerAgentId,
seq: 1,
eventType: "output",
stream: "stdout",
message: "private@example.test",
});
await db.insert(workspaceOperations).values({
id: operationId,
companyId,
heartbeatRunId: privateRunId,
issueId: null,
phase: "provision",
status: "succeeded",
command: "read private email",
stdoutExcerpt: "private@example.test",
logStore: "local_file",
logRef: "missing-private-operation-log.ndjson",
});
return { companyId, ownerAgentId, grantedAgentId, otherAgentId, issueId, privateRunId, maintenanceRunId, operationId };
}
function createApp(companyId: string, agentId: string) {
const app = express();
app.use(express.json());
app.use((req, _res, next) => {
req.actor = {
type: "agent",
agentId,
companyId,
source: "agent_jwt",
};
next();
});
app.use("/api", agentRoutes(db));
app.use("/api", activityRoutes(db));
app.use(errorHandler);
return app;
}
it("returns 404 for direct private-run content and metadata-only list stubs to a non-member", async () => {
const fixture = await seedFixture();
const app = createApp(fixture.companyId, fixture.otherAgentId);
for (const path of [
`/api/heartbeat-runs/${fixture.privateRunId}`,
`/api/heartbeat-runs/${fixture.privateRunId}/events`,
`/api/heartbeat-runs/${fixture.privateRunId}/log`,
`/api/heartbeat-runs/${fixture.privateRunId}/workspace-operations`,
`/api/workspace-operations/${fixture.operationId}/log`,
]) {
const response = await request(app).get(path);
expect(response.status, path).toBe(404);
expect(JSON.stringify(response.body)).not.toContain("private@example.test");
}
const history = await request(app).get(`/api/companies/${fixture.companyId}/heartbeat-runs`);
expect(history.status).toBe(200);
const privateStub = history.body.find((run: { id: string }) => run.id === fixture.privateRunId);
expect(privateStub).toMatchObject({
redacted: true,
durationMs: 60_000,
costUsd: 0.42,
usageJson: { inputTokens: 11, cachedInputTokens: 2, outputTokens: 7 },
});
expect(privateStub).not.toHaveProperty("issueId");
expect(privateStub).not.toHaveProperty("contextSnapshot");
expect(privateStub).not.toHaveProperty("resultJson");
expect(JSON.stringify(privateStub)).not.toContain("Confidential email triage");
expect(JSON.stringify(privateStub)).not.toContain("summary");
const summarizedHistory = await request(app)
.get(`/api/companies/${fixture.companyId}/heartbeat-runs?summary=true`);
const summarizedStub = summarizedHistory.body.find((run: { id: string }) => run.id === fixture.privateRunId);
expect(summarizedStub).toMatchObject({ redacted: true, durationMs: 60_000, costUsd: 0.42 });
expect(summarizedStub.usageJson).toMatchObject({ inputTokens: 11, outputTokens: 7 });
const live = await request(app).get(`/api/companies/${fixture.companyId}/live-runs`);
expect(live.status).toBe(200);
expect(live.body[0]).toMatchObject({ id: fixture.privateRunId, redacted: true, costUsd: 0.42 });
expect(live.body[0]).not.toHaveProperty("issueId");
const maintenance = await request(app).get(`/api/heartbeat-runs/${fixture.maintenanceRunId}`);
expect(maintenance.status).toBe(200);
expect(maintenance.body.resultJson.summary).toBe("Maintenance complete");
});
it("returns full private-run content to the owner and an explicitly granted agent", async () => {
const fixture = await seedFixture();
for (const agentId of [fixture.ownerAgentId, fixture.grantedAgentId]) {
const app = createApp(fixture.companyId, agentId);
const detail = await request(app).get(`/api/heartbeat-runs/${fixture.privateRunId}`);
expect(detail.status).toBe(200);
expect(detail.body).toMatchObject({ issueId: fixture.issueId });
expect(detail.body.resultJson.summary).toBe("Read the confidential email");
const events = await request(app).get(`/api/heartbeat-runs/${fixture.privateRunId}/events`);
expect(events.status).toBe(200);
expect(events.body[0].message).toBe("private@example.test");
const linkedIssues = await request(app).get(`/api/heartbeat-runs/${fixture.privateRunId}/issues`);
expect(linkedIssues.status).toBe(200);
expect(linkedIssues.body[0]).toMatchObject({ issueId: fixture.issueId, title: "Confidential email triage" });
}
});
it("returns no linked issue metadata to a non-member", async () => {
const fixture = await seedFixture();
const response = await request(createApp(fixture.companyId, fixture.otherAgentId))
.get(`/api/heartbeat-runs/${fixture.privateRunId}/issues`);
expect(response.status).toBe(200);
expect(response.body).toEqual([]);
});
it("logs would-deny decisions without enforcing them in shadow mode", async () => {
const fixture = await seedFixture();
process.env.PAPERCLIP_ISSUE_PRIVACY_MODE = "shadow";
try {
const response = await request(createApp(fixture.companyId, fixture.otherAgentId))
.get(`/api/heartbeat-runs/${fixture.privateRunId}`);
expect(response.status).toBe(200);
expect(response.body.resultJson.summary).toBe("Read the confidential email");
} finally {
process.env.PAPERCLIP_ISSUE_PRIVACY_MODE = "enforce";
}
});
});
+5
View File
@@ -10,6 +10,7 @@ import { sanitizeRecord } from "../redaction.js";
import { badRequest, forbidden } from "../errors.js";
import { agentActionAuditService } from "../services/agent-action-audit.js";
import { logActivity } from "../services/activity-log.js";
import { canActorReadHeartbeatRun } from "../services/heartbeat-run-privacy.js";
/** Max rows a single CSV export will stream (guards against runaway exports). */
const AUDIT_CSV_EXPORT_MAX_ROWS = 10_000;
@@ -299,6 +300,10 @@ export function activityRoutes(db: Db) {
}
assertCompanyAccess(req, run.companyId);
if (!(await assertCompanyScopeReadAllowed(req, res, run.companyId))) return;
if (!(await canActorReadHeartbeatRun(access, req.actor, run))) {
res.json([]);
return;
}
const result = await svc.issuesForRun(runId);
res.json(result);
});
+62 -11
View File
@@ -113,6 +113,10 @@ import {
changeConsentGateService,
touchesAgentProfileChangeConsentFields,
} from "../services/change-consent-gate.js";
import {
canActorReadHeartbeatRun,
redactHeartbeatRunListRow,
} from "../services/heartbeat-run-privacy.js";
const RUN_LOG_DEFAULT_LIMIT_BYTES = 256_000;
const RUN_LOG_MAX_LIMIT_BYTES = 1024 * 1024;
@@ -201,6 +205,27 @@ export function agentRoutes(
const instanceSettings = instanceSettingsService(db);
const strictSecretsMode = process.env.PAPERCLIP_SECRETS_STRICT_MODE === "true";
async function actorCanReadRun(req: Request, run: { companyId: string; issueId?: string | null }) {
return canActorReadHeartbeatRun(access, req.actor, run);
}
async function assertRunReadAllowed(
req: Request,
res: Response,
run: { companyId: string; issueId?: string | null },
notFoundMessage = "Heartbeat run not found",
) {
if (await actorCanReadRun(req, run)) return true;
res.status(404).json({ error: notFoundMessage });
return false;
}
async function serializeRunListRow<
T extends Record<string, unknown> & { companyId: string; issueId?: string | null },
>(req: Request, run: T) {
return await actorCanReadRun(req, run) ? run : redactHeartbeatRunListRow(run);
}
async function assertAgentEnvironmentSelection(
companyId: string,
adapterType: string,
@@ -3649,7 +3674,7 @@ export function agentRoutes(
const limit = limitParam ? Math.max(1, Math.min(1000, parseInt(limitParam, 10) || 200)) : undefined;
const summary = req.query.summary === "true" || req.query.summary === "1";
const runs = await heartbeat.list(companyId, agentId, limit, { summary });
res.json(runs);
res.json(await Promise.all(runs.map((run) => serializeRunListRow(req, run))));
});
router.get("/companies/:companyId/live-runs", async (req, res) => {
@@ -3689,7 +3714,16 @@ export function agentRoutes(
lastOutputStream: heartbeatRuns.lastOutputStream,
lastOutputBytes: heartbeatRuns.lastOutputBytes,
processStartedAt: heartbeatRuns.processStartedAt,
issueId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`.as("issueId"),
issueId: heartbeatRuns.issueId,
inputTokens: sql<number | null>`(${heartbeatRuns.usageJson} ->> 'inputTokens')::numeric`.as("inputTokens"),
cachedInputTokens: sql<number | null>`(${heartbeatRuns.usageJson} ->> 'cachedInputTokens')::numeric`.as("cachedInputTokens"),
outputTokens: sql<number | null>`(${heartbeatRuns.usageJson} ->> 'outputTokens')::numeric`.as("outputTokens"),
totalTokens: sql<number | null>`(${heartbeatRuns.usageJson} ->> 'totalTokens')::numeric`.as("totalTokens"),
costUsd: sql<number | null>`coalesce(
(${heartbeatRuns.resultJson} ->> 'costUsd')::numeric,
(${heartbeatRuns.resultJson} ->> 'cost_usd')::numeric,
(${heartbeatRuns.resultJson} ->> 'total_cost_usd')::numeric
)`.as("costUsd"),
};
const liveRunsQuery = db
@@ -3724,23 +3758,30 @@ export function agentRoutes(
.limit(targetRunCount - liveRuns.length);
const rows = [...liveRuns, ...recentRuns];
res.json(await Promise.all(rows.map(async (run) => ({
...heartbeat.decorateActiveRunStatus(run),
outputSilence: await heartbeat.buildRunOutputSilence(run),
}))));
res.json(await Promise.all(rows.map(async (run) => {
if (!(await actorCanReadRun(req, run))) return redactHeartbeatRunListRow(run);
return {
...heartbeat.decorateActiveRunStatus(run),
outputSilence: await heartbeat.buildRunOutputSilence(run),
};
})));
return;
}
res.json(await Promise.all(liveRuns.map(async (run) => ({
...heartbeat.decorateActiveRunStatus(run),
outputSilence: await heartbeat.buildRunOutputSilence(run),
}))));
res.json(await Promise.all(liveRuns.map(async (run) => {
if (!(await actorCanReadRun(req, run))) return redactHeartbeatRunListRow(run);
return {
...heartbeat.decorateActiveRunStatus(run),
outputSilence: await heartbeat.buildRunOutputSilence(run),
};
})));
});
router.get("/heartbeat-runs/:runId", async (req, res) => {
const runId = req.params.runId as string;
const run = await getAccessibleResource(req, res, heartbeat.getRun(runId), "Heartbeat run not found");
if (!run) return;
if (!(await assertRunReadAllowed(req, res, run))) return;
const retryExhaustedReason = await heartbeat.getRetryExhaustedReason(runId);
const decoratedRun = heartbeat.decorateActiveRunStatus(run);
res.json(
@@ -3809,6 +3850,7 @@ export function agentRoutes(
const runId = req.params.runId as string;
const run = await getAccessibleResource(req, res, heartbeat.getRun(runId), "Heartbeat run not found");
if (!run) return;
if (!(await assertRunReadAllowed(req, res, run))) return;
const afterSeq = Number(req.query.afterSeq ?? 0);
const limit = Number(req.query.limit ?? 200);
@@ -3827,6 +3869,7 @@ export function agentRoutes(
const runId = req.params.runId as string;
const run = await getAccessibleResource(req, res, heartbeat.getRunLogAccess(runId), "Heartbeat run not found");
if (!run) return;
if (!(await assertRunReadAllowed(req, res, run))) return;
const offset = Number(req.query.offset ?? 0);
const limitBytes = readRunLogLimitBytes(req.query.limitBytes);
@@ -3843,6 +3886,7 @@ export function agentRoutes(
const runId = req.params.runId as string;
const run = await getAccessibleResource(req, res, heartbeat.getRun(runId), "Heartbeat run not found");
if (!run) return;
if (!(await assertRunReadAllowed(req, res, run))) return;
const context = asRecord(run.contextSnapshot);
const executionWorkspaceId = asNonEmptyString(context?.executionWorkspaceId);
@@ -3854,6 +3898,13 @@ export function agentRoutes(
const operationId = req.params.operationId as string;
const operation = await getAccessibleResource(req, res, workspaceOperations.getById(operationId), "Workspace operation not found");
if (!operation) return;
const operationRun = operation.heartbeatRunId
? await heartbeat.getRunLogAccess(operation.heartbeatRunId)
: null;
if (!(await assertRunReadAllowed(req, res, {
companyId: operation.companyId,
issueId: operationRun?.issueId ?? operation.issueId ?? null,
}, "Workspace operation not found"))) return;
const offset = Number(req.query.offset ?? 0);
const limitBytes = readRunLogLimitBytes(req.query.limitBytes);
@@ -3910,7 +3961,7 @@ export function agentRoutes(
and(
eq(heartbeatRuns.companyId, issue.companyId),
inArray(heartbeatRuns.status, ["queued", "running"]),
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issue.id}`,
eq(heartbeatRuns.issueId, issue.id),
),
)
.orderBy(desc(heartbeatRuns.createdAt));
+3 -7
View File
@@ -415,7 +415,7 @@ export function activityService(db: Db) {
and(
eq(heartbeatRuns.companyId, companyId),
or(
sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`,
eq(heartbeatRuns.issueId, issueId),
sql`exists (
select 1
from ${activityLog}
@@ -522,7 +522,7 @@ export function activityService(db: Db) {
const run = await db
.select({
companyId: heartbeatRuns.companyId,
contextSnapshot: heartbeatRuns.contextSnapshot,
issueId: heartbeatRuns.issueId,
})
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
@@ -549,11 +549,7 @@ export function activityService(db: Db) {
)
.orderBy(issueIdAsText);
const context = run.contextSnapshot;
const contextIssueId =
context && typeof context === "object" && typeof (context as Record<string, unknown>).issueId === "string"
? ((context as Record<string, unknown>).issueId as string)
: null;
const contextIssueId = run.issueId;
if (!contextIssueId) return fromActivity;
if (fromActivity.some((issue) => issue.issueId === contextIssueId)) return fromActivity;
@@ -0,0 +1,127 @@
import type { AuthorizationActor, AuthorizationDecision } from "./authorization.js";
type RunIssueBinding = {
companyId: string;
issueId?: string | null;
};
export async function canActorReadHeartbeatRun(
access: {
decide(input: {
actor: AuthorizationActor;
action: "issue:read";
resource: {
type: "issue";
companyId: string;
issueId: string;
projectId: null;
parentIssueId: null;
assigneeAgentId: null;
assigneeUserId: null;
};
}): Promise<AuthorizationDecision>;
},
actor: AuthorizationActor,
run: RunIssueBinding,
): Promise<boolean> {
if (!run.issueId) return true;
const decision = await access.decide({
actor,
action: "issue:read",
resource: {
type: "issue",
companyId: run.companyId,
issueId: run.issueId,
projectId: null,
parentIssueId: null,
assigneeAgentId: null,
assigneeUserId: null,
},
});
return decision.allowed;
}
function asRecord(value: unknown): Record<string, unknown> | null {
return value && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown>
: null;
}
function finiteNumber(value: unknown): number | null {
if (typeof value === "number" && Number.isFinite(value)) return value;
if (typeof value === "string" && value.trim()) {
const parsed = Number(value);
if (Number.isFinite(parsed)) return parsed;
}
return null;
}
function firstFinite(...values: unknown[]): number | null {
for (const value of values) {
const parsed = finiteNumber(value);
if (parsed !== null) return parsed;
}
return null;
}
function pickTokenUsage(value: unknown) {
const usage = asRecord(value);
if (!usage) return null;
const picked = {
inputTokens: firstFinite(usage.inputTokens, usage.input_tokens),
cachedInputTokens: firstFinite(usage.cachedInputTokens, usage.cached_input_tokens),
outputTokens: firstFinite(usage.outputTokens, usage.output_tokens),
totalTokens: firstFinite(usage.totalTokens, usage.total_tokens),
};
return Object.values(picked).some((item) => item !== null) ? picked : null;
}
function runDurationMs(row: Record<string, unknown>) {
const startedAt = row.startedAt instanceof Date ? row.startedAt : new Date(String(row.startedAt ?? ""));
const finishedAt = row.finishedAt instanceof Date ? row.finishedAt : new Date(String(row.finishedAt ?? ""));
if (Number.isNaN(startedAt.getTime())) return null;
const endMs = Number.isNaN(finishedAt.getTime()) ? Date.now() : finishedAt.getTime();
return Math.max(0, endMs - startedAt.getTime());
}
/**
* Locked list-row disclosure for an issue-bound run the actor cannot read.
* Keep only identity, timing, status, and budget-oversight fields.
*/
export function redactHeartbeatRunListRow<T extends Record<string, unknown>>(row: T) {
const usage = asRecord(row.usageJson);
const result = asRecord(row.resultJson);
const costUsd = firstFinite(
row.costUsd,
result?.costUsd,
result?.cost_usd,
result?.total_cost_usd,
usage?.costUsd,
usage?.cost_usd,
usage?.total_cost_usd,
);
return {
id: row.id,
companyId: row.companyId,
agentId: row.agentId,
agentName: row.agentName,
adapterType: row.adapterType,
invocationSource: row.invocationSource,
status: row.status,
startedAt: row.startedAt ?? null,
finishedAt: row.finishedAt ?? null,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
durationMs: runDurationMs(row),
costUsd,
usageJson: pickTokenUsage(row.usageJson ?? {
inputTokens: row.inputTokens,
cachedInputTokens: row.cachedInputTokens,
outputTokens: row.outputTokens,
totalTokens: row.totalTokens,
}),
redacted: true,
};
}
+24 -2
View File
@@ -1938,9 +1938,19 @@ const heartbeatRunListColumns = {
id: heartbeatRuns.id,
companyId: heartbeatRuns.companyId,
agentId: heartbeatRuns.agentId,
issueId: heartbeatRuns.issueId,
invocationSource: heartbeatRuns.invocationSource,
triggerDetail: heartbeatRuns.triggerDetail,
status: heartbeatRuns.status,
inputTokens: sql<number | null>`(${heartbeatRuns.usageJson} ->> 'inputTokens')::numeric`.as("inputTokens"),
cachedInputTokens: sql<number | null>`(${heartbeatRuns.usageJson} ->> 'cachedInputTokens')::numeric`.as("cachedInputTokens"),
outputTokens: sql<number | null>`(${heartbeatRuns.usageJson} ->> 'outputTokens')::numeric`.as("outputTokens"),
totalTokens: sql<number | null>`(${heartbeatRuns.usageJson} ->> 'totalTokens')::numeric`.as("totalTokens"),
costUsd: sql<number | null>`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,
@@ -1995,7 +2005,10 @@ const heartbeatRunSummaryListColumns = {
} as const;
const heartbeatRunListContextColumns = {
contextIssueId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`.as("contextIssueId"),
contextIssueId: sql<string | null>`coalesce(
${heartbeatRuns.issueId}::text,
${heartbeatRuns.contextSnapshot} ->> 'issueId'
)`.as("contextIssueId"),
contextTaskId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'taskId'`.as("contextTaskId"),
contextTaskKey: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'taskKey'`.as("contextTaskKey"),
contextCommentId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'commentId'`.as("contextCommentId"),
@@ -2079,6 +2092,7 @@ const heartbeatRunSqlAsciiSafeColumns = {
const heartbeatRunLogAccessColumns = {
id: heartbeatRuns.id,
companyId: heartbeatRuns.companyId,
issueId: heartbeatRuns.issueId,
logStore: heartbeatRuns.logStore,
logRef: heartbeatRuns.logRef,
} as const;
@@ -2105,7 +2119,7 @@ const heartbeatRunIssueSummaryColumns = {
lastOutputSeq: heartbeatRuns.lastOutputSeq,
lastOutputStream: heartbeatRuns.lastOutputStream,
lastOutputBytes: heartbeatRuns.lastOutputBytes,
issueId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`.as("issueId"),
issueId: heartbeatRuns.issueId,
} as const;
function appendExcerpt(prev: string, chunk: string) {
@@ -9351,6 +9365,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
.values({
companyId: run.companyId,
agentId: run.agentId,
issueId,
invocationSource: "automation",
triggerDetail: "system",
status: "queued",
@@ -9601,6 +9616,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
.values({
companyId: run.companyId,
agentId: run.agentId,
issueId,
invocationSource: "automation",
triggerDetail: "system",
status: "queued",
@@ -10881,6 +10897,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
.values({
companyId: run.companyId,
agentId: run.agentId,
issueId,
invocationSource: "automation",
triggerDetail: "system",
status: "scheduled_retry",
@@ -15820,6 +15837,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
.values({
companyId: deferredAgent.companyId,
agentId: deferredAgent.id,
issueId: issue.id,
invocationSource: promotedSource,
triggerDetail: promotedTriggerDetail,
status: "queued",
@@ -15974,6 +15992,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
.values({
companyId: issue.companyId,
agentId: recoveryAgent.id,
issueId: issue.id,
invocationSource: "automation",
triggerDetail: "system",
status: "queued",
@@ -16140,6 +16159,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
.values({
companyId: issue.companyId,
agentId: recoveryAgent.id,
issueId: issue.id,
invocationSource: "automation",
triggerDetail: "system",
status: "queued",
@@ -17257,6 +17277,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
.values({
companyId: agent.companyId,
agentId,
issueId: readNonEmptyString(enrichedContextSnapshot.issueId),
invocationSource: source,
triggerDetail,
status: "queued",
@@ -17431,6 +17452,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
.values({
companyId: agent.companyId,
agentId,
issueId: readNonEmptyString(enrichedContextSnapshot.issueId),
invocationSource: source,
triggerDetail,
status: "queued",
+1
View File
@@ -3052,6 +3052,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
.values({
companyId: input.issue.companyId,
agentId: input.agentId,
issueId: input.issue.id,
invocationSource: "automation",
triggerDetail: "system",
status: "scheduled_retry",
@@ -38,6 +38,7 @@ function createRun(overrides: Partial<HeartbeatRun> = {}): HeartbeatRun {
id: "run-1",
companyId: "company-1",
agentId: "agent-1",
issueId: null,
responsibleUserId: null,
invocationSource: "on_demand",
triggerDetail: "manual",
+1
View File
@@ -133,6 +133,7 @@ function makeRun(id: string, status: HeartbeatRun["status"], createdAt: string,
id,
companyId: "company-1",
agentId,
issueId: null,
responsibleUserId: null,
invocationSource: "assignment",
triggerDetail: null,
+1
View File
@@ -815,6 +815,7 @@ describe("FailedRunInboxRow", () => {
id: "run-1",
companyId: "company-1",
agentId: "agent-1",
issueId: null,
responsibleUserId: null,
invocationSource: "assignment",
triggerDetail: null,
@@ -96,6 +96,7 @@ function makeHeartbeatRun(overrides: Partial<HeartbeatRun>): HeartbeatRun {
id: "run-fixture",
companyId,
agentId: "agent-codex",
issueId: null,
responsibleUserId: null,
invocationSource: "on_demand",
triggerDetail: "manual",