Skip to content

Commit 0fa1932

Browse files
committed
Sweep turn screenshots in every deployment, not only where a culler runs
1 parent 8f69e61 commit 0fa1932

6 files changed

Lines changed: 213 additions & 49 deletions

File tree

CHANGELOG.md

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,30 @@ than accepting one it cannot attribute. `scripts/start.sh` runs the worker local
2828
turns it on with `routines.enabled` and takes the secret as `secrets.workerSharedSecret`. No new port
2929
is opened for any of this — the worker only ever calls out to the server it already trusts.
3030

31+
### Turn screenshots are swept in every deployment, not one
32+
33+
A page a Bot opens is photographed and kept in `computer_page_frame`, so a conversation read back
34+
later shows what it was looking at. The reaper for those rows had one caller: the idle-computer
35+
culler, which refuses to run unless each Bot has its own computer and is scheduled only by the Helm
36+
chart's CronJob, which exists only when `computers.mode` is `sandbox`. On Compose, on the all-in-one
37+
image, and on the chart's own default of `shared`, nothing ever called it. One browsing Bot over
38+
ninety days is several hundred megabytes of rows that nothing was ever going to remove.
39+
40+
The sweep now runs on the server, on the same hourly timer that removes old audit rows, and does not
41+
wait for a retention policy to be configured: a month of screenshots is what the store already meant
42+
to keep. It also removes them in batches, because one statement over that much data held its locks
43+
for seventeen seconds.
44+
45+
Deployments using `computers.mode: sandbox` are unaffected in what they keep. The culler no longer
46+
purges frames, because the server does it there too and one owner is better than two.
47+
48+
**On upgrade, the first sweep removes the backlog.** A deployment that has been keeping every
49+
screenshot since it was installed will lose the ones older than a month, about a minute after the
50+
server starts. That is the window the store has always documented and the one sandbox deployments
51+
have been enforcing, but it has never been applied anywhere else, so it is worth knowing before the
52+
upgrade rather than after. It is drained in batches, forty thousand rows an hour, rather than in one
53+
statement.
54+
3155
### A channel a Bot has spoken in unseen shows a dot
3256

3357
The sidebar marks a channel when a Bot has said something since you last had it open: a dot beside

server/scripts/cull-idle-computers.ts

