From 187a90b7bc758a398c6d12c8c478dc9329d36a4f Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Thu, 30 Jul 2026 14:01:02 -0700 Subject: [PATCH] Add opt-in in-flight run-log mirroring with graceful-shutdown flush (#10512) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The run-log store records each agent run's output and can mirror completed logs to S3-compatible object storage > - The mirror uploads only on finalize, so a server restart mid-run loses the whole in-flight log > - Deployments and crashes are routine on ephemeral hosts, and lost run output makes failed runs impossible to debug > - This pull request adds an opt-in throttled mirror for still-running logs plus a graceful-shutdown flush > - The benefit is that a restart mid-run keeps the log tail up to the last mirror interval, and an orderly restart keeps everything ## Linked Issues or Issue Description No public issue exists — describing the feature inline (per the feature request template). **Subsystem affected** server/ — REST API & orchestration services **Problem or motivation** `RUN_LOG_S3_BUCKET` gives finished run logs durability, but the mirror uploads only on finalize. A run that is still writing when the server restarts leaves nothing in object storage. On hosts with ephemeral disks the local file is gone too, so the run's output is lost end to end and failed runs cannot be debugged. **Proposed solution** Mirror the in-flight log to the same object key on a throttled cadence (`RUN_LOG_S3_INFLIGHT_MIRROR_SECONDS`), and flush dirty tails during graceful shutdown. Keep it opt-in so existing deployments see zero new upload traffic unless they ask for it. **Alternatives considered** Per-append uploads (rejected: one PUT per output chunk is hostile to S3 endpoints and run latency). Chunked part objects with read-time stitching (rejected: complicates the read path, and S3 multipart minimum part sizes do not fit small tails). Persistent volumes (rejected upstream already: the data dir is deliberately an emptyDir in hardened cloud_tenant deployments). **Roadmap alignment** Not on ROADMAP.md; extends the existing run-log durability mirror without changing any default behavior. **Additional context** Ranged reads already serve partial objects like a live tail, so the read path needs no change; finalize overwrites the mirror with the complete file. **Related PRs (dedup search):** the finalize-only S3 mirror landed previously and this extends it; no duplicate or competing PR found for in-flight run-log mirroring. ## What Changed - `server/src/services/run-log-store.ts`: new opt-in `inflightMirrorMs` on the S3 options (`RUN_LOG_S3_INFLIGHT_MIRROR_SECONDS` env). When set, appends schedule at most one upload of the current file per interval, to the same key finalize uses. Ranged reads already serve that key, so a partial object behaves like a live tail and needs no read-path change. Finalize retires the in-flight bookkeeping and waits out an upload already on the wire, so a stale partial can never overwrite a finalized log. Upload failures warn, re-mark the tail dirty, and retry at most once per interval. - `server/src/services/run-log-store.ts`: new `flushInflightMirrors()` on the store and a module-level `flushInFlightRunLogMirrors()` for the shutdown path. Both are no-ops when the mirror is off. - `server/src/index.ts`: graceful shutdown flushes dirty in-flight tails after the heartbeat run drain, so runs the drain did not finalize (timeouts, the hot-restart skip path) still persist their output. - `server/src/services/run-log-store.test.ts`: five new tests — off-by-default (no uploads before finalize), tail preserved after a wipe without finalize, throttle coalescing with a single flush upload, finalize superseding the in-flight mirror and retiring its timer, and upload failures never breaking appends with recovery on the next flush. ## Verification - `pnpm vitest run server/src/services/run-log-store.test.ts` — 13 passed (8 existing + 5 new). - `pnpm vitest run server/src/__tests__/heartbeat-run-log.test.ts server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts` — 21 passed (consumers of the store, unchanged behavior). - `pnpm -C server run typecheck` — clean. - Self-hosted behavior is unchanged unless `RUN_LOG_S3_INFLIGHT_MIRROR_SECONDS` is set: with the variable unset there are zero new uploads and the finalize-only mirroring is byte-identical (asserted by the off-by-default test). ## Risks - Low. The feature is opt-in; unset env preserves today's behavior exactly. When enabled, worst case is one extra PUT per interval per active run, and every upload is best-effort — a failing endpoint warns and never breaks appends, finalization, or shutdown. The finalize path awaits any in-flight upload before writing the complete file, closing the only overwrite race the design introduces. Timers are `unref`ed so the mirror never keeps the process alive. ## Model Used Claude Fable 5 (`claude-fable-5`, Anthropic; Claude Code CLI with extended thinking and tool use; tests executed locally via Vitest). ## 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 (no duplicates found for in-flight run-log mirroring; the finalize-only mirror landed previously and this extends it) - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` 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 - [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 --- server/src/index.ts | 11 ++ server/src/services/run-log-store.test.ts | 170 +++++++++++++++++++++ server/src/services/run-log-store.ts | 171 +++++++++++++++++++++- 3 files changed, 349 insertions(+), 3 deletions(-) diff --git a/server/src/index.ts b/server/src/index.ts index 50a5278091..80523982a9 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -72,6 +72,7 @@ import { maybePersistWorktreeRuntimePorts } from "./worktree-config.js"; import { initTelemetry, getTelemetryClient } from "./telemetry.js"; import { conflict } from "./errors.js"; import { coordinateHeartbeatSchedulerShutdown } from "./shutdown.js"; +import { flushInFlightRunLogMirrors } from "./services/run-log-store.js"; import type { InstanceDatabaseBackupRunResult, InstanceDatabaseBackupTrigger, @@ -1325,6 +1326,16 @@ export async function startServer(): Promise { } } + // Whatever the drain did not finalize (timed-out runs, the hot-restart + // skip path) still has a local-only tail when the in-flight run-log + // mirror is enabled; upload those tails now so an orderly restart + // never loses run output. No-op when the mirror is off. + try { + await flushInFlightRunLogMirrors(); + } catch (err) { + logger.error({ err, signal }, "run-log in-flight mirror flush failed"); + } + const appShutdown = (app as { locals?: { paperclipShutdown?: () => void } }).locals?.paperclipShutdown; appShutdown?.(); diff --git a/server/src/services/run-log-store.test.ts b/server/src/services/run-log-store.test.ts index 029262ee07..276321c7db 100644 --- a/server/src/services/run-log-store.test.ts +++ b/server/src/services/run-log-store.test.ts @@ -166,3 +166,173 @@ describe("createDurableRunLogStore", () => { await expect(store.read(handle)).rejects.toThrow(/not found/i); }); }); + +describe("in-flight mirror", () => { + it("is OFF by default: appends never upload before finalize", async () => { + const { provider, calls } = createMemoryProvider(); + const store = createDurableRunLogStore({ basePath: baseDir, s3: { provider } }); + const handle = await store.begin(begin); + await store.append(handle, { stream: "stdout", chunk: "tail-1", ts: "t1" }); + await store.append(handle, { stream: "stdout", chunk: "tail-2", ts: "t2" }); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(calls.put).toBe(0); + // flush is a no-op with the mirror off + await store.flushInflightMirrors?.(); + expect(calls.put).toBe(0); + }); + + it("mirrors the running log after the interval so a wipe without finalize preserves the tail", async () => { + const { provider, objects } = createMemoryProvider(); + const store = createDurableRunLogStore({ + basePath: baseDir, + s3: { provider, keyPrefix: "run-logs", inflightMirrorMs: 10 }, + }); + const handle = await store.begin(begin); + await store.append(handle, { stream: "stdout", chunk: "crash-tail", ts: "t1" }); + const key = `run-logs/${handle.logRef}`; + await vi.waitFor(() => expect(objects.has(key)).toBe(true), { timeout: 2000 }); + // Simulate a crash mid-run: local dir wiped, run never finalized. + await fs.rm(baseDir, { recursive: true, force: true }); + const res = await store.read(handle); + expect(res.content).toContain("crash-tail"); + }); + + it("throttles: rapid appends coalesce into one upload, delivered by flush", async () => { + const { provider, objects, calls } = createMemoryProvider(); + // Interval far beyond the test's lifetime: nothing may upload until the + // flush, and the flush must carry every appended line in ONE put. + const store = createDurableRunLogStore({ + basePath: baseDir, + s3: { provider, keyPrefix: "run-logs", inflightMirrorMs: 60_000 }, + }); + const handle = await store.begin(begin); + for (let i = 0; i < 5; i++) { + await store.append(handle, { stream: "stdout", chunk: `line-${i}`, ts: `t${i}` }); + } + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(calls.put).toBe(0); // throttled: no per-append PUT + await store.flushInflightMirrors?.(); + expect(calls.put).toBe(1); + const body = objects.get(`run-logs/${handle.logRef}`)!.toString("utf8"); + for (let i = 0; i < 5; i++) expect(body).toContain(`line-${i}`); + // Nothing dirty afterwards: flushing again uploads nothing. + await store.flushInflightMirrors?.(); + expect(calls.put).toBe(1); + }); + + it("finalize supersedes the in-flight mirror and retires its timer", async () => { + const { provider, objects, calls } = createMemoryProvider(); + const store = createDurableRunLogStore({ + basePath: baseDir, + s3: { provider, keyPrefix: "run-logs", inflightMirrorMs: 10 }, + }); + const handle = await store.begin(begin); + await store.append(handle, { stream: "stdout", chunk: "early-tail", ts: "t1" }); + const key = `run-logs/${handle.logRef}`; + await vi.waitFor(() => expect(objects.has(key)).toBe(true), { timeout: 2000 }); + await store.append(handle, { stream: "stdout", chunk: "final-line", ts: "t2" }); + await store.finalize(handle); + const body = objects.get(key)!.toString("utf8"); + expect(body).toContain("early-tail"); + expect(body).toContain("final-line"); + // No retired timer fires a stale partial upload over the finalized log. + const putsAfterFinalize = calls.put; + await new Promise((resolve) => setTimeout(resolve, 60)); + expect(calls.put).toBe(putsAfterFinalize); + expect(objects.get(key)!.toString("utf8")).toContain("final-line"); + }); + + it("shutdown flush re-checks dirtiness after an upload already on the wire completes", async () => { + const { provider, objects, calls } = createMemoryProvider(); + let releaseFirstPut: (() => void) | null = null; + const firstPutGate = new Promise((resolve) => { + releaseFirstPut = resolve; + }); + let putStarts = 0; + const realPut = provider.putObject.bind(provider); + provider.putObject = async (input) => { + putStarts++; + if (putStarts === 1) await firstPutGate; // hold the first upload on the wire + return realPut(input); + }; + const store = createDurableRunLogStore({ + basePath: baseDir, + s3: { provider, keyPrefix: "run-logs", inflightMirrorMs: 10 }, + }); + const handle = await store.begin(begin); + await store.append(handle, { stream: "stdout", chunk: "mirrored-line", ts: "t1" }); + await vi.waitFor(() => expect(putStarts).toBe(1), { timeout: 2000 }); + // Re-dirty the entry while the first upload is still in flight. + await store.append(handle, { stream: "stdout", chunk: "tail-during-upload", ts: "t2" }); + const flush = store.flushInflightMirrors!(); + releaseFirstPut!(); + await flush; + // A single-pass flush would exit after the first upload and leave the + // tail on an unref'ed timer that never fires once the process exits. + const body = objects.get(`run-logs/${handle.logRef}`)!.toString("utf8"); + expect(body).toContain("tail-during-upload"); + expect(calls.put).toBeGreaterThanOrEqual(2); + }); + + it("bounds the in-flight upload to the stat'ed size when appends race the stream", async () => { + const { provider, objects } = createMemoryProvider(); + const store = createDurableRunLogStore({ + basePath: baseDir, + s3: { provider, keyPrefix: "run-logs", inflightMirrorMs: 60_000 }, + }); + const handle = await store.begin(begin); + await store.append(handle, { stream: "stdout", chunk: "counted-line", ts: "t1" }); + const absPath = path.join(baseDir, handle.logRef); + const sizeAtStat = (await fs.stat(absPath)).size; + // Grow the file between the mirror's stat() and its stream reaching EOF: + // the upload must carry exactly the stat'ed bytes, not the racing tail + // (an unbounded stream would violate the declared contentLength). + const realStat = fs.stat.bind(fs); + const statSpy = vi.spyOn(fs, "stat").mockImplementation(async (target, ...rest) => { + const result = await realStat(target as Parameters[0], ...(rest as [])); + if (String(target).endsWith(".ndjson")) { + await fs.appendFile(absPath, `${JSON.stringify({ ts: "t2", stream: "stdout", chunk: "raced-append" })}\n`); + } + return result; + }); + try { + await store.flushInflightMirrors!(); + } finally { + statSpy.mockRestore(); + } + const body = objects.get(`run-logs/${handle.logRef}`)!; + expect(body.length).toBe(sizeAtStat); + expect(body.toString("utf8")).toContain("counted-line"); + expect(body.toString("utf8")).not.toContain("raced-append"); + }); + + it("in-flight upload failures never break appends and recover on the next flush", async () => { + const { provider, objects, calls } = createMemoryProvider(); + let failPuts = true; + const realPut = provider.putObject.bind(provider); + provider.putObject = async (input) => { + if (failPuts) throw new Error("endpoint down"); + return realPut(input); + }; + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + try { + const store = createDurableRunLogStore({ + basePath: baseDir, + s3: { provider, keyPrefix: "run-logs", inflightMirrorMs: 60_000 }, + }); + const handle = await store.begin(begin); + await store.append(handle, { stream: "stdout", chunk: "survives-outage", ts: "t1" }); + await store.flushInflightMirrors?.(); // upload fails, append flow unaffected + expect(warn).toHaveBeenCalled(); + await store.append(handle, { stream: "stdout", chunk: "post-outage", ts: "t2" }); + failPuts = false; + await store.flushInflightMirrors?.(); + const body = objects.get(`run-logs/${handle.logRef}`)!.toString("utf8"); + expect(body).toContain("survives-outage"); + expect(body).toContain("post-outage"); + expect(calls.put).toBe(1); // only the successful upload reached storage + } finally { + warn.mockRestore(); + } + }); +}); diff --git a/server/src/services/run-log-store.ts b/server/src/services/run-log-store.ts index 8bf10d1ed4..90f7f975fe 100644 --- a/server/src/services/run-log-store.ts +++ b/server/src/services/run-log-store.ts @@ -37,6 +37,10 @@ export interface RunLogStore { ): Promise; finalize(handle: RunLogHandle): Promise; read(handle: RunLogHandle, opts?: RunLogReadOptions): Promise; + // Optional so existing fakes/fixtures keep compiling: uploads every dirty + // in-flight mirror immediately (graceful-shutdown path). No-op when the + // in-flight mirror is not enabled. + flushInflightMirrors?(): Promise; } function safeSegments(...segments: string[]) { @@ -63,7 +67,16 @@ export interface DurableRunLogStoreOptions { // served from there on read whenever the local file is missing (e.g. the pod // rolled and wiped the emptyDir). When omitted, the store is local-only (the // historical behaviour: a restart loses the log). - s3?: { provider: StorageProvider; keyPrefix?: string }; + s3?: { + provider: StorageProvider; + keyPrefix?: string; + // When > 0, ALSO mirror the still-running log to the same object key at + // most once per this interval (plus a flush hook for graceful shutdown), + // so a crash mid-run loses at most one interval's tail instead of the + // whole log. Off (undefined/0) preserves the historical finalize-only + // mirroring: no extra PUT traffic unless explicitly opted in. + inflightMirrorMs?: number; + }; } // Run-log store with TRANSPARENT durability. The store id stays "local_file" so @@ -75,15 +88,114 @@ export interface DurableRunLogStoreOptions { // for "Run log not found" after a deploy/restart (the /paperclip data dir is an // emptyDir in cloud_tenant mode -- persistence is disabled to avoid the // operator's privileged selinux-relabel init container in our hardened ns). +// +// Optionally (inflightMirrorMs > 0) the still-running log is ALSO mirrored to +// the same key at a throttled cadence and flushed on graceful shutdown, so a +// restart mid-run preserves the tail up to the last mirror instead of losing +// the whole in-flight log. Finalize retires the in-flight bookkeeping (waiting +// out any upload already on the wire) before writing the complete file, so a +// stale partial can never overwrite a finalized log. export function createDurableRunLogStore(options: DurableRunLogStoreOptions): RunLogStore { const { basePath } = options; const s3 = options.s3; const s3Prefix = normalizeKeyPrefix(s3?.keyPrefix); + const inflightMirrorMs = s3?.inflightMirrorMs && s3.inflightMirrorMs > 0 ? s3.inflightMirrorMs : 0; function s3Key(logRef: string): string { return s3Prefix ? `${s3Prefix}/${logRef}` : logRef; } + // In-flight mirror bookkeeping, keyed by logRef. The mirror uploads the + // CURRENT (partial) file to the SAME key finalize uses: readers already + // range-read that key, so a partial object is served exactly like a live + // tail, and finalize simply overwrites it with the complete file. One + // entry exists only between the first post-interval-eligible append and + // finalize. + interface InflightMirrorEntry { + dirty: boolean; + lastMirrorAt: number; + timer: NodeJS.Timeout | null; + upload: Promise | null; + } + const inflightMirrors = new Map(); + + function mirrorInflightNow(logRef: string, entry: InflightMirrorEntry): Promise { + entry.dirty = false; + const upload = (async () => { + const absPath = resolveWithin(basePath, logRef); + const stat = await fs.stat(absPath); + if (stat.size === 0) return true; + await s3!.provider.putObject({ + objectKey: s3Key(logRef), + // Bound the stream to the stat'ed size: the run is still appending, + // and an unbounded stream that grows past stat.size would violate + // the declared contentLength and fail (or truncate) the upload. + // Bytes appended after the stat stay dirty and ride the next mirror. + body: createReadStream(absPath, { start: 0, end: stat.size - 1 }), + contentType: "application/x-ndjson", + contentLength: stat.size, + }); + return true; + })().catch((err) => { + // Best-effort like the finalize mirror: a failing upload must never + // break the run, but a persistently broken mirror should be visible. + console.warn( + `[run-log-store] Failed to mirror in-flight run log to object storage (key: ${s3Key(logRef)}):`, + err, + ); + // Re-dirty so the tail retries next interval even without new appends; + // the lastMirrorAt stamp below bounds retries to one per interval. + entry.dirty = true; + return false; + }).finally(() => { + // Stamp AFTER the attempt so a slow or failing endpoint self-throttles + // to one attempt per interval instead of hot-looping. + entry.lastMirrorAt = Date.now(); + entry.upload = null; + if (entry.dirty) scheduleInflightMirror(logRef, entry); + }); + entry.upload = upload; + return upload; + } + + function scheduleInflightMirror(logRef: string, entry: InflightMirrorEntry): void { + if (entry.timer || entry.upload) return; + const delay = Math.max(0, inflightMirrorMs - (Date.now() - entry.lastMirrorAt)); + entry.timer = setTimeout(() => { + entry.timer = null; + void mirrorInflightNow(logRef, entry); + }, delay); + // Never keep the process alive just to mirror a tail. + entry.timer.unref?.(); + } + + function noteInflightAppend(logRef: string): void { + if (!s3 || inflightMirrorMs <= 0) return; + let entry = inflightMirrors.get(logRef); + if (!entry) { + // First mirror lands one full interval after the first append: a run + // that finalizes sooner is covered by the finalize upload, and this + // keeps the steady-state cost at one PUT per interval per active run. + entry = { dirty: false, lastMirrorAt: Date.now(), timer: null, upload: null }; + inflightMirrors.set(logRef, entry); + } + entry.dirty = true; + scheduleInflightMirror(logRef, entry); + } + + async function retireInflightMirror(logRef: string): Promise { + const entry = inflightMirrors.get(logRef); + if (!entry) return; + inflightMirrors.delete(logRef); + if (entry.timer) { + clearTimeout(entry.timer); + entry.timer = null; + } + // An upload still in flight could otherwise finish AFTER finalize's + // complete-file upload and overwrite it with a stale partial. + if (entry.upload) await entry.upload; + } + async function ensureDir(relativeDir: string) { const dir = resolveWithin(basePath, relativeDir); await fs.mkdir(dir, { recursive: true }); @@ -165,6 +277,7 @@ export function createDurableRunLogStore(options: DurableRunLogStoreOptions): Ru await ensureDir(relDir); const absPath = resolveWithin(basePath, relPath); await fs.writeFile(absPath, "", "utf8"); + await retireInflightMirror(relPath); return { store: "local_file", logRef: relPath }; }, @@ -182,11 +295,13 @@ export function createDurableRunLogStore(options: DurableRunLogStoreOptions): Ru }); const persisted = `${line}\n`; await fs.appendFile(absPath, persisted, "utf8"); + noteInflightAppend(handle.logRef); return Buffer.byteLength(persisted, "utf8"); }, async finalize(handle) { if (handle.store !== "local_file") return { bytes: 0, compressed: false }; + await retireInflightMirror(handle.logRef); const absPath = resolveWithin(basePath, handle.logRef); const stat = await fs.stat(absPath).catch(() => null); if (!stat) throw notFound("Run log not found"); @@ -231,6 +346,39 @@ export function createDurableRunLogStore(options: DurableRunLogStoreOptions): Ru // Local file gone (pod rolled) -> serve from the S3 mirror if configured. return readS3Range(handle.logRef, offset, limitBytes); }, + + async flushInflightMirrors() { + if (!s3 || inflightMirrorMs <= 0) return; + const flushEntry = async (logRef: string, entry: InflightMirrorEntry) => { + // Loop until the entry is clean: an append that lands while an + // upload is on the wire re-dirties the entry, and its follow-up + // mirror sits on an unref'ed timer that would never fire once the + // process exits — so re-check after every await instead of trusting + // a single pass. A FAILED attempt ends the loop instead of retrying: + // hot-looping a down endpoint at shutdown would spin forever, and + // the flush is best-effort by design. + for (;;) { + if (entry.timer) { + clearTimeout(entry.timer); + entry.timer = null; + } + if (entry.upload) { + await entry.upload; + continue; + } + if (!entry.dirty) return; + const uploaded = await mirrorInflightNow(logRef, entry); + if (!uploaded) { + if (entry.timer) { + clearTimeout(entry.timer); + entry.timer = null; + } + return; + } + } + }; + await Promise.all([...inflightMirrors].map(([logRef, entry]) => flushEntry(logRef, entry))); + }, }; } @@ -239,7 +387,7 @@ export function createDurableRunLogStore(options: DurableRunLogStoreOptions): Ru // NOT redirect the product's workspace/file storage (smaller blast radius). // Unset RUN_LOG_S3_BUCKET -> no mirror -> local-only (safe degrade). Creds come // from the standard AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY chain. -function resolveRunLogS3(): { provider: StorageProvider; keyPrefix?: string } | undefined { +function resolveRunLogS3(): DurableRunLogStoreOptions["s3"] { const bucket = process.env.RUN_LOG_S3_BUCKET?.trim(); if (!bucket) return undefined; const provider = createS3StorageProvider({ @@ -251,7 +399,16 @@ function resolveRunLogS3(): { provider: StorageProvider; keyPrefix?: string } | ? process.env.RUN_LOG_S3_FORCE_PATH_STYLE === "true" : true, // Cubbit (and most S3-compatible endpoints) need path-style }); - return { provider, keyPrefix: process.env.RUN_LOG_S3_PREFIX?.trim() || "run-logs" }; + // Opt-in in-flight tail mirroring: at most one partial upload per interval + // per active run, so a crash loses at most one interval's tail. Unset/0 + // keeps the historical finalize-only mirroring. + const inflightSeconds = Number.parseFloat(process.env.RUN_LOG_S3_INFLIGHT_MIRROR_SECONDS ?? ""); + return { + provider, + keyPrefix: process.env.RUN_LOG_S3_PREFIX?.trim() || "run-logs", + inflightMirrorMs: + Number.isFinite(inflightSeconds) && inflightSeconds > 0 ? Math.round(inflightSeconds * 1000) : undefined, + }; } let cachedStore: RunLogStore | null = null; @@ -262,3 +419,11 @@ export function getRunLogStore() { cachedStore = createDurableRunLogStore({ basePath, s3: resolveRunLogS3() }); return cachedStore; } + +// Graceful-shutdown hook: upload every dirty in-flight run-log tail before +// the process exits, so an orderly restart (deploy, SIGTERM) loses nothing +// even for runs that never reach finalize. No-op when the store was never +// created or in-flight mirroring is off. +export async function flushInFlightRunLogMirrors(): Promise { + await cachedStore?.flushInflightMirrors?.(); +}