diff --git a/cloudflare/README.md b/cloudflare/README.md index 6b58b444..08adc02c 100644 --- a/cloudflare/README.md +++ b/cloudflare/README.md @@ -66,8 +66,9 @@ is the only one accepted. Changing the hostname orphans existing passkeys; people enroll new ones after signing in with a provider. GitHub is authoritative for enrolled project state. D1 holds accounts, -sessions, GitHub App state, and jobs. The one Durable Object, `RunnerHub`, holds -runners' live connections and keeps no records of its own. +sessions, GitHub App state, and jobs. The R2 bucket bound as `OUTPUT` holds raw +agent output. The one Durable Object, `RunnerHub`, holds runners' live +connections and keeps no records of its own. ### Jobs @@ -79,8 +80,8 @@ observations show ready and unclaimed. `iterate` jobs are untrusted and route only to Linux runners that declare VM builds. Without a `flake.nix`, the route offers a `bootstrap` job instead. A paused project gets no new jobs and none are claimed; resuming it dispatches the jobs that waited. Disconnecting a project -deletes its jobs and tells their runners to stop. Duplicate requests return the -active job for the same target. +deletes its jobs and their output and tells their runners to stop. Duplicate +requests return the active job for the same target. Runners are a user's own machines. Each one registers at `POST /api/runners` and gets a bearer token, which kix stores only as a hash. With that token a @@ -91,8 +92,7 @@ routed, and the open connection is the runner's presence. Over the socket the runner sends `status` messages: its opaque accounts, the attempts it is running, and its free job slots. It also sends `checkpoint`, `question`, `output`, and `result` messages. Each gets a `reply` with the same -`id` (`src/jobs/protocol.ts`). Output chunks carry a per-attempt `seq`, so a -retried chunk lands once, and maintainers read output in pages. The hub pushes `assign` and `cancel` messages +`id` (`src/jobs/protocol.ts`). The hub pushes `assign` and `cancel` messages whenever jobs or connections change. A status that lists an assigned attempt acknowledges it; until then the assignment holds one of the runner's slots and the hub resends it, and cancelling a @@ -111,6 +111,36 @@ An account limit pauses the job with an account decision; kix never switches accounts on its own. An answer requeues the job and resumes its session on the runner that holds it. +### Raw output + +Runners upload raw agent output as `output` messages of at most 1 MiB of UTF-8 +(`src/jobs/output.ts`). Each attempt numbers its chunks with a contiguous `seq` +from 0 and sends them in order: kix accepts only the attempt's next `seq`, +once the chunk before it is stored, or an identical retry of a chunk it still +holds, which succeeds without writing again. A different chunk with a used `seq` is refused, and so is a retry of an +expired one. Each chunk is one R2 object at `output///`, +readable as soon as it lands, and D1 keeps a small index row per chunk plus each +attempt's totals. + +An attempt keeps at most 64 MiB and 4096 chunks. The chunk that would pass +either cap, and every later one, gets a reply with `truncated: true` instead of +an error; the job keeps running and its summaries are unaffected. Expiring +chunks never give that capacity back. + +Maintainers read output at `GET /projects/github///jobs//output`, +in pages of four chunks ordered by `(attempt, seq)`; `?after=.` +continues from a page's `next`, and `truncated` lists the attempts that reached +the cap. Every request rechecks current maintainer authority. + +Output is kept for 30 days from each chunk's upload. A bucket lifecycle rule +deletes `output/` objects, and a daily cron (`triggers.crons` in +`wrangler.jsonc`) deletes their index rows; reads skip a row whose object is +already gone. Disconnecting a project queues its jobs' output for deletion in +the same D1 batch that deletes the jobs, then deletes each `output//` +prefix at once. The cron retries any deletion that failed. An upload racing the +disconnect finds its index row gone, queues its job's prefix for the cron again, +and deletes its own object. + `cloudflare/` is a Bun workspace of the repository root, so the Worker imports the CLI's core as `quest/core` (`../src/core`), the same parser, checks, and readiness the CLI runs. @@ -169,7 +199,8 @@ replay, and ownership. Job tests race dispatch passes, reassign a disconnected runner's job and fence its late writes, and cover routing, questions, account limits, and cancellation. They drive real WebSocket connections through the hub, including a replaced connection and the alarm that reassigns work, and check the -maintainer-only routes. No OAuth credentials or +maintainer-only routes. Output tests use local R2 for chunk ordering, retries, +the caps, expiry, and cleanup after a disconnect. No OAuth credentials or Cloudflare account are required for checks. After changing bindings, regenerate types: @@ -192,17 +223,25 @@ from CI. In the Nix shell, enter `cloudflare/` and authenticate: cd cloudflare bun run wrangler login bun run wrangler d1 create quest-auth +bun run wrangler r2 bucket create quest-output +bun run wrangler r2 bucket lifecycle add quest-output output-expiry output/ --expire-days 30 ``` +The lifecycle rule is bucket configuration, not part of `wrangler.jsonc`, so +check it on an existing bucket with +`bun run wrangler r2 bucket lifecycle list quest-output`. Keep it at the +30 days the Worker's cron uses (`RETENTION` in `src/jobs/output.ts`). + The checked-in config deploys to `https://kix.quest` as a Workers custom domain, which requires the `kix.quest` zone in the same Cloudflare account; `workers.dev` is disabled so sign-in has one origin. `AUTH_URL` is the only authentication origin and supplies generated absolute links. For an existing deployment, retain the Worker name, D1 database ID, and -`AUTH_SECRET` so account data is preserved. The `d1 create` command above is +`AUTH_SECRET` so account data is preserved. The `create` commands above are for a fresh deployment only. For another environment, change `routes`, -`AUTH_URL`, Worker name, account, database, and OAuth credentials together. +`AUTH_URL`, Worker name, account, database, bucket, and OAuth credentials +together. Register the production GitHub OAuth app with callback `https://kix.quest/api/auth/callback/github`. For Google sign-in, also register diff --git a/cloudflare/migrations/0008_output.sql b/cloudflare/migrations/0008_output.sql new file mode 100644 index 00000000..be3ee5ea --- /dev/null +++ b/cloudflare/migrations/0008_output.sql @@ -0,0 +1,37 @@ +-- Raw agent output moves to R2, one object per chunk at +-- output///. Nothing is deployed yet, so the D1 text is +-- dropped rather than copied. +DROP TABLE job_output; +-- Each attempt's output totals. It lives as long as the job, so expiring +-- chunks never give capacity back and an expired `seq` is never reused. +CREATE TABLE job_output_attempts ( + job TEXT NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, + attempt INTEGER NOT NULL, + runner TEXT NOT NULL, + -- The only `seq` the attempt accepts next; also its count of chunks. + next INTEGER NOT NULL DEFAULT 0, + bytes INTEGER NOT NULL DEFAULT 0, + -- Set once a chunk is refused for the attempt's cap; later chunks are too. + truncated INTEGER NOT NULL DEFAULT 0 CHECK(truncated IN (0,1)), + PRIMARY KEY(job,attempt) +); +-- The index of stored chunks, readable only by current project maintainers. +-- A row reserves its `seq` for one text, by hash, before the object is +-- written; `complete` marks the object written. The daily cron deletes rows +-- older than the bucket's 30-day lifecycle. +CREATE TABLE job_output ( + job TEXT NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, + attempt INTEGER NOT NULL, + seq INTEGER NOT NULL, + hash TEXT NOT NULL, + complete INTEGER NOT NULL DEFAULT 0 CHECK(complete IN (0,1)), + at INTEGER NOT NULL, + PRIMARY KEY(job,attempt,seq) +); +CREATE INDEX job_output_at ON job_output(at); +-- Jobs whose output objects a disconnect still has to delete. Written in the +-- same batch that deletes the project, and retried by the daily cron. +CREATE TABLE output_cleanup ( + job TEXT PRIMARY KEY NOT NULL, + at INTEGER NOT NULL +); diff --git a/cloudflare/src/github/service.ts b/cloudflare/src/github/service.ts index 80600276..9365ba4d 100644 --- a/cloudflare/src/github/service.ts +++ b/cloudflare/src/github/service.ts @@ -554,11 +554,15 @@ export class GitHubService { // Recheck the local revocation gate after potentially long content reads. await this.#gate(installation); if (state === "disconnected") { - // Jobs go with the project, so read the running attempts in the same transaction. + // Jobs go with the project, so read the running attempts in the same transaction, + // and queue the deletion of any output objects, which R2 cannot do in it. const [running] = await this.#env.DB.batch([ this.#env.DB.prepare( "SELECT actor,runner,id AS job,attempt FROM jobs WHERE repository=? AND state='running'", ).bind(repo.id), + this.#env.DB.prepare( + "INSERT OR IGNORE INTO output_cleanup(job,at) SELECT DISTINCT a.job,? FROM job_output_attempts a JOIN jobs j ON j.id=a.job WHERE j.repository=?", + ).bind(Date.now(), repo.id), this.#env.DB.prepare("DELETE FROM github_projects WHERE id=?").bind(repo.id), ]); return running!.results; diff --git a/cloudflare/src/index.tsx b/cloudflare/src/index.tsx index fef8066a..df49436d 100644 --- a/cloudflare/src/index.tsx +++ b/cloudflare/src/index.tsx @@ -12,11 +12,12 @@ import { Home } from "./home"; import { type Enrollment, GitHubError, GitHubService } from "./github/service"; import { ProjectsPage, ProjectSetupPage, ProjectStopControls, ProjectStatePage, projectURL } from "./projects"; import { githubWebhooks } from "./github/webhook"; +import { purge, retain } from "./jobs/output"; import { jobs, resume, stop } from "./jobs/routes"; import { JobError } from "./jobs/store"; import { Layout, type Page, type User } from "./layout"; -const app = new Hono<{ Bindings: Env }>(); +export const app = new Hono<{ Bindings: Env }>(); // Browsers send `Origin: null` on form posts under the default "no-referrer" // policy, which would fail the origin check on /sign-in and /sign-out. app.use(secureHeaders({ referrerPolicy: "same-origin" })); @@ -164,6 +165,7 @@ app.post("/projects/github/:owner/:repository", async (c) => { ); if (state === "enabled") await resume(c.env, Number(id)); await stop(c.env, stopped); + if (state === "disconnected") await purge(c.env); return c.redirect( state === "disconnected" ? "/projects" : projectURL(c.req.param("owner"), c.req.param("repository")), 303, @@ -436,5 +438,10 @@ app.onError((error, c) => { console.error({ event: "request_failed", path: c.req.path, name: error.name }); return c.json({ error: "Request failed" }, 500); }); -export default app; +export default { + fetch: app.fetch, + async scheduled(_controller, env) { + await retain(env, Date.now()); + }, +} satisfies ExportedHandler; export { RunnerHub } from "./jobs/hub"; diff --git a/cloudflare/src/jobs/hub.ts b/cloudflare/src/jobs/hub.ts index ea35d08c..834d8304 100644 --- a/cloudflare/src/jobs/hub.ts +++ b/cloudflare/src/jobs/hub.ts @@ -1,4 +1,5 @@ import { DurableObject } from "cloudflare:workers"; +import * as output from "./output"; import { MessageError, type RunnerMessage, parse } from "./protocol"; import * as store from "./store"; import { JobError } from "./store"; @@ -149,8 +150,8 @@ export class RunnerHub extends DurableObject { case "question": return store.ask(db, runner, message.job, message.attempt, message, now); case "output": - await store.output(db, runner, message, now); - return {}; + // Past the attempt's cap the reply says `truncated`, and the job carries on. + return output.append(this.env, runner, message, now); case "result": await store.finish(db, runner, message.job, message.attempt, message, now); return {}; diff --git a/cloudflare/src/jobs/output.ts b/cloudflare/src/jobs/output.ts new file mode 100644 index 00000000..923b9d58 --- /dev/null +++ b/cloudflare/src/jobs/output.ts @@ -0,0 +1,183 @@ +// Raw agent output: one R2 object per chunk, indexed in D1. +// +// An attempt's chunks arrive in order, `seq` 0, 1, 2, ... Each is reserved +// first: an index row records its text's hash while the job exists, so a +// conflicting chunk is refused before anything reaches R2. The object is +// written next and the row marked complete, only while the attempt is +// current. A chunk is reserved only once the one before it is complete, so an +// attempt's one pending chunk is its last, and readers that see complete rows +// never page past a chunk that may still land. R2 cannot join a D1 batch, so +// a disconnect or reassignment between those steps leaves no row to complete, +// and the upload deletes its own object. +import { JobError, type Attempt, type Runner } from "./store"; + +type Storage = Pick; + +/** Raw output kept per attempt, in UTF-8 bytes. */ +export const MAX_BYTES = 64 * 1024 * 1024; +/** Chunks kept per attempt, so tiny chunks cannot multiply rows and objects. */ +export const MAX_CHUNKS = 4096; +/** How long output is kept after each chunk's upload; the bucket lifecycle rule matches it. */ +export const RETENTION = 30 * 24 * 60 * 60_000; + +const key = (job: string, attempt: number, seq: number) => `output/${job}/${attempt}/${seq}`; + +async function digest(text: string) { + const bytes = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(text)); + return Array.from(new Uint8Array(bytes), (b) => b.toString(16).padStart(2, "0")).join(""); +} + +/** + * Store a chunk of the current attempt's output, in any job state. Only the + * attempt's next `seq` is accepted, or an identical retry of one still + * indexed. Past the attempt's cap, chunks are refused as `truncated`; the + * job carries on. + */ +export async function append( + { DB: db, OUTPUT: bucket }: Storage, + runner: Runner, + { job, attempt, seq, text }: Attempt & { seq: number; text: string }, + now: number, +): Promise<{ truncated?: true }> { + const bytes = new TextEncoder().encode(text).length; + const hash = await digest(text); + const current = "EXISTS(SELECT 1 FROM jobs WHERE id=?1 AND attempt=?2 AND runner=?3)"; + const next = "job=?1 AND attempt=?2 AND runner=?3 AND next=?4"; + const params = [job, attempt, runner.id, seq, bytes, hash, now]; + const statement = (sql: string, n: number) => db.prepare(sql).bind(...params.slice(0, n)); + const [, , , , read] = await db.batch([ + statement( + `INSERT INTO job_output_attempts(job,attempt,runner) SELECT ?1,?2,?3 WHERE ${current} ON CONFLICT DO NOTHING`, + 3, + ), + // The truncation is recorded once, by the first chunk over the cap. + statement( + `UPDATE job_output_attempts SET truncated=1 WHERE ${next} AND truncated=0 AND (bytes+?5>${MAX_BYTES} OR next>=${MAX_CHUNKS}) AND ${current}`, + 5, + ), + statement( + `INSERT INTO job_output(job,attempt,seq,hash,at) SELECT ?1,?2,?4,?6,?7 FROM job_output_attempts WHERE ${next} AND truncated=0 AND ${current} + AND NOT EXISTS(SELECT 1 FROM job_output WHERE job=?1 AND attempt=?2 AND seq=?4-1 AND complete=0)`, + 7, + ), + statement( + `UPDATE job_output_attempts SET next=next+1,bytes=bytes+?5 WHERE ${next} AND EXISTS(SELECT 1 FROM job_output WHERE job=?1 AND attempt=?2 AND seq=?4)`, + 5, + ), + statement( + `SELECT a.runner,a.next,a.truncated,o.hash,o.complete,${current} AS current FROM (SELECT 1) + LEFT JOIN job_output_attempts a ON a.job=?1 AND a.attempt=?2 LEFT JOIN job_output o ON o.job=?1 AND o.attempt=?2 AND o.seq=?4`, + 4, + ), + ]); + const row = read!.results[0] as { + runner: string | null; + next: number | null; + truncated: number | null; + hash: string | null; + complete: number | null; + current: number; + }; + if (row.hash !== null && row.runner === runner.id) { + if (row.hash !== hash) throw new JobError("A different chunk has that sequence number"); + // A retry of a completed chunk writes nothing, so its lifecycle does not restart. + if (row.complete) return {}; + const name = key(job, attempt, seq); + await bucket.put(name, text); + const done = await db + .prepare( + `UPDATE job_output SET complete=1,at=?6 WHERE job=?1 AND attempt=?2 AND seq=?4 AND hash=?5 AND ${current}`, + ) + // Retention starts with the upload, as the bucket lifecycle's does. + .bind(job, attempt, runner.id, seq, hash, now) + .run(); + if (done.meta.changes === 1) return {}; + // The job was disconnected or reassigned while the object was written. A + // disconnect's cleanup may already have listed the prefix, so queue + // another in case this delete fails. + await db + .prepare( + "INSERT OR IGNORE INTO output_cleanup(job,at) SELECT ?1,?2 WHERE NOT EXISTS(SELECT 1 FROM jobs WHERE id=?1)", + ) + .bind(job, now) + .run(); + await bucket.delete(name); + throw new JobError("Attempt is no longer current"); + } + if (!row.current || row.runner !== runner.id || row.next === null) + throw new JobError("Attempt is no longer current"); + if (seq < row.next) throw new JobError("That chunk has expired"); + if (row.truncated) return { truncated: true }; + if (seq === row.next) throw new JobError(`Chunk ${seq - 1} is not stored yet`); + throw new JobError(`Expected sequence ${row.next}`); +} + +/** + * A page of stored raw output, at most a few MiB, in order; `next` continues + * it. Only stored chunks are listed: an attempt's one pending chunk is its + * last and lands only while the attempt is current, so a cursor never passes + * a chunk that may still land. + * `truncated` lists the attempts whose output reached the cap. + */ +export async function read( + { DB: db, OUTPUT: bucket }: Storage, + job: string, + after: { attempt: number; seq: number } | null, +) { + const page = 4; + const [rows, capped] = await db.batch([ + db + .prepare( + "SELECT attempt,seq,at FROM job_output WHERE job=? AND complete=1 AND (attempt,seq)>(?,?) ORDER BY attempt,seq LIMIT ?", + ) + .bind(job, after?.attempt ?? 0, after?.seq ?? -1, page), + db.prepare("SELECT attempt FROM job_output_attempts WHERE job=? AND truncated=1 ORDER BY attempt").bind(job), + ]); + const index = rows!.results as { attempt: number; seq: number; at: number }[]; + const chunks = await Promise.all( + index.map(async (chunk) => { + const object = await bucket.get(key(job, chunk.attempt, chunk.seq)); + // The lifecycle rule may delete an object before the cron deletes its row. + return object && { ...chunk, text: await object.text() }; + }), + ); + const last = index.length === page ? index.at(-1)! : null; + return { + output: chunks.filter((chunk) => chunk !== null), + next: last && { attempt: last.attempt, seq: last.seq }, + truncated: (capped!.results as { attempt: number }[]).map((row) => row.attempt), + }; +} + +/** Delete the output objects of jobs a disconnect removed. A failure stays queued for the next run. */ +export async function purge({ DB: db, OUTPUT: bucket }: Storage) { + const { results } = await db.prepare("SELECT job FROM output_cleanup ORDER BY at,job").all<{ job: string }>(); + for (const { job } of results) { + try { + const prefix = `output/${job}/`; + for ( + let listed = await bucket.list({ prefix }); + listed.objects.length; + listed = await bucket.list({ prefix }) + ) + await bucket.delete(listed.objects.map((object) => object.key)); + await db.prepare("DELETE FROM output_cleanup WHERE job=?").bind(job).run(); + } catch (error) { + console.error({ event: "output_cleanup_failed", name: error instanceof Error ? error.name : "unknown" }); + } + } +} + +/** The daily cron: forget expired chunks, whose objects the lifecycle rule deletes, and retry cleanups. */ +export async function retain(storage: Storage, now: number) { + const batch = 1000; + for (;;) { + const deleted = await storage.DB.prepare( + "DELETE FROM job_output WHERE rowid IN (SELECT rowid FROM job_output WHERE at<=? LIMIT ?)", + ) + .bind(now - RETENTION, batch) + .run(); + if (deleted.meta.changes < batch) break; + } + await purge(storage); +} diff --git a/cloudflare/src/jobs/protocol.ts b/cloudflare/src/jobs/protocol.ts index 300bb489..b087b99b 100644 --- a/cloudflare/src/jobs/protocol.ts +++ b/cloudflare/src/jobs/protocol.ts @@ -123,7 +123,7 @@ function fields(body: Record, ref: { id?: string | number }): R }; } case "output": - // Each chunk is one D1 row, which caps values at 2 MB of UTF-8. + // Each chunk is one R2 object; the bound keeps a page of output to a few MiB. if (typeof body.text !== "string" || !body.text || new TextEncoder().encode(body.text).length > 1024 * 1024) throw new JobError("Output text required, at most 1 MiB per message"); return { ...ref, type: "output", ...target(body), seq: seq(body.seq), text: body.text }; diff --git a/cloudflare/src/jobs/routes.ts b/cloudflare/src/jobs/routes.ts index f891d557..6b8a2635 100644 --- a/cloudflare/src/jobs/routes.ts +++ b/cloudflare/src/jobs/routes.ts @@ -3,6 +3,7 @@ import { createAuth } from "../auth"; import { GitHubError, GitHubService, type Stopped } from "../github/service"; import type { ProjectState, Pull } from "../github/state"; import { RUNNER_HEADER } from "./hub"; +import * as output from "./output"; import { data, id, text } from "./protocol"; import * as store from "./store"; import { JobError } from "./store"; @@ -231,10 +232,6 @@ jobs.get(`${base}/:job/output`, async (c) => { if (after !== undefined && !cursor) throw new JobError("Invalid output cursor"); c.header("Cache-Control", "no-store"); return c.json( - await store.readOutput( - c.env.DB, - job.id, - cursor ? { attempt: Number(cursor[1]), seq: Number(cursor[2]) } : null, - ), + await output.read(c.env, job.id, cursor ? { attempt: Number(cursor[1]), seq: Number(cursor[2]) } : null), ); }); diff --git a/cloudflare/src/jobs/store.ts b/cloudflare/src/jobs/store.ts index c1de6479..f76be3dc 100644 --- a/cloudflare/src/jobs/store.ts +++ b/cloudflare/src/jobs/store.ts @@ -402,29 +402,6 @@ export async function ask( return { question: existing.id, cancel: !!existing.cancel }; } -/** Append raw agent output from the current attempt, in any state. */ -export async function output( - db: D1Database, - runner: Runner, - { job, attempt, seq, text }: Attempt & { seq: number; text: string }, - now: number, -) { - const result = await db - .prepare( - "INSERT INTO job_output(job,attempt,seq,text,at) SELECT ?,?,?,?,? WHERE EXISTS(SELECT 1 FROM jobs WHERE id=? AND attempt=? AND runner=?) ON CONFLICT(job,attempt,seq) DO NOTHING", - ) - .bind(job, attempt, seq, text, now, job, attempt, runner.id) - .run(); - if (result.meta.changes === 1) return; - // A retry of a chunk that already landed. - const stored = await db - .prepare("SELECT text FROM job_output WHERE job=? AND attempt=? AND seq=?") - .bind(job, attempt, seq) - .first<{ text: string }>(); - if (!stored) throw new JobError("Attempt is no longer current"); - if (stored.text !== text) throw new JobError("A different chunk has that sequence number"); -} - /** * Finish the current attempt. An account limit pauses the job with a * decision instead of moving it to another account. A requested cancel wins @@ -664,16 +641,3 @@ export async function cancel(db: D1Database, job: string, user: string, now: num ]); if (stopped!.meta.changes + asked!.meta.changes === 0) throw new JobError("Job already finished"); } - -/** A page of raw output, at most a few MiB, in order; `next` continues it. */ -export async function readOutput(db: D1Database, job: string, after: { attempt: number; seq: number } | null) { - const page = 4; - const { results } = await db - .prepare( - "SELECT attempt,seq,text,at FROM job_output WHERE job=? AND (attempt,seq)>(?,?) ORDER BY attempt,seq LIMIT ?", - ) - .bind(job, after?.attempt ?? 0, after?.seq ?? -1, page) - .all<{ attempt: number; seq: number; text: string; at: number }>(); - const last = results.length === page ? results.at(-1)! : null; - return { output: results, next: last && { attempt: last.attempt, seq: last.seq } }; -} diff --git a/cloudflare/test/github.test.ts b/cloudflare/test/github.test.ts index f3f696dd..13ce96a4 100644 --- a/cloudflare/test/github.test.ts +++ b/cloudflare/test/github.test.ts @@ -2,7 +2,7 @@ import { env } from "cloudflare:workers"; import { createPublicKey, verify } from "node:crypto"; import { privateKey } from "./github-key"; import { beforeEach, describe, expect, it, vi } from "vitest"; -import app from "../src/index"; +import { app } from "../src/index"; import { GitHubService, type GitHubRepository } from "../src/github/service"; import { deliveryRetention } from "../src/github/webhook"; diff --git a/cloudflare/test/jobs.test.ts b/cloudflare/test/jobs.test.ts index fba85c24..75050f63 100644 --- a/cloudflare/test/jobs.test.ts +++ b/cloudflare/test/jobs.test.ts @@ -1,7 +1,8 @@ -import { runDurableObjectAlarm } from "cloudflare:test"; +import { createScheduledController, runDurableObjectAlarm } from "cloudflare:test"; import { env } from "cloudflare:workers"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import app from "../src/index"; +import worker, { app } from "../src/index"; +import * as output from "../src/jobs/output"; import * as store from "../src/jobs/store"; import { GRACE } from "../src/jobs/store"; import { parse } from "../src/jobs/protocol"; @@ -81,6 +82,7 @@ beforeEach(async () => { db.prepare("DELETE FROM runners"), db.prepare("DELETE FROM github_projects"), db.prepare("DELETE FROM github_installations"), + db.prepare("DELETE FROM output_cleanup"), ]); await user("owner"); await user("other"); @@ -299,7 +301,7 @@ describe("job coordination", () => { ]); const late = 31 + GRACE; await expect(store.checkpoint(db, old, id, 1, "Still here", null, late)).rejects.toThrow("no longer current"); - await expect(store.output(db, old, { job: id, attempt: 1, seq: 0, text: "late" }, late)).rejects.toThrow( + await expect(output.append(env, old, { job: id, attempt: 1, seq: 0, text: "late" }, late)).rejects.toThrow( "no longer current", ); await expect( @@ -529,9 +531,10 @@ describe("runner connections", () => { live.send({ id: 7, type: "result", job: id, attempt: 1, outcome: "cancelled", summary: "Stopped" }); expect(await live.next()).toEqual({ type: "reply", id: 7 }); expect(await state(id)).toMatchObject({ state: "cancelled", summary: "Stopped" }); - expect(await store.readOutput(db, id, null)).toMatchObject({ + expect(await output.read(env, id, null)).toMatchObject({ output: [{ attempt: 1, seq: 0, text: "raw log" }], next: null, + truncated: [], }); }); @@ -590,6 +593,170 @@ describe("runner connections", () => { }); }); +describe("raw output", () => { + async function started() { + const machine = await runner("owner", "machine"); + await connected(machine, [account("main")]); + const id = await job(10, "owner"); + await run({ [machine.id]: 1 }, 1); + return { machine, id }; + } + const chunk = (job: string, seq: number, text: string) => ({ job, attempt: 1, seq, text }); + const object = (job: string, seq: number) => env.OUTPUT.head(`output/${job}/1/${seq}`); + const totals = (job: string) => + db.prepare("SELECT next,bytes,truncated FROM job_output_attempts WHERE job=?").bind(job).first(); + + it("accepts each attempt's chunks in order and retries without writing again", async () => { + const { machine, id } = await started(); + await expect(output.append(env, machine, chunk(id, 1, "early"), 2)).rejects.toThrow("Expected sequence 0"); + expect(await output.append(env, machine, chunk(id, 0, "first"), 2)).toEqual({}); + const written = (await object(id, 0))!.version; + expect(await output.append(env, machine, chunk(id, 0, "first"), 3)).toEqual({}); + expect((await object(id, 0))!.version).toBe(written); + // A conflicting chunk is refused before it reaches R2. + await expect(output.append(env, machine, chunk(id, 0, "other"), 3)).rejects.toThrow("A different chunk"); + expect((await object(id, 0))!.version).toBe(written); + expect(await output.append(env, machine, chunk(id, 1, "界"), 4)).toEqual({}); + expect(await totals(id)).toEqual({ next: 2, bytes: 8, truncated: 0 }); + // Another of the owner's runners cannot write into or replay this attempt. + const other = await runner("owner", "other"); + await expect(output.append(env, other, chunk(id, 2, "x"), 5)).rejects.toThrow("no longer current"); + await expect(output.append(env, other, chunk(id, 0, "first"), 5)).rejects.toThrow("no longer current"); + }); + + it("completes a reserved chunk when a retry follows a failed object write", async () => { + const { machine, id } = await started(); + const failing = { DB: db, OUTPUT: { put: () => Promise.reject(new Error("R2 down")) } as unknown as R2Bucket }; + await expect(output.append(failing, machine, chunk(id, 0, "kept"), 2)).rejects.toThrow("R2 down"); + expect(await object(id, 0)).toBeNull(); + // The reservation holds the sequence number for that text. + await expect(output.append(env, machine, chunk(id, 0, "changed"), 3)).rejects.toThrow("A different chunk"); + expect(await output.append(env, machine, chunk(id, 0, "kept"), 3)).toEqual({}); + // Retention counts from the successful upload, as the bucket lifecycle does. + expect((await output.read(env, id, null)).output).toEqual([{ attempt: 1, seq: 0, at: 3, text: "kept" }]); + expect(await output.append(env, machine, chunk(id, 0, "kept"), 4)).toEqual({}); + expect((await output.read(env, id, null)).output).toMatchObject([{ at: 3 }]); + }); + + it("never pages past a chunk that may still land", async () => { + const { machine, id } = await started(); + const failing = { DB: db, OUTPUT: { put: () => Promise.reject(new Error("R2 down")) } as unknown as R2Bucket }; + for (const seq of [0, 1, 2]) await output.append(env, machine, chunk(id, seq, `${seq}`), 2); + await expect(output.append(failing, machine, chunk(id, 3, "3"), 2)).rejects.toThrow("R2 down"); + // The pending chunk would fill the page; the cursor stops before it. + expect(await output.read(env, id, null)).toMatchObject({ + output: [{ seq: 0 }, { seq: 1 }, { seq: 2 }], + next: null, + }); + // Later chunks wait for it, so it stays the attempt's last. + await expect(output.append(env, machine, chunk(id, 4, "4"), 3)).rejects.toThrow("Chunk 3 is not stored yet"); + expect(await output.append(env, machine, chunk(id, 3, "3"), 3)).toEqual({}); + expect(await output.read(env, id, null)).toMatchObject({ + output: [{}, {}, {}, { seq: 3 }], + next: { attempt: 1, seq: 3 }, + }); + expect(await output.append(env, machine, chunk(id, 4, "4"), 3)).toEqual({}); + await expect(output.append(failing, machine, chunk(id, 5, "5"), 4)).rejects.toThrow("R2 down"); + // Once the job is reassigned, a late retry cannot land behind a reader's cursor. + await db.prepare("UPDATE jobs SET attempt=2 WHERE id=?").bind(id).run(); + await expect(output.append(env, machine, chunk(id, 5, "5"), 5)).rejects.toThrow("no longer current"); + expect(await object(id, 5)).toBeNull(); + expect(await output.read(env, id, { attempt: 1, seq: 3 })).toMatchObject({ output: [{ seq: 4 }], next: null }); + }); + + it("refuses chunks past the attempt's cap as truncated while the job carries on", async () => { + const { machine, id } = await started(); + expect(await output.append(env, machine, chunk(id, 0, "kept"), 2)).toEqual({}); + await db + .prepare("UPDATE job_output_attempts SET bytes=? WHERE job=?") + .bind(output.MAX_BYTES - 4, id) + .run(); + expect(await output.append(env, machine, chunk(id, 1, "12345"), 3)).toEqual({ truncated: true }); + // Smaller and later chunks are refused too, so the kept output stays a prefix. + expect(await output.append(env, machine, chunk(id, 1, "1"), 4)).toEqual({ truncated: true }); + expect(await output.append(env, machine, chunk(id, 2, "1"), 4)).toEqual({ truncated: true }); + expect(await object(id, 1)).toBeNull(); + expect(await totals(id)).toEqual({ next: 1, bytes: output.MAX_BYTES - 4, truncated: 1 }); + expect(await store.checkpoint(db, machine, id, 1, "Still going", null, 5)).toEqual({ cancel: false }); + expect(await state(id)).toMatchObject({ state: "running", summary: "Still going" }); + expect(await output.read(env, id, null)).toMatchObject({ + output: [{ seq: 0, text: "kept" }], + truncated: [1], + }); + }); + + it("caps the number of chunks per attempt", async () => { + const { machine, id } = await started(); + await output.append(env, machine, chunk(id, 0, "x"), 2); + await db.prepare("UPDATE job_output_attempts SET next=? WHERE job=?").bind(output.MAX_CHUNKS, id).run(); + expect(await output.append(env, machine, chunk(id, output.MAX_CHUNKS, "x"), 3)).toEqual({ truncated: true }); + expect(await totals(id)).toMatchObject({ next: output.MAX_CHUNKS, truncated: 1 }); + }); + + it("forgets chunks after the retention period without giving back their capacity", async () => { + const { machine, id } = await started(); + const now = Date.now(); + await output.append(env, machine, chunk(id, 0, "old"), now - output.RETENTION - 1); + await output.append(env, machine, chunk(id, 1, "new"), now); + await worker.scheduled(createScheduledController(), env); + expect(await output.read(env, id, null)).toMatchObject({ output: [{ seq: 1, text: "new" }], next: null }); + // An expired sequence number is never reclaimed, and the attempt keeps counting. + await expect(output.append(env, machine, chunk(id, 0, "old"), now)).rejects.toThrow("That chunk has expired"); + expect(await totals(id)).toEqual({ next: 2, bytes: 6, truncated: 0 }); + // Reads skip a row whose object the bucket lifecycle already deleted. + await env.OUTPUT.delete(`output/${id}/1/1`); + expect(await output.read(env, id, null)).toEqual({ output: [], next: null, truncated: [] }); + }); + + it("deletes its own object when the job is disconnected during the write", async () => { + const { machine, id } = await started(); + const racing = { + DB: db, + OUTPUT: { + put: async (key: string, value: string) => { + await db.prepare("DELETE FROM github_projects WHERE id=10").run(); + return env.OUTPUT.put(key, value); + }, + delete: (key: string) => env.OUTPUT.delete(key), + } as unknown as R2Bucket, + }; + await expect(output.append(racing, machine, chunk(id, 0, "racing"), 2)).rejects.toThrow("no longer current"); + expect(await object(id, 0)).toBeNull(); + }); + + it("queues a cleanup when an upload racing a disconnect cannot delete its object", async () => { + const { machine, id } = await started(); + const racing = { + DB: db, + OUTPUT: { + put: async (key: string, value: string) => { + await db.prepare("DELETE FROM github_projects WHERE id=10").run(); + return env.OUTPUT.put(key, value); + }, + delete: () => Promise.reject(new Error("R2 down")), + } as unknown as R2Bucket, + }; + await expect(output.append(racing, machine, chunk(id, 0, "racing"), 2)).rejects.toThrow("R2 down"); + expect(await object(id, 0)).not.toBeNull(); + await worker.scheduled(createScheduledController(), env); + expect(await object(id, 0)).toBeNull(); + expect((await db.prepare("SELECT job FROM output_cleanup").all()).results).toEqual([]); + }); + + it("retries a failed output cleanup from the daily cron", async () => { + const error = vi.spyOn(console, "error").mockImplementation(() => {}); + await env.OUTPUT.put("output/gone/1/0", "orphan"); + await db.prepare("INSERT INTO output_cleanup(job,at) VALUES('gone',0)").run(); + const failing = { DB: db, OUTPUT: { list: () => Promise.reject(new Error("R2 down")) } as unknown as R2Bucket }; + await output.purge(failing); + expect(error).toHaveBeenCalledWith(expect.objectContaining({ event: "output_cleanup_failed" })); + expect((await db.prepare("SELECT job FROM output_cleanup").all()).results).toEqual([{ job: "gone" }]); + await output.retain(env, Date.now()); + expect(await env.OUTPUT.head("output/gone/1/0")).toBeNull(); + expect((await db.prepare("SELECT job FROM output_cleanup").all()).results).toEqual([]); + }); +}); + describe("runner messages", () => { it("limits output chunks by their UTF-8 size", () => { const output = (text: string) => JSON.stringify({ id: 1, type: "output", job: "j", attempt: 1, seq: 0, text }); @@ -623,6 +790,7 @@ describe("job routes", () => { let flake = true; let pullFlake = true; let role = "maintain"; + let github: (input: RequestInfo | URL) => Promise; beforeEach(async () => { flake = true; @@ -669,7 +837,7 @@ describe("job routes", () => { ) .bind(JSON.stringify(pull), JSON.stringify(issue)), ]); - vi.spyOn(globalThis, "fetch").mockImplementation(async (input) => { + github = async (input) => { const url = new URL(String(input)); const path = url.pathname; if (path.endsWith("/installation")) @@ -692,16 +860,26 @@ describe("job routes", () => { archived: false, disabled: false, }); - if (path.startsWith("/user/")) return Response.json({ id: Number(who.id), login: "maintainer" }); - if (path.endsWith("/permission")) - return Response.json({ user: { id: Number(who.id) }, permission: "write", role_name: role }); + if (path.startsWith("/user/")) { + const id = path.split("/").at(-1)!; + return Response.json({ id: Number(id), login: `person${id}` }); + } + // Only the signed-in maintainer holds the maintain role; anyone else may read. + const permission = path.match(/\/collaborators\/person(\d+)\/permission$/); + if (permission) + return Response.json( + permission[1] === who.id + ? { user: { id: Number(who.id) }, permission: "write", role_name: role } + : { user: { id: Number(permission[1]) }, permission: "read", role_name: "read" }, + ); const ref = url.searchParams.get("ref"); if (path === "/repos/upstream/project/contents/flake.nix" && (ref === commit || ref === head)) return (ref === commit ? flake : pullFlake) ? Response.json({ type: "file" }) : new Response(null, { status: 404 }); throw new Error(`Unexpected GitHub request ${path}`); - }); + }; + vi.spyOn(globalThis, "fetch").mockImplementation(github); }); function post(path: string, body: unknown, headers: Record = { Cookie: cookie, Origin: origin }) { @@ -712,7 +890,7 @@ describe("job routes", () => { ); } - it("stops a disconnected project's running jobs over their runners' connections", async () => { + it("stops a disconnected project's running jobs and deletes their output", async () => { const machine = await runner(maintainer, "machine"); const live = await connect(machine.token); live.send(status(1, "main")); @@ -722,6 +900,11 @@ describe("job routes", () => { expect(await live.next()).toMatchObject({ type: "assign", job: { id, attempt: 1 } }); live.send(status(2, "main", [{ job: id, attempt: 1 }], 0)); await live.next(); + for (const seq of [0, 1]) { + live.send({ id: 3, type: "output", job: id, attempt: 1, seq, text: `chunk ${seq}` }); + expect(await live.next()).toEqual({ type: "reply", id: 3 }); + } + expect((await env.OUTPUT.list({ prefix: `output/${id}/` })).objects).toHaveLength(2); const response = await app.request( `${origin}/projects/github/upstream/project`, { @@ -734,6 +917,11 @@ describe("job routes", () => { expect(response.status).toBe(303); expect(await live.next()).toEqual({ type: "cancel", job: id, attempt: 1 }); expect(await store.readJob(db, id)).toBeNull(); + expect((await env.OUTPUT.list({ prefix: `output/${id}/` })).objects).toEqual([]); + expect(await db.prepare("SELECT job FROM output_cleanup").all()).toMatchObject({ results: [] }); + // The runner's late chunk finds no job. + live.send({ id: 4, type: "output", job: id, attempt: 1, seq: 2, text: "late" }); + expect(await live.next()).toEqual({ type: "reply", id: 4, error: "Attempt is no longer current" }); }); it("authorizes each workflow against current observations and deduplicates requests", async () => { @@ -800,7 +988,7 @@ describe("job routes", () => { }; await store.dispatch(db, maintainer, new Map([[machine.id, 1]]), Date.now()); for (const [seq, text] of ["secret transcript", "b", "c", "d", "e"].entries()) - await store.output(db, machine, { job: id, attempt: 1, seq, text }, Date.now()); + await output.append(env, machine, { job: id, attempt: 1, seq, text }, Date.now()); const { question } = await store.ask( db, machine, @@ -809,19 +997,28 @@ describe("job routes", () => { { summary: "Which?", session: "s", prompt: { q: 1 } }, Date.now(), ); - const output = (query = "") => + const page = (query = "") => app.request(`${base}/${id}/output${query}`, { headers: { Cookie: cookie } }, testEnv); // Output comes in bounded pages. - const first = (await (await output()).json()) as { output: { text: string }[]; next: unknown }; + const first = (await (await page()).json()) as { output: { text: string }[]; next: unknown }; expect(first.output.map((chunk) => chunk.text)).toEqual(["secret transcript", "b", "c", "d"]); expect(first.next).toEqual({ attempt: 1, seq: 3 }); - expect(await (await output("?after=1.3")).json()).toEqual({ + expect(await (await page("?after=1.3")).json()).toEqual({ output: [expect.objectContaining({ seq: 4, text: "e" })], next: null, + truncated: [], }); - expect((await output("?after=x")).status).toBe(409); + expect((await page("?after=x")).status).toBe(409); + // Signed-out visitors and other signed-in users never see raw output. + expect((await app.request(`${base}/${id}/output`, {}, testEnv)).status).toBe(403); + const stranger = (await oauth(identity("github"))).session!; + vi.spyOn(globalThis, "fetch").mockImplementation(github); + const refused = await app.request(`${base}/${id}/output`, { headers: { Cookie: stranger } }, testEnv); + expect(refused.status).toBe(403); + expect(await refused.json()).toEqual({ error: "Current GitHub maintainer authority required" }); + // A maintainer whose role was revoked is refused on the next request. role = "write"; - expect((await output()).status).toBe(403); + expect((await page()).status).toBe(403); expect((await post(`/${id}/answer`, { question, answer: "A" })).status).toBe(403); role = "maintain"; expect((await post(`/${id}/answer`, { question, answer: "A" })).status).toBe(204); diff --git a/cloudflare/test/logins.test.ts b/cloudflare/test/logins.test.ts index bc36f01a..30d99d68 100644 --- a/cloudflare/test/logins.test.ts +++ b/cloudflare/test/logins.test.ts @@ -2,7 +2,7 @@ import { SELF } from "cloudflare:test"; import { env } from "cloudflare:workers"; import { afterEach, describe, expect, it, vi } from "vitest"; import { createAuth } from "../src/auth"; -import app from "../src/index"; +import { app } from "../src/index"; import { authenticator, cookie, enrolled, identity, oauth, options, origin, verify } from "./login-fixtures"; afterEach(() => vi.restoreAllMocks()); diff --git a/cloudflare/test/projects.test.ts b/cloudflare/test/projects.test.ts index 0e7aa99a..0a5f8444 100644 --- a/cloudflare/test/projects.test.ts +++ b/cloudflare/test/projects.test.ts @@ -1,7 +1,7 @@ import { env } from "cloudflare:workers"; import { createHash } from "node:crypto"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import app from "../src/index"; +import { app } from "../src/index"; import { readBoard } from "../src/board/model"; import type { ActivityPage } from "../src/github/state"; import { GitHubService } from "../src/github/service"; diff --git a/cloudflare/test/worker.test.ts b/cloudflare/test/worker.test.ts index a9623bdf..251bba69 100644 --- a/cloudflare/test/worker.test.ts +++ b/cloudflare/test/worker.test.ts @@ -2,7 +2,7 @@ import { env } from "cloudflare:workers"; import { SELF } from "cloudflare:test"; import { afterEach, describe, expect, it, vi } from "vitest"; import * as projects from "../src/board/project"; -import app from "../src/index"; +import { app } from "../src/index"; afterEach(() => vi.restoreAllMocks()); diff --git a/cloudflare/worker-configuration.d.ts b/cloudflare/worker-configuration.d.ts index ca37cd46..c89de3ba 100644 --- a/cloudflare/worker-configuration.d.ts +++ b/cloudflare/worker-configuration.d.ts @@ -1,7 +1,8 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types` (hash: 099574dd1b3e250feaa9267f9efc97b5) +// Generated by Wrangler by running `wrangler types` (hash: 405d73cc2b46c37c28e88d59fca322f9) // Runtime types generated with workerd@1.20261012.1 2026-10-01 nodejs_compat interface __BaseEnv_Env { + OUTPUT: R2Bucket; DB: D1Database; AUTH_URL: "https://kix.quest"; GITHUB_APP_ID: ""; diff --git a/cloudflare/wrangler.jsonc b/cloudflare/wrangler.jsonc index bece0d2a..7629ea7f 100644 --- a/cloudflare/wrangler.jsonc +++ b/cloudflare/wrangler.jsonc @@ -31,6 +31,11 @@ "migrations_dir": "migrations", }, ], + // Raw agent output, one object per chunk under output///. A bucket lifecycle + // rule expires output/ after 30 days; see README.md. + "r2_buckets": [{ "binding": "OUTPUT", "bucket_name": "quest-output" }], + // Daily: delete expired output index rows and retry output cleanup after a disconnect. + "triggers": { "crons": ["17 4 * * *"] }, // One RunnerHub per owner holds their runners' live connections and routes their jobs. "durable_objects": { "bindings": [{ "name": "RUNNERS", "class_name": "RunnerHub" }] }, "exports": { diff --git a/quest/m0/kix/README.md b/quest/m0/kix/README.md index 0a107077..65e5cce6 100644 --- a/quest/m0/kix/README.md +++ b/quest/m0/kix/README.md @@ -98,7 +98,6 @@ operator-paid-hosting decisions: - [Webhook reconcile](/quest/m0/kix/reconcile.md) - new issues and PRs arrive in the inbox ready to act on, without a manual refresh - [Inbox merge](/quest/m0/kix/merge.md) - PR items merge the reviewed head through GitHub's rules - [Co-maintainer inbox](/quest/m0/kix/members.md) - every current maintainer sees enrolled projects in their inbox -- [Output retention](/quest/m0/kix/output.md) - raw agent output in R2, kept 30 days and capped at 64 MiB per attempt - [Agent activity](/quest/m0/kix/activity.md) - agent cards and map highlights built from checkpoint summaries - [Mobile web app](/quest/m0/kix/mobile.md) - an installable PWA with Web Push for decisions - [Bot publication](/quest/m0/kix/publication.md) - validated agent results become normal GitHub PRs diff --git a/quest/m0/kix/activity.md b/quest/m0/kix/activity.md index f779e025..7900554e 100644 --- a/quest/m0/kix/activity.md +++ b/quest/m0/kix/activity.md @@ -33,4 +33,3 @@ Decided 2026-10-15 in the decisions-not-implementations interview: - [Runner](/quest/m0/kix/runner.md) - the checkpoint tool, summaries, and retained raw output - [Decision inbox](/quest/m0/kix/prompts.md) - the signed-in shell the cards sit beside -- [Output retention](/quest/m0/kix/output.md) - the bounded raw output behind the disclosure diff --git a/quest/m0/kix/output.md b/quest/m0/kix/output.md deleted file mode 100644 index 5fa05e35..00000000 --- a/quest/m0/kix/output.md +++ /dev/null @@ -1,61 +0,0 @@ -# [M] Retain raw agent output in R2 - -## Goal - -Raw agent output that runners upload is stored in R2 instead of D1, and kept -for 30 days with at most 64 MiB per attempt. When an attempt reaches the cap, -later chunks are refused with a `truncated` marker, while the job keeps running -and its summaries stay intact. The maintainer-only output endpoint keeps paging -by `(attempt, seq)`. D1 holds only a small index. Expired output, and the output -of disconnected projects, is deleted without operator action. - -## Plan - -Decided 2026-10-17 in the planning interview, as a follow-up of the merged -job-coordination PR (#135), which stored output text in D1 `job_output` rows -with no total cap: - -- m0, before [agent activity](/quest/m0/kix/activity.md), so launch does not - ship unbounded transcripts in D1. -- Store one R2 object per chunk at `output///`. A D1 index - row per chunk keeps paging and retry idempotency (same `seq` and text succeeds; a different chunk with the same - `seq` is refused). Objects are crash-safe and readable while a job runs. -- Cap: 64 MiB of UTF-8 per attempt, and a bound on chunks per attempt so many - tiny chunks cannot multiply rows and objects before the byte cap is reached. - Once either is reached, refuse further chunks with `truncated`, record the - truncation once so the disclosure shows it, and never fail the job. -- Make `seq` contiguous per attempt, starting at 0: the service accepts only - the next `seq`, or an idempotent retry of one whose row remains. The runner - protocol is unpublished and the runner sends chunks in order, so this - replaces today's acceptance of any unused `seq`. -- Keep a compact per-attempt record (bytes and chunks accepted, next `seq`, - truncated) that outlives the expiring index rows while the attempt can still - send output. Expiry then cannot give capacity back, and a replay of an - expired `seq` is refused rather than reclaimed. -- Retention: a bucket lifecycle rule deletes `output/` objects 30 days after - upload, and a daily Worker cron deletes their index rows. The clock runs from - each chunk's upload. Reads ignore an index row whose object is already gone. -- Disconnecting a project deletes its jobs' output objects immediately, along - with the job records it already deletes. R2 cannot join the D1 batch, so - reserve each chunk first: insert its index row with the text's hash, only - while the job exists and the `seq` is unused or holds the same hash. A - conflicting chunk is refused before anything reaches R2, so it can never - overwrite accepted output. Then write the object and mark the row complete; - if the row is gone by then, delete the object just written. A retry that - matches a completed row succeeds without writing again, so retries never - restart an object's 30-day lifecycle. Disconnect - records a cleanup row per job in the same D1 batch that deletes the - project, then deletes each `output//` prefix and its cleanup row. The - daily cron retries any cleanup row left by a crash or failed R2 call, so an - upload racing a disconnect is removed by one side or the other. -- No data migration: kix.quest is not deployed yet - ([deploy](/quest/m0/kix/deploy.md)), so replace `job_output.text` rather - than copying existing rows. -- Keep every read behind the existing per-request maintainer check, and test - that other users and revoked maintainers are refused. -- Documentation stays inline: update `cloudflare/README.md` (bucket binding, - lifecycle rule, cron, cap) and the runner protocol notes in the same change. - -## Related - -- [Runner](/quest/m0/kix/runner.md) - uploads the raw output this stores diff --git a/quest/m0/kix/runner.md b/quest/m0/kix/runner.md index 81bc7169..283c4505 100644 --- a/quest/m0/kix/runner.md +++ b/quest/m0/kix/runner.md @@ -37,9 +37,13 @@ replacing the hosted Agent SDK runtime: steps. The runner sends the summary and a work-in-progress bundle; the service publishes it to the job's fenced branch, so another runner can continue after reassignment. Runners never hold GitHub App credentials. -- Raw agent output is uploaded for [retention](/quest/m0/kix/output.md) - behind the maintainer-only disclosure; it is never streamed to the UI by - default. +- Raw agent output is uploaded as `output` chunks behind the maintainer-only + disclosure; it is never streamed to the UI by default. Send each attempt's + chunks in order with a contiguous `seq` from 0, batched so an attempt stays + under 4096 chunks, and retry a failed chunk with the same `seq` and text. A + `truncated` reply means the attempt reached its 64 MiB cap: stop uploading + and keep the job running. See the Worker's + [raw output notes](/cloudflare/README.md#raw-output). - A job that hits an account limit pauses with a decision to switch accounts. The runner never rotates accounts automatically. - Use the existing guide and skills as workflow sources. Explicitly load