Lines changed: 0 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@
1111
* losing anything, and a failing CronJob that pages somebody at 3am should mean something worse.
1212
*/
1313
import { randomUUID } from "node:crypto";
14-
import { createPageFrameStore } from "../src/computer/page-frames";
1514
import { createComputerProvider } from "../src/computer/provider";
1615
import { loadConfig } from "../src/config";
1716
import { createDatabase } from "../src/db/client";
@@ -36,21 +35,11 @@ if (config.computer.provider !== "sandbox") {
3635

3736
const database = createDatabase(config.databaseUrl);
3837
const queue = createWorkQueue(database);
39-
const pageFrames = createPageFrameStore(database);
4038
const provider = createComputerProvider(config.computer);
4139

4240
// A name for the lease, so a stuck claim can be traced back to the pod that took it.
4341
const owner = `culler/${process.env.HOSTNAME ?? randomUUID().slice(0, 8)}`;
4442

45-
/**
46-
* How long a turn's screenshot is kept.
47-
*
48-
* A month, because reading back a conversation is the thing these exist for and people do that long
49-
* after the run. Past that the transcript names the page it opened instead, which is the same
50-
* sentence with less in it rather than a broken one.
51-
*/
52-
const FRAME_RETENTION_MS = 30 * 24 * 60 * 60 * 1000;
53-
5443
try {
5544
const options = {
5645
database,
@@ -86,27 +75,13 @@ try {
8675
*/
8776
finishedOlderThanMs: config.computer.idleAfterMs,
8877
});
89-
/*
90-
* And the screenshots, which had a reaper and nothing calling it.
91-
*
92-
* A page is a row and a Bot that browses makes them for as long as it runs, so this table only
93-
* ever grew: written on every navigation, taken out by a profile wipe and by nothing else. Kept
94-
* long enough that reading back a conversation from last month still shows what it opened, and not
95-
* for ever, because these are the largest thing this deployment stores and the least useful once
96-
* nobody is reading that conversation any more.
97-
*
98-
* Here rather than in the API server because this is already the sweep that runs on a schedule
99-
* with a claim under it, and a second timer would be a second thing to get wrong.
100-
*/
101-
const framesPurged = await pageFrames.purge(FRAME_RETENTION_MS);
10278
console.info(
10379
JSON.stringify({
10480
type: "computer-cull",
10581
offered,
10682
suspended: report.suspended,
10783
skipped: report.skipped,
10884
purged,
109-
framesPurged,
11085
}),
11186
);
11287
} finally {

server/src/audit-retention.ts

Lines changed: 27 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,10 @@
2121
* every action writes to.
2222
*/
2323
import postgres from "postgres";
24+
import {
25+
FRAME_RETENTION_MS,
26+
type PageFrameStore,
27+
} from "./computer/page-frames";
2428

2529
/**
2630
* The advisory lock this takes, as an arbitrary but fixed number.
@@ -117,18 +121,39 @@ export type RetentionSweeper = { stop: () => void };
117121
* Not immediately at boot: a deployment rolling several servers would have all of them contend for
118122
* the lock in the same second, and the one that wins would compete with start-up for the database.
119123
*/
120-
export function startAuditRetention(
124+
export function startRetentionSweeps(
121125
databaseUrl: string,
122126
retentionDays: number | undefined,
127+
// Swept whatever the audit policy is: their only other caller runs in `computers.mode: sandbox` alone.
128+
pageFrames?: PageFrameStore,
123129
options: { intervalMs?: number; firstRunMs?: number } = {},
124130
): RetentionSweeper {
125-
if (!retentionDays || retentionDays < 1) return { stop: () => undefined };
131+
const sweepsAudit = Boolean(retentionDays && retentionDays >= 1);
132+
if (!sweepsAudit && !pageFrames) return { stop: () => undefined };
126133

127134
const intervalMs = options.intervalMs ?? 60 * 60_000;
128135
const firstRunMs = options.firstRunMs ?? 60_000;
129136
const timers: ReturnType<typeof setInterval>[] = [];
130137

131138
const run = () => {
139+
if (pageFrames) {
140+
void pageFrames
141+
.purge(FRAME_RETENTION_MS)
142+
.then((removed) => {
143+
if (removed === 0) return;
144+
console.info(JSON.stringify({ type: "page-frames-swept", removed }));
145+
})
146+
.catch((error) => {
147+
console.error(
148+
JSON.stringify({
149+
type: "page-frames-sweep-failed",
150+
note: "Old turn screenshots were not removed. Nothing is broken; the table is larger than it should be.",
151+
error: String(error),
152+
}),
153+
);
154+
});
155+
}
156+
if (!sweepsAudit || !retentionDays) return;
132157
void sweepAuditTrail(databaseUrl, retentionDays)
133158
.then(({ deleted }) => {
134159
if (deleted === null || deleted === 0) return;

server/src/computer/page-frames.ts

Lines changed: 29 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
* Written where the navigation happens, which is the one moment the screen is certainly showing the
55
* page that was asked for, and read back when somebody reopens the conversation that asked for it.
66
*/
7-
import { and, eq, lt, sql } from "drizzle-orm";
7+
import { and, eq, sql } from "drizzle-orm";
88
import type { Database } from "../db/client";
99
import { computerPageFrame } from "../db/schema";
1010

@@ -17,6 +17,19 @@ import { computerPageFrame } from "../db/schema";
1717
*/
1818
const MAX_FRAME_BYTES = 4 * 1024 * 1024;
1919

20+
/**
21+
* How long a turn's screenshot is kept.
22+
*
23+
* A month, because reading back a conversation is the thing these exist for and people do that long
24+
* after the run. Past that the transcript names the page it opened instead, which is the same
25+
* sentence with less in it rather than a broken one.
26+
*/
27+
export const FRAME_RETENTION_MS = 30 * 24 * 60 * 60 * 1000;
28+
29+
// Small batches: a frame is hundreds of kilobytes, so the audit sweep's five thousand would be a gigabyte a statement.
30+
const PURGE_BATCH = 200;
31+
const MAX_PURGE_BATCHES = 200;
32+
2033
/**
2134
* How big a base64 string actually is, in bytes.
2235
*
@@ -124,16 +137,21 @@ export function createPageFrameStore(database: Database): PageFrameStore {
124137
},
125138

126139
async purge(olderThanMs) {
127-
const gone = await database
128-
.delete(computerPageFrame)
129-
.where(
130-
lt(
131-
computerPageFrame.capturedAt,
132-
sql`now() - make_interval(secs => ${olderThanMs / 1000})`,
133-
),
134-
)
135-
.returning({ url: computerPageFrame.url });
136-
return gone.length;
140+
// Batched like the audit sweep: one statement over ninety days of one Bot held its locks for 17s.
141+
let removed = 0;
142+
for (let batch = 0; batch < MAX_PURGE_BATCHES; batch += 1) {
143+
const result = (await database.execute(sql`
144+
delete from ${computerPageFrame} where ctid in (
145+
select ctid from ${computerPageFrame}
146+
where ${computerPageFrame.capturedAt} < now() - make_interval(secs => ${olderThanMs / 1000})
147+
limit ${PURGE_BATCH}
148+
)
149+
`)) as unknown as { count?: number };
150+
const count = result?.count ?? 0;
151+
removed += count;
152+
if (count < PURGE_BATCH) break;
153+
}
154+
return removed;
137155
},
138156
};
139157
}

server/src/index.ts

Lines changed: 9 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ import type { AgentActor } from "./agents/profile-types";
1111
import { createRuntimeAgentLoader } from "./agents/runtime-agents";
1212
import { createApp } from "./app";
1313
import { createAuditReader, createAuditStore, recordAuditEvent } from "./audit";
14-
import { startAuditRetention } from "./audit-retention";
14+
import { startRetentionSweeps } from "./audit-retention";
1515
import { createAuth } from "./auth";
1616
import { DEV_ACTOR, initializeDevActorUser } from "./auth/dev-actor";
1717
import { createRoleRepository } from "./auth/guards";
@@ -248,15 +248,13 @@ const policyListener = await startPolicyListener(
248248
* unavailable, and the row is a note for a reader rather than something the server depends on.
249249
*/
250250
const bootAuditStore = createAuditStore(database);
251-
/*
252-
* Old audit rows removed on a schedule, when a deployment has asked for that.
253-
*
254-
* One server sweeps rather than all of them, decided by an advisory lock. Off unless
255-
* `AUDIT_RETENTION_DAYS` is set. See audit-retention.ts.
256-
*/
257-
const auditRetention = startAuditRetention(
251+
// One store: the gateway writes through it, a route reads it, and the sweep below takes the old ones out.
252+
const pageFrameStore = createPageFrameStore(database);
253+
// Housekeeping on a schedule: audit rows when asked for, screenshots always, one timer. See audit-retention.ts.
254+
const retentionSweeps = startRetentionSweeps(
258255
config.databaseUrl,
259256
config.auditRetentionDays,
257+
pageFrameStore,
260258
);
261259
const computerGateway = computerProvider
262260
? createComputerGateway({
@@ -269,7 +267,7 @@ const computerGateway = computerProvider
269267
snapshots: createSnapshotStore(database),
270268
// So wiping a profile takes the pictures of its signed-in pages with it, which is what the
271269
// sentence on that button already promised.
272-
pageFrames: createPageFrameStore(database),
270+
pageFrames: pageFrameStore,
273271
allowPrivateHosts: config.computer?.allowPrivateHosts,
274272
token: config.computer?.token,
275273
})
@@ -711,7 +709,7 @@ const app = createApp(
711709
// Chooses the coworker for an untagged message, on the deployment's own model and key.
712710
intentRouter,
713711
// What a browsing turn's screen looked like when it finished, so the transcript can show it later.
714-
createPageFrameStore(database),
712+
pageFrameStore,
715713
// What a due routine actually does: a turn, run as its owner, into the thread they will open.
716714
routineRunner,
717715
// A person's own standing instructions: the list, and a switch to stop one.
@@ -880,7 +878,7 @@ for (const signal of ["SIGINT", "SIGTERM"] as const) {
880878
void Promise.allSettled([
881879
channelActivityListener.stop(),
882880
policyListener.stop(),
883-
Promise.resolve(auditRetention.stop()),
881+
Promise.resolve(retentionSweeps.stop()),
884882
]).finally(() => process.exit(0));
885883
});
886884
}
Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
import { afterEach, describe, expect, test } from "bun:test";
2+
import { eq, sql } from "drizzle-orm";
3+
import { startRetentionSweeps } from "../src/audit-retention";
4+
import {
5+
createPageFrameStore,
6+
FRAME_RETENTION_MS,
7+
} from "../src/computer/page-frames";
8+
import { createDatabase } from "../src/db/client";
9+
import { computerPageFrame } from "../src/db/schema";
10+
import { TEST_POOL } from "./support/database";
11+
12+
/**
13+
* The screenshots have to be able to stop growing, in every deployment rather than in one.
14+
*
15+
* Their reaper had a single caller, `scripts/cull-idle-computers.ts`, which refuses to run unless the
16+
* provider is `sandbox` and is scheduled only by the chart's culler CronJob, which renders only in
17+
* that mode. Compose, the all-in-one image and the chart's own default of `computers.mode: shared`
18+
* therefore wrote a row per navigation and removed none, ever.
19+
*
20+
* Against a real database because the interval arithmetic and the batching are both SQL.
21+
*/
22+
23+
const databaseUrl =
24+
process.env.DATABASE_URL ??
25+
"postgres://openbot:openbot@localhost:5432/openbot";
26+
const database = createDatabase(databaseUrl, TEST_POOL);
27+
const store = createPageFrameStore(database);
28+
29+
const COMPUTER = `frame-retention-${crypto.randomUUID().slice(0, 8)}`;
30+
31+
async function frame(daysAgo: number, index: number): Promise<void> {
32+
await store.save({
33+
computerId: COMPUTER,
34+
toolCallId: `turn-${index}`,
35+
url: `https://example.com/${index}`,
36+
title: `page ${index}`,
37+
frame: "iVBORw0KGgo=",
38+
});
39+
await database
40+
.update(computerPageFrame)
41+
.set({ capturedAt: sql`now() - make_interval(days => ${daysAgo})` })
42+
.where(eq(computerPageFrame.toolCallId, `turn-${index}`));
43+
}
44+
45+
const kept = () =>
46+
database
47+
.select({ toolCallId: computerPageFrame.toolCallId })
48+
.from(computerPageFrame)
49+
.where(eq(computerPageFrame.computerId, COMPUTER));
50+
51+
afterEach(async () => {
52+
await database
53+
.delete(computerPageFrame)
54+
.where(eq(computerPageFrame.computerId, COMPUTER));
55+
});
56+
57+
describe("page frame retention", () => {
58+
test("a deployment that has not configured audit retention still sweeps frames", async () => {
59+
await frame(90, 1);
60+
await frame(31, 2);
61+
await frame(1, 3);
62+
63+
/*
64+
* `undefined` is the whole point of this case. It is what a deployment that never set
65+
* `AUDIT_RETENTION_DAYS` has, and the sweeper used to return before starting a timer at all,
66+
* which is how the frames went unswept everywhere the culler does not run.
67+
*/
68+
const sweeps = startRetentionSweeps(databaseUrl, undefined, store, {
69+
firstRunMs: 10,
70+
intervalMs: 3_600_000,
71+
});
72+
try {
73+
for (let attempt = 0; attempt < 100; attempt += 1) {
74+
if ((await kept()).length === 1) break;
75+
await new Promise((resolve) => setTimeout(resolve, 50));
76+
}
77+
} finally {
78+
sweeps.stop();
79+
}
80+
81+
expect((await kept()).map((row) => row.toolCallId)).toEqual(["turn-3"]);
82+
});
83+
84+
test("purge removes everything past the window and nothing inside it", async () => {
85+
await frame(40, 1);
86+
await frame(29, 2);
87+
88+
await store.purge(FRAME_RETENTION_MS);
89+
expect((await kept()).map((row) => row.toolCallId)).toEqual(["turn-2"]);
90+
});
91+
92+
test("purge past its batch size removes every eligible row", async () => {
93+
for (let index = 1; index <= 205; index += 1) await frame(45, index);
94+
95+
await store.purge(FRAME_RETENTION_MS);
96+
expect(await kept()).toHaveLength(0);
97+
});
98+
99+
test("purge leaves a frame inside the window alone", async () => {
100+
await frame(1, 1);
101+
102+
await store.purge(FRAME_RETENTION_MS);
103+
expect(await kept()).toHaveLength(1);
104+
});
105+
106+
/*
107+
* Asserted on what survives for this computer rather than on what `purge` returns: the sweep is
108+
* deployment-wide by design, so its count moves with whatever else is in the table.
109+
*
110+
* Two replicas sweeping at once, which is the ordinary case: this half takes no advisory lock,
111+
* because a delete keyed on age run twice removes the same rows once. The risk it has to be shown
112+
* not to have is double-counting or deadlocking on the same `ctid`s.
113+
*/
114+
test("two sweeps at once remove each row once and neither stalls", async () => {
115+
for (let index = 1; index <= 400; index += 1) await frame(45, index);
116+
117+
await Promise.all([
118+
store.purge(FRAME_RETENTION_MS),
119+
store.purge(FRAME_RETENTION_MS),
120+
]);
121+
122+
expect(await kept()).toHaveLength(0);
123+
});
124+
});

0 commit comments

Comments
 (0)