mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 10:13:11 +02:00
perf: narrow runtime progress database reads (#17355)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Active runs publish workspace and provider progress. > - Each update must check the current run status before it publishes. > - This check fetched the full run, including prompts and runner configuration. > - This pull request reads only the four fields that progress publication uses. > - The benefit is less database transfer while each update still checks durable state. ## Linked Issues or Issue Description Related: #17348 narrows authorization reads. This PR applies a separate projection to runtime progress. GitHub searches for runtime progress and run-status read PRs found no duplicate. **What existing behavior does this improve?** The database read before a runtime progress update. **Current behavior** Each update calls `getRun`. This reads large run context and runner configuration even though progress publication only uses the run ID, company ID, agent ID, and status. **Proposed behavior** Use a fresh four-column read for each update. Keep the existing active-status check, event fields, error propagation, and late-progress suppression. **Reason and benefit** Transfer progress can emit many updates for one run. A small status check avoids repeatedly transferring payloads that it does not use. **Breaking changes** None. No API response, schema, scheduling, or cancellation policy changes. ## What Changed - Add `getRunStatus` to the run-state owner. The full reader remains available to consumers that need execution data. - Wire the narrow reader into the lifecycle progress check. Remove the unused context field from that check's input contract. - Cover the exact selected fields, missing runs, database errors, large stored payloads, terminal transitions, and the public heartbeat service wiring. Document the read boundary. ## Verification - Run-state and lifecycle suites: 44 tests passed before the additional public-service regression. - Lifecycle, public-service wiring, and native cancellation suites: 36 tests passed. - `pnpm -r typecheck`: passed. - `pnpm build`: passed. - The [full PR workflow](https://github.com/paperclipai/paperclip/actions/runs/40013092548) passed on `95bf496f689049812646b5ccddcdf2cca21efad0`. All 54 reported checks passed; two conditional checks were skipped. This includes test shards, browser checks, Runner verification, production bundle boot, and clean npm installation. - `GIT_CONFIG_COUNT=1 GIT_CONFIG_KEY_0=init.defaultBranch GIT_CONFIG_VALUE_0=master pnpm test:run`: an additional local all-in-one run is still in progress; no full-local-pass claim. The process-scoped Git default matches existing master-only fixtures and does not change user configuration. - The initial description check failed because the dedup-search checkbox was missing. Searches are complete and the template checkbox is now included. The rerun passed on the same source commit. - Greptile Apex: 5/5 on `95bf496f689049812646b5ccddcdf2cca21efad0`, no findings or review threads. - No hosted deployment or bill reduction is claimed. CI and Apex are complete; this PR is merge-ready. ## Risks - Low scope: the query selects fewer columns and preserves the existing status read on every progress callback. It adds no worker, cache, timer, or idle query. - A run can still finish after the status read and before publication, as it could before this change. This PR does not change that concurrency boundary. - No migration or operator configuration is required. ## Model Used OpenAI Codex (GPT-6 family). The runtime does not expose the exact model ID or context window. ## 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 #` 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 (focused suites above; full suite is still running) - [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 - [x] 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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
837add0caa
commit
ef9e17304c
6 files changed
+108
-9
No files matched your search
@@ -2858,6 +2858,12 @@ package avatar PNGs using the API renderer; static hosting needs no API proxy.
|
||||
|
||||
### Investigating polling load
|
||||
|
||||
Runtime progress updates recheck the persisted run status before publication.
|
||||
This check selects only the run ID, company ID, agent ID, and status through
|
||||
`createHeartbeatRunState().getRunStatus`. It does not fetch prompts, runner
|
||||
configuration, or provider results. Keep this read fresh so terminal or deleted
|
||||
runs cannot publish late progress from an old executor snapshot.
|
||||
|
||||
Run the deterministic native event replay with the normal server test runner:
|
||||
|
||||
```sh
|
||||
|
||||
@@ -810,6 +810,7 @@ export function heartbeatService(
|
||||
getAgent,
|
||||
resolveSessionBeforeForWakeup,
|
||||
getRun,
|
||||
getRunStatus,
|
||||
ensureRuntimeState,
|
||||
getTaskSession,
|
||||
getLatestAgentConfigRevision,
|
||||
@@ -1306,7 +1307,7 @@ export function heartbeatService(
|
||||
issuesSvc,
|
||||
recovery,
|
||||
companySkills,
|
||||
getRun,
|
||||
getRunStatus,
|
||||
emitTerminalAgentTaskRun,
|
||||
getIssueExecutionContext,
|
||||
budgets,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { randomBytes } from "node:crypto";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
@@ -7,6 +8,7 @@ import type { CurrentUserRedactionOptions } from "../../log-redaction.js";
|
||||
import { publishLiveEvent } from "../live-events.js";
|
||||
import { getHeartbeatRunRuntimeStatus } from "../heartbeat-run-runtime-status.js";
|
||||
import { createHeartbeatRunState } from "./run-state.js";
|
||||
import { heartbeatService } from "../heartbeat.js";
|
||||
import { createHeartbeatRunPreparation } from "./run-preparation.js";
|
||||
import { createHeartbeatLifecycle, persistHeartbeatRunProcessMetadata, type HeartbeatLifecycleDependencies } from "./run-lifecycle.js";
|
||||
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../delivery-work-notifications.js";
|
||||
@@ -19,7 +21,7 @@ function callbacks(db: Db) {
|
||||
return {
|
||||
...state,
|
||||
...createHeartbeatRunPreparation(db),
|
||||
getRun: vi.fn(state.getRun),
|
||||
getRunStatus: vi.fn(state.getRunStatus),
|
||||
issuesSvc: {
|
||||
addComment: vi.fn<HeartbeatLifecycleDependencies["issuesSvc"]["addComment"]>(),
|
||||
listReviewAttention: vi.fn<HeartbeatLifecycleDependencies["issuesSvc"]["listReviewAttention"]>(),
|
||||
@@ -84,12 +86,25 @@ describe("heartbeat lifecycle module boundary", () => {
|
||||
const database = guardedDatabase();
|
||||
const deps = callbacks(database.db);
|
||||
const run = { id: "late-progress", companyId: "company", agentId: "agent", status: "running", contextSnapshot: {} } as Run;
|
||||
deps.getRun.mockResolvedValue({ ...run, status: "succeeded" });
|
||||
deps.getRunStatus.mockResolvedValue({ ...run, status: "succeeded" });
|
||||
expect(await createHeartbeatLifecycle(database.db, deps).recordCurrentHeartbeatRunRuntimeProgress(run, { phase: "finalize", message: "late update" }, null)).toBeNull();
|
||||
expect(deps.getRun).toHaveBeenCalledWith(run.id);
|
||||
expect(deps.getRunStatus).toHaveBeenCalledWith(run.id);
|
||||
expect(getHeartbeatRunRuntimeStatus(run.id)).toBeNull();
|
||||
expect(database.access).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not publish progress when the current status read fails", async () => {
|
||||
const database = guardedDatabase();
|
||||
const deps = callbacks(database.db);
|
||||
deps.getRunStatus.mockRejectedValue(new Error("status read failed"));
|
||||
const run = { id: "unavailable-progress", companyId: "company", agentId: "agent", status: "running" };
|
||||
vi.mocked(publishLiveEvent).mockClear();
|
||||
await expect(createHeartbeatLifecycle(database.db, deps)
|
||||
.recordCurrentHeartbeatRunRuntimeProgress(run, { phase: "restore", message: "restoring" }, null))
|
||||
.rejects.toThrow("status read failed");
|
||||
expect(publishLiveEvent).not.toHaveBeenCalled();
|
||||
expect(database.access).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
@@ -118,6 +133,53 @@ describe.skipIf(!support.supported)("heartbeat lifecycle database wiring", () =>
|
||||
return { company, agent, wake, run };
|
||||
}
|
||||
|
||||
it.each(["succeeded", "failed", "cancelled", "timed_out", "interrupted", "deleted"])(
|
||||
"rechecks progress after a run becomes %s without reading its large payloads",
|
||||
async (status) => {
|
||||
const { run } = await fixture({
|
||||
contextSnapshot: { prompt: randomBytes(128 * 1024).toString("hex") },
|
||||
runnerProfileJson: { configuration: randomBytes(128 * 1024).toString("hex") },
|
||||
resultJson: { providerData: randomBytes(128 * 1024).toString("hex") },
|
||||
});
|
||||
const deps = callbacks(db);
|
||||
const lifecycle = createHeartbeatLifecycle(db, deps);
|
||||
const update = { phase: "restore" as const, message: "Restoring workspace", currentToolName: "sync" };
|
||||
const progress = await lifecycle.recordCurrentHeartbeatRunRuntimeProgress(run, update, null);
|
||||
expect(progress).toMatchObject({
|
||||
companyId: run.companyId, agentId: run.agentId, runId: run.id,
|
||||
phase: update.phase, message: update.message, currentToolName: "sync",
|
||||
});
|
||||
expect(await deps.getRunStatus.mock.results[0].value).toEqual({
|
||||
id: run.id, companyId: run.companyId, agentId: run.agentId, status: "running",
|
||||
});
|
||||
expect(publishLiveEvent).toHaveBeenCalledTimes(1);
|
||||
expect(publishLiveEvent).toHaveBeenCalledWith(expect.objectContaining({
|
||||
companyId: run.companyId,
|
||||
type: "heartbeat.run.progress",
|
||||
payload: expect.objectContaining({ runId: run.id, agentId: run.agentId, phase: update.phase, message: update.message }),
|
||||
}));
|
||||
|
||||
if (status === "deleted") await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
else await db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, run.id));
|
||||
|
||||
expect(await lifecycle.recordCurrentHeartbeatRunRuntimeProgress(run, update, null)).toBeNull();
|
||||
expect(deps.getRunStatus).toHaveBeenCalledTimes(2);
|
||||
expect(getHeartbeatRunRuntimeStatus(run.id)).toBeNull();
|
||||
expect(publishLiveEvent).toHaveBeenCalledTimes(1);
|
||||
},
|
||||
);
|
||||
|
||||
it("wires fresh progress status reads through the public heartbeat service", async () => {
|
||||
const { run } = await fixture();
|
||||
const service = heartbeatService(db);
|
||||
const update = { phase: "config_sync" as const, message: "Syncing workspace" };
|
||||
expect(await service.recordRuntimeProgress(run, update, null)).toMatchObject({ runId: run.id, message: update.message });
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await service.recordRuntimeProgress(run, update, null)).toBeNull();
|
||||
expect(getHeartbeatRunRuntimeStatus(run.id)).toBeNull();
|
||||
expect(publishLiveEvent).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("preserves sandbox process birth instead of inspecting a colliding controller PID", async () => {
|
||||
const { run } = await fixture();
|
||||
const remoteBirth = "2026-10-09T00:00:00.000Z";
|
||||
|
||||
@@ -193,7 +193,7 @@ type HeartbeatRun = typeof heartbeatRuns.$inferSelect;
|
||||
|
||||
/** Lifecycle owns run writes; the service supplies reporting and admission effects. */
|
||||
export interface HeartbeatLifecycleDependencies extends Pick<ReturnType<typeof createHeartbeatRunState>,
|
||||
"getRun" | "resolveSessionBeforeForWakeup" | "ensureRuntimeState">,
|
||||
"getRunStatus" | "resolveSessionBeforeForWakeup" | "ensureRuntimeState">,
|
||||
Pick<ReturnType<typeof createHeartbeatRunPreparation>,
|
||||
"getIssueExecutionContext" | "resolveResponsibleUserIdForRunContext">,
|
||||
Pick<HeartbeatRetryDependencies, "getAgentInvokability"> {
|
||||
@@ -388,7 +388,7 @@ export function publishHeartbeatRunRuntimeProgress(status: {
|
||||
function recordHeartbeatRunRuntimeProgress(
|
||||
run: Pick<
|
||||
typeof heartbeatRuns.$inferSelect,
|
||||
"id" | "companyId" | "agentId" | "status" | "contextSnapshot"
|
||||
"id" | "companyId" | "agentId" | "status"
|
||||
>,
|
||||
update: RuntimeStatusUpdate,
|
||||
issueId: string | null,
|
||||
@@ -635,7 +635,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
|
||||
issuesSvc,
|
||||
recovery,
|
||||
companySkills,
|
||||
getRun,
|
||||
getRunStatus,
|
||||
emitTerminalAgentTaskRun,
|
||||
getIssueExecutionContext,
|
||||
budgets,
|
||||
@@ -992,7 +992,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
|
||||
async function recordCurrentHeartbeatRunRuntimeProgress(
|
||||
run: Pick<
|
||||
typeof heartbeatRuns.$inferSelect,
|
||||
"id" | "companyId" | "agentId" | "status" | "contextSnapshot"
|
||||
"id" | "companyId" | "agentId" | "status"
|
||||
>,
|
||||
update: RuntimeStatusUpdate,
|
||||
issueId: string | null,
|
||||
@@ -1002,7 +1002,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
|
||||
return null;
|
||||
}
|
||||
|
||||
const currentRun = await getRun(run.id);
|
||||
const currentRun = await getRunStatus(run.id);
|
||||
if (!currentRun || !isHeartbeatRunRuntimeStatusActive(currentRun.status)) {
|
||||
clearHeartbeatRunRuntimeStatus(run.id);
|
||||
return null;
|
||||
|
||||
@@ -92,6 +92,22 @@ describe("heartbeat run state module", () => {
|
||||
expect(raw.columns[0].resultJson).toBe(heartbeatRuns.resultJson);
|
||||
});
|
||||
|
||||
it("loads current run status without large fields or an encoding probe", async () => {
|
||||
const binding = { id: "run", companyId: "company", agentId: "agent", status: "running" };
|
||||
const reader = readingDatabase("SQL_ASCII", [binding], []);
|
||||
const state = extracted.createHeartbeatRunState(reader.db);
|
||||
expect(await state.getRunStatus("run")).toEqual(binding);
|
||||
expect(await state.getRunStatus("missing")).toBeNull();
|
||||
expect(reader.execute).not.toHaveBeenCalled();
|
||||
expect(reader.columns).toEqual(Array(2).fill({
|
||||
id: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status,
|
||||
}));
|
||||
expect(reader.scopes).toEqual([["run"], ["missing"]]);
|
||||
});
|
||||
|
||||
it("loads run read bindings without large fields or an encoding probe", async () => {
|
||||
const binding = { id: "run", companyId: "company", scopeKind: "issue", issueId: "issue" };
|
||||
const reader = readingDatabase("SQL_ASCII", [binding], []);
|
||||
|
||||
@@ -1119,6 +1119,19 @@ export function createHeartbeatRunState(db: Db) {
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function getRunStatus(runId: string) {
|
||||
return db
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, runId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function getRunReadAccess(runId: string) {
|
||||
return db
|
||||
.select(heartbeatRunReadAccessColumns)
|
||||
@@ -1684,6 +1697,7 @@ export function createHeartbeatRunState(db: Db) {
|
||||
clearTaskSessions,
|
||||
resolveExplicitResumeSessionOverride,
|
||||
hasResolvablePriorSessionWorkspaceForWake,
|
||||
getRunStatus,
|
||||
getRunReadAccess,
|
||||
getRunLogAccess,
|
||||
|
||||
|
||||
Reference in new issue
Block a user