diff --git a/server/src/__tests__/company-skills-runtime-selection.test.ts b/server/src/__tests__/company-skills-runtime-selection.test.ts index b2ae2568caf..9f95ad76938 100644 --- a/server/src/__tests__/company-skills-runtime-selection.test.ts +++ b/server/src/__tests__/company-skills-runtime-selection.test.ts @@ -98,6 +98,35 @@ describeDatabase("selected runtime skill reconciliation", () => { expect(await fs.readFile(path.join(old,"SKILL.md"),"utf8")).toBe("Old installation content"); }); + it("does not re-authorize each delivered skill version after the selected rows were read", async () => { + const [agent] = await db.insert(agents).values({ companyId, name: "Versioned runtime reader", adapterType: "codex_local" }).returning(); + await db.insert(companyMemberships).values({ companyId, principalType: "agent", principalId: agent!.id, membershipRole: "member" }); + const actor: AuthorizationActor = { type: "agent", agentId: agent!.id, companyId, source: "agent_jwt" }; + const installed = await withSkillActor(actor, () => companySkillService(db).listRuntimeSkillEntries(companyId, { + selectionConfig: { paperclipSkillSync: { desiredSkills: [] } }, includeLegacy: true, snapshotLatest: true, + })); + const read = () => withSkillActor(actor, () => companySkillService(db).listRuntimeSkillEntries(companyId, { + onlyKeys: new Set(installed.map(entry => entry.key)), snapshotLatest: true, + })); + const sourceReads: unknown[] = []; + const select = db.select.bind(db); + const spy = vi.spyOn(db, "select").mockImplementation(((...args: Parameters) => { + const builder = select(...args); + const from = builder.from.bind(builder); + builder.from = ((table: unknown) => { + if (table === companySkillSources) sourceReads.push(table); + return from(table as never); + }) as typeof builder.from; + return builder; + }) as typeof db.select); + try { + const entries = await read(); + expect(entries.length).toBeGreaterThan(2); + expect(entries.every(entry => entry.sourceStatus === "available" && entry.versionId)).toBe(true); + expect(sourceReads).toHaveLength(0); + } finally { spy.mockRestore(); } + }, 30_000); + it("bounds project-source authorization work without caching changed permissions", async () => { const [agent] = await db.insert(agents).values({ companyId, name: "Selected runtime reader", adapterType: "codex_local" }).returning(); await db.insert(companyMemberships).values({ companyId, principalType: "agent", principalId: agent!.id, membershipRole: "member" }); diff --git a/server/src/services/company-skills.ts b/server/src/services/company-skills.ts index ea18c85e475..0be3726541d 100644 --- a/server/src/services/company-skills.ts +++ b/server/src/services/company-skills.ts @@ -5981,7 +5981,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo } async function materializeVersionSnapshot(companyId: string, skill: CompanySkill, version: StoredSkillVersion): Promise { - return executionContent.withVersion({ companyId, skillId: skill.id, versionId: version.id }, async directory => directory); + return executionContent.withReadableVersion({ companyId, skillId: skill.id, versionId: version.id }, async directory => directory); } function resolveRuntimeSkillMaterializedPath(companyId: string, skill: Pick) { diff --git a/server/src/services/object-byte-journal.test.ts b/server/src/services/object-byte-journal.test.ts index 4412249353c..d0eae2640d7 100644 --- a/server/src/services/object-byte-journal.test.ts +++ b/server/src/services/object-byte-journal.test.ts @@ -96,7 +96,6 @@ describe("object byte journal", () => { const controller = new AbortController(); const a = createObjectByteJournal({ provider: s.provider, prefix: "company/a/log" }); const first = a.claim({ - expectedOwner: null, assertAuthority: authority, signal: controller.signal, }); @@ -104,7 +103,6 @@ describe("object byte journal", () => { await probing; const b = createObjectByteJournal({ provider: s.provider, prefix: "company/b/log" }); const second = b.claim({ - expectedOwner: null, assertAuthority: async () => { admitted(); }, @@ -121,7 +119,7 @@ describe("object byte journal", () => { it("recovers committed bytes and stable ranges with no local filesystem", async () => { const s = storage(), a = createObjectByteJournal({ provider: s.provider, prefix: "company/run/log" }); - let writer = await a.claim({ expectedOwner: null, assertAuthority: authority }); + let writer = await a.claim({ assertAuthority: authority }); const expected = Buffer.from("first\n再開\n".repeat(20)); writer = await a.append(writer, expected.subarray(0, 17)); writer = await a.append(writer, expected.subarray(17)); @@ -134,10 +132,10 @@ describe("object byte journal", () => { it("fences stale appends/finalization and preserves a pinned reader snapshot", async () => { const s = storage(), a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); - let old = await a.claim({ expectedOwner: null, assertAuthority: authority }); + let old = await a.claim({ assertAuthority: authority }); old = await a.append(old, Buffer.from("before")); const snapshot = old.head; - let next = await a.claim({ expectedOwner: old.head.owner, assertAuthority: authority }); + let next = await a.claim({ assertAuthority: authority }); next = await a.append(next, Buffer.from("after")); await expect(a.append(old, Buffer.from("lost"))).rejects.toBeInstanceOf( ObjectJournalConflictError, @@ -149,7 +147,7 @@ describe("object byte journal", () => { it("reconciles an uncertain manifest commit without duplicating output", async () => { const s = storage(), a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); - let writer = await a.claim({ expectedOwner: null, assertAuthority: authority }); + let writer = await a.claim({ assertAuthority: authority }); s.loseResponse(); writer = await a.append(writer, Buffer.from("once")); expect(writer.head.end).toBe(4); @@ -158,7 +156,7 @@ describe("object byte journal", () => { it("keeps index objects bounded over multiple tree levels", async () => { const s = storage(), a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); - let writer = await a.claim({ expectedOwner: null, assertAuthority: authority }); + let writer = await a.claim({ assertAuthority: authority }); for (let i = 0; i < 180; i++) writer = await a.append(writer, Buffer.from(String(i).padStart(4, "0"))); expect(await a.readRange(writer.head, 252, 16)).toEqual(Buffer.from("0063006400650066")); @@ -180,7 +178,7 @@ describe("object byte journal", () => { it("retries a transient head conflict without fencing its current writer", async () => { const s = storage(), journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); - let writer = await journal.claim({ expectedOwner: null, assertAuthority: authority }); + let writer = await journal.claim({ assertAuthority: authority }); const original = s.provider.putObjectConditional!; let conflicts = 0; s.provider.putObjectConditional = async (input) => { @@ -205,14 +203,14 @@ describe("object byte journal", () => { }; const journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); await expect( - journal.claim({ expectedOwner: null, assertAuthority: authority }), + journal.claim({ assertAuthority: authority }), ).rejects.toThrow("conflict"); expect(s.objects.size).toBe(0); }); it("retries transient segment failures and acknowledges exactly one batch", async () => { const s = storage(), journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); - let writer = await journal.claim({ expectedOwner: null, assertAuthority: authority }); + let writer = await journal.claim({ assertAuthority: authority }); const original = s.provider.putObjectConditional!; let failures = 2; s.provider.putObjectConditional = async (input) => { @@ -226,7 +224,7 @@ describe("object byte journal", () => { it("rejects corrupt published data and cancelled reads", async () => { const s = storage(), a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); - let writer = await a.claim({ expectedOwner: null, assertAuthority: authority }); + let writer = await a.claim({ assertAuthority: authority }); writer = await a.append(writer, Buffer.from("data")); const data = [...s.objects].find(([, bytes]) => bytes.toString() === "data")!; s.objects.set(data[0], Buffer.from("oops")); @@ -236,7 +234,7 @@ describe("object byte journal", () => { it("sheds only interactive reads under saturated object IO and never a writer", async () => { const s = storage(), a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/busy" }); - let writer = await a.claim({ expectedOwner: null, assertAuthority: authority }); + let writer = await a.claim({ assertAuthority: authority }); writer = await a.append(writer, Buffer.from("data")); const get = s.provider.getObject; let open!: () => void; @@ -255,7 +253,7 @@ describe("object byte journal", () => { const reads = Array.from({ length: 40 }, () => reader.head()); await expect(reader.head()).rejects.toBeInstanceOf(ObjectJournalBusyError); const other = createObjectByteJournal({ provider: s.provider, prefix: "c/r/writer" }); - const claimed = other.claim({ expectedOwner: null, assertAuthority: authority }); + const claimed = other.claim({ assertAuthority: authority }); open(); const otherWriter = await other.append(await claimed, Buffer.from("more")); expect(otherWriter.head.end).toBe(4); @@ -264,7 +262,7 @@ describe("object byte journal", () => { it("releases a queued read's place when its caller aborts", async () => { const s = storage(), a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/abort" }); - let writer = await a.claim({ expectedOwner: null, assertAuthority: authority }); + let writer = await a.claim({ assertAuthority: authority }); writer = await a.append(writer, Buffer.from("data")); const get = s.provider.getObject; let open!: () => void; @@ -292,14 +290,14 @@ describe("object byte journal", () => { it("retains a deletion tombstone and denies a subsequent claim", async () => { const s = storage(), a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); - let writer = await a.claim({ expectedOwner: null, assertAuthority: authority }); + let writer = await a.claim({ assertAuthority: authority }); writer = await a.append(writer, Buffer.from("private")); await a.close(writer); await a.remove(); expect([...s.objects.keys()]).toEqual(["c/r/log/head.json"]); expect((await a.head())!.head.deleted).toBe(true); await expect( - a.claim({ expectedOwner: writer.head.owner, assertAuthority: authority }), + a.claim({ assertAuthority: authority }), ).rejects.toThrow(); }); it("does not claim ownership when execution authority rejects", async () => { @@ -307,7 +305,6 @@ describe("object byte journal", () => { a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); await expect( a.claim({ - expectedOwner: null, assertAuthority: async () => { throw new Error("revoked"); }, @@ -326,7 +323,6 @@ describe("object byte journal", () => { release = resolve; }); const pending = journal.claim({ - expectedOwner: null, assertAuthority: async () => { enter(); await blocked; @@ -348,7 +344,7 @@ describe("object byte journal", () => { }; const journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); await expect( - journal.claim({ expectedOwner: null, assertAuthority: authority }), + journal.claim({ assertAuthority: authority }), ).rejects.toThrow("does not enforce conditional"); expect(s.objects.size).toBe(0); }); @@ -356,7 +352,7 @@ describe("object byte journal", () => { vi.useFakeTimers({ toFake: ["Date"] }); const s = storage(), journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); - let writer = await journal.claim({ expectedOwner: null, assertAuthority: authority }); + let writer = await journal.claim({ assertAuthority: authority }); writer = await journal.append(writer, Buffer.from("private")); await expect(journal.remove()).rejects.toThrow("cleanup_pending"); expect((await journal.head())!.head.deleted).toBe(true); @@ -370,7 +366,7 @@ describe("object byte journal", () => { it("does not publish a partial batch when a later segment fails", async () => { const s = storage(), journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" }); - const writer = await journal.claim({ expectedOwner: null, assertAuthority: authority }); + const writer = await journal.claim({ assertAuthority: authority }); const original = s.provider.putObjectConditional!; const failed = Buffer.alloc(1024 * 1024, 2); const key = createHash("sha256").update(failed).digest("hex"); diff --git a/server/src/services/object-byte-journal.ts b/server/src/services/object-byte-journal.ts index 38cd52c527b..dfbaf23273f 100644 --- a/server/src/services/object-byte-journal.ts +++ b/server/src/services/object-byte-journal.ts @@ -531,13 +531,11 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) { }, signal); } async function claim(input: { - expectedOwner: string | null; assertAuthority: () => Promise; signal?: AbortSignal; }) { const current = await head(input.signal); - if (current?.head.deleted || (current?.head.owner ?? null) !== input.expectedOwner) - throw new ObjectJournalConflictError(); + if (current?.head.deleted) throw new ObjectJournalConflictError(); await input.assertAuthority(); await qualifyConditionalWrites(provider, prefix, input.signal); await input.assertAuthority(); diff --git a/server/src/services/object-run-log-store.test.ts b/server/src/services/object-run-log-store.test.ts index 6016a931ce2..754a776965d 100644 --- a/server/src/services/object-run-log-store.test.ts +++ b/server/src/services/object-run-log-store.test.ts @@ -118,6 +118,22 @@ describe("S3 run log store", () => { expect((await store.finalize(handle)).bytes).toBe(Buffer.byteLength(content)); }); + it("begins a new run log with one head read and no source proof read", async () => { + const m = memory(), options = { provider: m.provider, keyPrefix: "logs", flushMs: 1, + assertWriterAuthority: async () => undefined }; + const get = m.provider.getObject.bind(m.provider), reads: string[] = []; + m.provider.getObject = async value => { reads.push(value.objectKey); return get(value); }; + const headReads = () => reads.filter(key => key.endsWith("/head.json")).length; + const proofReads = () => reads.filter(key => key.includes("/objects/source-prefix-proof-v1-")).length; + const handle = await createObjectRunLogStore(options).begin(input); + expect(handle.initialSourcePrefixVerified).toBe(false); + expect({ head: headReads(), proof: proofReads() }).toEqual({ head: 1, proof: 0 }); + reads.length = 0; + const resumed = await createObjectRunLogStore(options).begin(input); + expect(resumed.initialSourcePrefixVerified).toBe(false); + expect({ head: headReads(), proof: proofReads() }).toEqual({ head: 1, proof: 1 }); + }); + it("keeps live byte appends deliverable after a proof-only outage and requires cold prefix verification", async () => { const m = memory(), options = { provider: m.provider, keyPrefix: "logs", flushMs: 1, assertWriterAuthority: async () => undefined }; diff --git a/server/src/services/object-run-log-store.ts b/server/src/services/object-run-log-store.ts index d37f4f700de..97cf79fd7f5 100644 --- a/server/src/services/object-run-log-store.ts +++ b/server/src/services/object-run-log-store.ts @@ -254,14 +254,12 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL return { async begin(input) { const logRef = `${input.companyId}/${input.agentId}/${input.runId}`; - const store = journal(logRef), - current = await store.head(input.signal); + const store = journal(logRef); const writer = await store.claim({ - expectedOwner: current?.head.owner ?? null, assertAuthority: () => options.assertWriterAuthority(input), signal: input.signal, }); - const sourcePrefixVerified = await store.hasVerifiedSourcePrefix(writer.head, input.signal); + const sourcePrefixVerified = writer.head.generation > 1 && await store.hasVerifiedSourcePrefix(writer.head, input.signal); const handle: RunLogHandle = { store: "s3_segments", logRef, diff --git a/server/src/services/provider-trace-object-store.ts b/server/src/services/provider-trace-object-store.ts index 308afdbf8c1..5d9bf61074c 100644 --- a/server/src/services/provider-trace-object-store.ts +++ b/server/src/services/provider-trace-object-store.ts @@ -172,9 +172,7 @@ export function createProviderTraceObjectStore( try { for (const source of channelsFor(ref, filePath)) { capture.signal.throwIfAborted(); - const current = await source.journal.head(capture.signal); const writer = await source.journal.claim({ - expectedOwner: current?.head.owner ?? null, assertAuthority, signal: capture.signal, }); diff --git a/server/src/services/run-evidence-migration.ts b/server/src/services/run-evidence-migration.ts index 50d0ba84ebc..2aee95888ac 100644 --- a/server/src/services/run-evidence-migration.ts +++ b/server/src/services/run-evidence-migration.ts @@ -160,7 +160,6 @@ export async function migrateRunEvidence( let writer: Awaited> | null = null; try { writer = options.apply ? await journal.claim({ - expectedOwner: (await journal.head(options.signal))?.head.owner ?? null, signal: options.signal, assertAuthority: async () => { if (scopedArchive) diff --git a/server/src/services/skill-execution-content.ts b/server/src/services/skill-execution-content.ts index 50c19517e53..68ebb9bdf50 100644 --- a/server/src/services/skill-execution-content.ts +++ b/server/src/services/skill-execution-content.ts @@ -17,8 +17,11 @@ export function skillExecutionContent(db: Db, options: { actor?: AuthorizationAc const packages = skillPackageStore(options.packageDb ?? db); const runtime = skillPackageRuntime(options.packageDb ?? db); async function withVersion(reference: SkillContentReference, consume: (directory: string) => Promise): Promise { + await skillAccessService(db, options.actor).assertSkill(reference.companyId, reference.skillId); + return withReadableVersion(reference, consume); + } + async function withReadableVersion(reference: SkillContentReference, consume: (directory: string) => Promise): Promise { const { companyId, skillId, versionId } = reference; - await skillAccessService(db, options.actor).assertSkill(companyId, skillId); return withSkillVersionCacheLock(db, companyId, async () => { const version = await packages.version(companyId, skillId, versionId); if (!version) throw notFound("Skill version not found"); @@ -48,5 +51,5 @@ export function skillExecutionContent(db: Db, options: { actor?: AuthorizationAc return consume(directory); }); } - return { withVersion }; + return { withVersion, withReadableVersion }; }