diff --git a/server/src/services/object-byte-journal.ts b/server/src/services/object-byte-journal.ts index 7ece64a9f32..f9a5b3003ae 100644 --- a/server/src/services/object-byte-journal.ts +++ b/server/src/services/object-byte-journal.ts @@ -476,7 +476,8 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) { if (!Number.isSafeInteger(sourceCursor) || sourceCursor < writer.head.sourceCursor) throw new Error("object_journal_invalid_source_prefix_cursor"); const current = await head(signal); - if (!current || current.etag !== writer.etag || current.head.closed || current.head.deleted) + if (!current) throw new ObjectJournalSourcePrefixProofUnavailableError(writer); + if (current.etag !== writer.etag || current.head.closed || current.head.deleted) throw new ObjectJournalConflictError(); // One-time verification can repair legacy max-cursor metadata, never payload bytes. if (sourceCursor !== writer.head.sourceCursor) diff --git a/server/src/services/object-run-log-store.test.ts b/server/src/services/object-run-log-store.test.ts index 754a776965d..2a637780194 100644 --- a/server/src/services/object-run-log-store.test.ts +++ b/server/src/services/object-run-log-store.test.ts @@ -289,6 +289,40 @@ describe("S3 run log store", () => { .rejects.toThrow("object_journal_source_prefix_proof_invalid"); }); + it("resumes a run log through the prefix read and proof while the store still hides the head it just published", async () => { + const m = memory(), + options = { provider: m.provider, keyPrefix: "logs", flushMs: 10, assertWriterAuthority: async () => undefined }; + const lagged = { ...input, runId: "lagged-run" }; + const a = createObjectRunLogStore(options), + handle = await a.begin(lagged); + await Promise.all([1, 2, 3].map((seq) => a.append(handle, event(seq)))); + const b = createObjectRunLogStore(options), + successor = await b.begin(lagged); + const original = m.provider.getObject; + let hiddenHeadReads = 0; + m.provider.getObject = async (i) => { + if (i.objectKey === "logs/segments-v1/company/agent/lagged-run/head.json") { + hiddenHeadReads++; + throw Object.assign(Error("not yet visible"), { name: "NoSuchKey", $metadata: { httpStatusCode: 404 } }); + } + return original(i); + }; + const read = await b.read(successor, { limitBytes: successor.initialBytes }); + expect(read.content.trim().split("\n").map((line) => JSON.parse(line).seq)).toEqual([1, 2, 3]); + await b.recordVerifiedSourcePrefix!(successor, 0); + expect(successor.initialSourcePrefixVerified).toBe(false); + expect(hiddenHeadReads).toBeGreaterThanOrEqual(2); + m.provider.getObject = original; + await b.append(successor, event(4)); + const resumed = await b.read(successor); + expect(resumed.content.trim().split("\n").map((line) => JSON.parse(line).seq)).toEqual([1, 2, 3, 4]); + }); + it("fails a read of a run log that was never written", async () => { + const m = memory(), + store = createObjectRunLogStore({ provider: m.provider, keyPrefix: "logs", assertWriterAuthority: async () => undefined }); + await expect(store.read({ store: "s3_segments", logRef: "company/agent/never-written", attemptId: "attempt" })) + .rejects.toThrow("Object run log not found"); + }); it("batches concurrent appends, confirms durability, and resumes offsets/sequence on a replacement", async () => { const m = memory(), options = { diff --git a/server/src/services/object-run-log-store.ts b/server/src/services/object-run-log-store.ts index 97cf79fd7f5..1065d61d6aa 100644 --- a/server/src/services/object-run-log-store.ts +++ b/server/src/services/object-run-log-store.ts @@ -356,7 +356,7 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL async read(handle, opts = {}) { if (handle.store !== "s3_segments" && handle.store !== "s3_archive") throw new Error("Invalid object run log backend"); const store = journal(handle.logRef, opts.priority ?? "interactive"), - snapshot = await store.head(opts.signal); + snapshot = await store.head(opts.signal) ?? writers.get(handle)?.writer; if (!snapshot || snapshot.head.deleted) throw new Error("Object run log not found"); const offset = Math.max(0, opts.offset ?? 0), limit = opts.limitBytes ?? 256_000;