mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
perf: remove repeated reads and checks from run preparation (#20358)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Before each run starts, Core prepares the run: it opens the run log, resolves skills, and prepares the AI runtime > - Every run pays this preparation time before the agent can begin, for a new task or a follow-up, on a cold or a warm computer > - A trace on a hosted test deployment showed some preparation steps that repeat a read or a check that a later step does again > - This pull request removes three of these repeated steps and keeps the check that each one duplicated > - The benefit is that each run starts sooner, with no new mechanism and no change to what a run can see or write ## Linked Issues or Issue Description **What existing behavior does this improve?** Run preparation (`task.prepare`). On a hosted test deployment it takes about 2.8 to 3.6 s. Skills preparation has a P50 of 275 ms (1,161 ms on a deployment with more skills and projects). `heartbeat.prepare.run_output` takes 386 to 724 ms. **Subsystem affected** Run logs in object storage (`object-byte-journal`, `object-run-log-store`) and runtime skill delivery (`company-skills`, `skill-execution-content`). **Current behavior** - An object journal claim is preceded by a head read. Its only use is to give the current owner to `claim()`. `claim()` then reads the head again, and it rejects the claim only when the two reads differ. The run log, provider trace capture, and run evidence migration all do this. - Run log `begin` reads the source-prefix proof object for every claim. For a new run log, no proof can exist, so the read always returns "missing". - Each delivered skill version calls `withVersion`, which authorizes the skill again for the run's actor. The skill row was read with the same actor's visibility condition in the same call. With an agent actor this repeats the company source and project permission reads once per version. **Proposed behavior** - `claim()` reads the head once. It still refuses a deleted journal. The conditional head write still rejects any change after that read, and a stale writer is still fenced. - `begin` does not read a proof for a first claim (generation 1). A resumed log still reads and validates its proof, so a changed or malformed proof still fails closed. - Runtime skill delivery authorizes each skill once, when it reads the selected rows. Other callers of `withVersion` keep their own check. **Reason and benefit** Each removed step protects nothing that a later step does not protect again: - Head read: the claim builds on its own head read, and the conditional write enforces it. - Proof read: a proof is written only by a writer that holds this log's head. The journal never removes a head (deletion leaves a tombstone that `claim()` refuses). So a first claim cannot find a proof. - Skill check: every caller of `materializeVersionSnapshot` passes a row that it read through the same actor's visibility condition (`readFullSkillRows`, `getById`, `listFull`). Expected saving per run: two object-storage GETs on `run_output` (about 100 ms each on the trace), and the per-version visibility reads in skills preparation (4 versions per run on the test deployment, about 388 ms per version on a heavier deployment). **Breaking changes** None. `claim()` no longer takes `expectedOwner`. It is an internal API with three callers in this repository, all updated. Refs #20284 (it made run-log tail reads parallel and cached; this change covers the begin path). ## What Changed - `object-byte-journal`: remove the `expectedOwner` input and comparison from `claim()`. Remove the extra head read in its three callers (run log `begin`, provider trace capture, run evidence migration). - `object-run-log-store`: `begin` reads the source-prefix proof only when the claimed head has `generation > 1`. - `skill-execution-content`: split `withVersion` into the access check and `withReadableVersion`. `withVersion` keeps the check for its other callers. - `company-skills`: `materializeVersionSnapshot` uses `withReadableVersion`, because each of its callers already read the skill row under the actor's visibility condition. - Tests: - A new run log `begin` reads `head.json` once and reads no proof. A resumed log reads `head.json` once and reads its proof once. - Runtime skill delivery for an agent actor does no per-version source permission reads. ## Verification - `cd server && npx vitest run src/services/object-byte-journal.test.ts src/services/object-run-log-store.test.ts src/services/object-run-log-store-tail-read.test.ts src/services/run-log-store.test.ts src/services/provider-trace-object-store.test.ts` - `cd server && npx vitest run src/__tests__/company-skills-runtime-selection.test.ts src/__tests__/company-skills-runtime-concurrency.test.ts src/__tests__/heartbeat-runtime-skills.test.ts src/__tests__/company-skills-service.test.ts src/__tests__/managed-skill-content.test.ts` (embedded Postgres) - `cd server && pnpm run typecheck` - Each new test fails on the base commit: the run log test sees 2 head reads and 1 proof read; the skill test sees 1 source permission read per delivered version. - Existing tests still pass for: stale-writer fencing after a takeover, a deleted journal that refuses a claim, a malformed proof that fails closed, and a denied project-scoped skill that is not delivered. - I did not run the full `pnpm test:run`. CI covers it. ## Risks - Low. Two concurrent claims of the same journal: before, the second claim failed when the first published between its two head reads. Now the later claim takes over, and the earlier writer's next append fails with a conflict. This is the same result as when the later claim starts after the first publish. Writer authority (`assertAuthority`) is unchanged. - A skill whose visibility is revoked in the moment between its row read and its materialization in the same call is still delivered for that run. This was already true for a revocation just after the old per-version check. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use (shell, file edits, test runs). ## 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 - [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 - [ ] 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
380b3d49c0
commit
0e3fdebea8
9 files changed
+71
-34
No files matched your search
@@ -98,6 +98,35 @@ describeDatabase("selected runtime skill reconciliation", () => {
|
|||||||
expect(await fs.readFile(path.join(old,"SKILL.md"),"utf8")).toBe("Old installation content");
|
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<typeof db.select>) => {
|
||||||
|
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 () => {
|
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();
|
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" });
|
await db.insert(companyMemberships).values({ companyId, principalType: "agent", principalId: agent!.id, membershipRole: "member" });
|
||||||
|
|||||||
@@ -5981,7 +5981,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
|
|||||||
}
|
}
|
||||||
|
|
||||||
async function materializeVersionSnapshot(companyId: string, skill: CompanySkill, version: StoredSkillVersion): Promise<string> {
|
async function materializeVersionSnapshot(companyId: string, skill: CompanySkill, version: StoredSkillVersion): Promise<string> {
|
||||||
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<CompanySkill, "key" | "slug">) {
|
function resolveRuntimeSkillMaterializedPath(companyId: string, skill: Pick<CompanySkill, "key" | "slug">) {
|
||||||
|
|||||||
@@ -96,7 +96,6 @@ describe("object byte journal", () => {
|
|||||||
const controller = new AbortController();
|
const controller = new AbortController();
|
||||||
const a = createObjectByteJournal({ provider: s.provider, prefix: "company/a/log" });
|
const a = createObjectByteJournal({ provider: s.provider, prefix: "company/a/log" });
|
||||||
const first = a.claim({
|
const first = a.claim({
|
||||||
expectedOwner: null,
|
|
||||||
assertAuthority: authority,
|
assertAuthority: authority,
|
||||||
signal: controller.signal,
|
signal: controller.signal,
|
||||||
});
|
});
|
||||||
@@ -104,7 +103,6 @@ describe("object byte journal", () => {
|
|||||||
await probing;
|
await probing;
|
||||||
const b = createObjectByteJournal({ provider: s.provider, prefix: "company/b/log" });
|
const b = createObjectByteJournal({ provider: s.provider, prefix: "company/b/log" });
|
||||||
const second = b.claim({
|
const second = b.claim({
|
||||||
expectedOwner: null,
|
|
||||||
assertAuthority: async () => {
|
assertAuthority: async () => {
|
||||||
admitted();
|
admitted();
|
||||||
},
|
},
|
||||||
@@ -121,7 +119,7 @@ describe("object byte journal", () => {
|
|||||||
it("recovers committed bytes and stable ranges with no local filesystem", async () => {
|
it("recovers committed bytes and stable ranges with no local filesystem", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
a = createObjectByteJournal({ provider: s.provider, prefix: "company/run/log" });
|
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));
|
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(0, 17));
|
||||||
writer = await a.append(writer, expected.subarray(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 () => {
|
it("fences stale appends/finalization and preserves a pinned reader snapshot", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
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"));
|
old = await a.append(old, Buffer.from("before"));
|
||||||
const snapshot = old.head;
|
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"));
|
next = await a.append(next, Buffer.from("after"));
|
||||||
await expect(a.append(old, Buffer.from("lost"))).rejects.toBeInstanceOf(
|
await expect(a.append(old, Buffer.from("lost"))).rejects.toBeInstanceOf(
|
||||||
ObjectJournalConflictError,
|
ObjectJournalConflictError,
|
||||||
@@ -149,7 +147,7 @@ describe("object byte journal", () => {
|
|||||||
it("reconciles an uncertain manifest commit without duplicating output", async () => {
|
it("reconciles an uncertain manifest commit without duplicating output", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
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();
|
s.loseResponse();
|
||||||
writer = await a.append(writer, Buffer.from("once"));
|
writer = await a.append(writer, Buffer.from("once"));
|
||||||
expect(writer.head.end).toBe(4);
|
expect(writer.head.end).toBe(4);
|
||||||
@@ -158,7 +156,7 @@ describe("object byte journal", () => {
|
|||||||
it("keeps index objects bounded over multiple tree levels", async () => {
|
it("keeps index objects bounded over multiple tree levels", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
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++)
|
for (let i = 0; i < 180; i++)
|
||||||
writer = await a.append(writer, Buffer.from(String(i).padStart(4, "0")));
|
writer = await a.append(writer, Buffer.from(String(i).padStart(4, "0")));
|
||||||
expect(await a.readRange(writer.head, 252, 16)).toEqual(Buffer.from("0063006400650066"));
|
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 () => {
|
it("retries a transient head conflict without fencing its current writer", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
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!;
|
const original = s.provider.putObjectConditional!;
|
||||||
let conflicts = 0;
|
let conflicts = 0;
|
||||||
s.provider.putObjectConditional = async (input) => {
|
s.provider.putObjectConditional = async (input) => {
|
||||||
@@ -205,14 +203,14 @@ describe("object byte journal", () => {
|
|||||||
};
|
};
|
||||||
const journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
const journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
||||||
await expect(
|
await expect(
|
||||||
journal.claim({ expectedOwner: null, assertAuthority: authority }),
|
journal.claim({ assertAuthority: authority }),
|
||||||
).rejects.toThrow("conflict");
|
).rejects.toThrow("conflict");
|
||||||
expect(s.objects.size).toBe(0);
|
expect(s.objects.size).toBe(0);
|
||||||
});
|
});
|
||||||
it("retries transient segment failures and acknowledges exactly one batch", async () => {
|
it("retries transient segment failures and acknowledges exactly one batch", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
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!;
|
const original = s.provider.putObjectConditional!;
|
||||||
let failures = 2;
|
let failures = 2;
|
||||||
s.provider.putObjectConditional = async (input) => {
|
s.provider.putObjectConditional = async (input) => {
|
||||||
@@ -226,7 +224,7 @@ describe("object byte journal", () => {
|
|||||||
it("rejects corrupt published data and cancelled reads", async () => {
|
it("rejects corrupt published data and cancelled reads", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
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"));
|
writer = await a.append(writer, Buffer.from("data"));
|
||||||
const data = [...s.objects].find(([, bytes]) => bytes.toString() === "data")!;
|
const data = [...s.objects].find(([, bytes]) => bytes.toString() === "data")!;
|
||||||
s.objects.set(data[0], Buffer.from("oops"));
|
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 () => {
|
it("sheds only interactive reads under saturated object IO and never a writer", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/busy" });
|
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"));
|
writer = await a.append(writer, Buffer.from("data"));
|
||||||
const get = s.provider.getObject;
|
const get = s.provider.getObject;
|
||||||
let open!: () => void;
|
let open!: () => void;
|
||||||
@@ -255,7 +253,7 @@ describe("object byte journal", () => {
|
|||||||
const reads = Array.from({ length: 40 }, () => reader.head());
|
const reads = Array.from({ length: 40 }, () => reader.head());
|
||||||
await expect(reader.head()).rejects.toBeInstanceOf(ObjectJournalBusyError);
|
await expect(reader.head()).rejects.toBeInstanceOf(ObjectJournalBusyError);
|
||||||
const other = createObjectByteJournal({ provider: s.provider, prefix: "c/r/writer" });
|
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();
|
open();
|
||||||
const otherWriter = await other.append(await claimed, Buffer.from("more"));
|
const otherWriter = await other.append(await claimed, Buffer.from("more"));
|
||||||
expect(otherWriter.head.end).toBe(4);
|
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 () => {
|
it("releases a queued read's place when its caller aborts", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/abort" });
|
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"));
|
writer = await a.append(writer, Buffer.from("data"));
|
||||||
const get = s.provider.getObject;
|
const get = s.provider.getObject;
|
||||||
let open!: () => void;
|
let open!: () => void;
|
||||||
@@ -292,14 +290,14 @@ describe("object byte journal", () => {
|
|||||||
it("retains a deletion tombstone and denies a subsequent claim", async () => {
|
it("retains a deletion tombstone and denies a subsequent claim", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
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"));
|
writer = await a.append(writer, Buffer.from("private"));
|
||||||
await a.close(writer);
|
await a.close(writer);
|
||||||
await a.remove();
|
await a.remove();
|
||||||
expect([...s.objects.keys()]).toEqual(["c/r/log/head.json"]);
|
expect([...s.objects.keys()]).toEqual(["c/r/log/head.json"]);
|
||||||
expect((await a.head())!.head.deleted).toBe(true);
|
expect((await a.head())!.head.deleted).toBe(true);
|
||||||
await expect(
|
await expect(
|
||||||
a.claim({ expectedOwner: writer.head.owner, assertAuthority: authority }),
|
a.claim({ assertAuthority: authority }),
|
||||||
).rejects.toThrow();
|
).rejects.toThrow();
|
||||||
});
|
});
|
||||||
it("does not claim ownership when execution authority rejects", async () => {
|
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" });
|
a = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
||||||
await expect(
|
await expect(
|
||||||
a.claim({
|
a.claim({
|
||||||
expectedOwner: null,
|
|
||||||
assertAuthority: async () => {
|
assertAuthority: async () => {
|
||||||
throw new Error("revoked");
|
throw new Error("revoked");
|
||||||
},
|
},
|
||||||
@@ -326,7 +323,6 @@ describe("object byte journal", () => {
|
|||||||
release = resolve;
|
release = resolve;
|
||||||
});
|
});
|
||||||
const pending = journal.claim({
|
const pending = journal.claim({
|
||||||
expectedOwner: null,
|
|
||||||
assertAuthority: async () => {
|
assertAuthority: async () => {
|
||||||
enter();
|
enter();
|
||||||
await blocked;
|
await blocked;
|
||||||
@@ -348,7 +344,7 @@ describe("object byte journal", () => {
|
|||||||
};
|
};
|
||||||
const journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
const journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
||||||
await expect(
|
await expect(
|
||||||
journal.claim({ expectedOwner: null, assertAuthority: authority }),
|
journal.claim({ assertAuthority: authority }),
|
||||||
).rejects.toThrow("does not enforce conditional");
|
).rejects.toThrow("does not enforce conditional");
|
||||||
expect(s.objects.size).toBe(0);
|
expect(s.objects.size).toBe(0);
|
||||||
});
|
});
|
||||||
@@ -356,7 +352,7 @@ describe("object byte journal", () => {
|
|||||||
vi.useFakeTimers({ toFake: ["Date"] });
|
vi.useFakeTimers({ toFake: ["Date"] });
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
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"));
|
writer = await journal.append(writer, Buffer.from("private"));
|
||||||
await expect(journal.remove()).rejects.toThrow("cleanup_pending");
|
await expect(journal.remove()).rejects.toThrow("cleanup_pending");
|
||||||
expect((await journal.head())!.head.deleted).toBe(true);
|
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 () => {
|
it("does not publish a partial batch when a later segment fails", async () => {
|
||||||
const s = storage(),
|
const s = storage(),
|
||||||
journal = createObjectByteJournal({ provider: s.provider, prefix: "c/r/log" });
|
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 original = s.provider.putObjectConditional!;
|
||||||
const failed = Buffer.alloc(1024 * 1024, 2);
|
const failed = Buffer.alloc(1024 * 1024, 2);
|
||||||
const key = createHash("sha256").update(failed).digest("hex");
|
const key = createHash("sha256").update(failed).digest("hex");
|
||||||
|
|||||||
@@ -531,13 +531,11 @@ export function createObjectByteJournal(options: ObjectByteJournalOptions) {
|
|||||||
}, signal);
|
}, signal);
|
||||||
}
|
}
|
||||||
async function claim(input: {
|
async function claim(input: {
|
||||||
expectedOwner: string | null;
|
|
||||||
assertAuthority: () => Promise<void>;
|
assertAuthority: () => Promise<void>;
|
||||||
signal?: AbortSignal;
|
signal?: AbortSignal;
|
||||||
}) {
|
}) {
|
||||||
const current = await head(input.signal);
|
const current = await head(input.signal);
|
||||||
if (current?.head.deleted || (current?.head.owner ?? null) !== input.expectedOwner)
|
if (current?.head.deleted) throw new ObjectJournalConflictError();
|
||||||
throw new ObjectJournalConflictError();
|
|
||||||
await input.assertAuthority();
|
await input.assertAuthority();
|
||||||
await qualifyConditionalWrites(provider, prefix, input.signal);
|
await qualifyConditionalWrites(provider, prefix, input.signal);
|
||||||
await input.assertAuthority();
|
await input.assertAuthority();
|
||||||
|
|||||||
@@ -118,6 +118,22 @@ describe("S3 run log store", () => {
|
|||||||
expect((await store.finalize(handle)).bytes).toBe(Buffer.byteLength(content));
|
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 () => {
|
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,
|
const m = memory(), options = { provider: m.provider, keyPrefix: "logs", flushMs: 1,
|
||||||
assertWriterAuthority: async () => undefined };
|
assertWriterAuthority: async () => undefined };
|
||||||
|
|||||||
@@ -254,14 +254,12 @@ export function createObjectRunLogStore(options: ObjectRunLogStoreOptions): RunL
|
|||||||
return {
|
return {
|
||||||
async begin(input) {
|
async begin(input) {
|
||||||
const logRef = `${input.companyId}/${input.agentId}/${input.runId}`;
|
const logRef = `${input.companyId}/${input.agentId}/${input.runId}`;
|
||||||
const store = journal(logRef),
|
const store = journal(logRef);
|
||||||
current = await store.head(input.signal);
|
|
||||||
const writer = await store.claim({
|
const writer = await store.claim({
|
||||||
expectedOwner: current?.head.owner ?? null,
|
|
||||||
assertAuthority: () => options.assertWriterAuthority(input),
|
assertAuthority: () => options.assertWriterAuthority(input),
|
||||||
signal: input.signal,
|
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 = {
|
const handle: RunLogHandle = {
|
||||||
store: "s3_segments",
|
store: "s3_segments",
|
||||||
logRef,
|
logRef,
|
||||||
|
|||||||
@@ -172,9 +172,7 @@ export function createProviderTraceObjectStore(
|
|||||||
try {
|
try {
|
||||||
for (const source of channelsFor(ref, filePath)) {
|
for (const source of channelsFor(ref, filePath)) {
|
||||||
capture.signal.throwIfAborted();
|
capture.signal.throwIfAborted();
|
||||||
const current = await source.journal.head(capture.signal);
|
|
||||||
const writer = await source.journal.claim({
|
const writer = await source.journal.claim({
|
||||||
expectedOwner: current?.head.owner ?? null,
|
|
||||||
assertAuthority,
|
assertAuthority,
|
||||||
signal: capture.signal,
|
signal: capture.signal,
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -160,7 +160,6 @@ export async function migrateRunEvidence(
|
|||||||
let writer: Awaited<ReturnType<typeof journal.claim>> | null = null;
|
let writer: Awaited<ReturnType<typeof journal.claim>> | null = null;
|
||||||
try { writer = options.apply
|
try { writer = options.apply
|
||||||
? await journal.claim({
|
? await journal.claim({
|
||||||
expectedOwner: (await journal.head(options.signal))?.head.owner ?? null,
|
|
||||||
signal: options.signal,
|
signal: options.signal,
|
||||||
assertAuthority: async () => {
|
assertAuthority: async () => {
|
||||||
if (scopedArchive)
|
if (scopedArchive)
|
||||||
|
|||||||
@@ -17,8 +17,11 @@ export function skillExecutionContent(db: Db, options: { actor?: AuthorizationAc
|
|||||||
const packages = skillPackageStore(options.packageDb ?? db);
|
const packages = skillPackageStore(options.packageDb ?? db);
|
||||||
const runtime = skillPackageRuntime(options.packageDb ?? db);
|
const runtime = skillPackageRuntime(options.packageDb ?? db);
|
||||||
async function withVersion<T>(reference: SkillContentReference, consume: (directory: string) => Promise<T>): Promise<T> {
|
async function withVersion<T>(reference: SkillContentReference, consume: (directory: string) => Promise<T>): Promise<T> {
|
||||||
|
await skillAccessService(db, options.actor).assertSkill(reference.companyId, reference.skillId);
|
||||||
|
return withReadableVersion(reference, consume);
|
||||||
|
}
|
||||||
|
async function withReadableVersion<T>(reference: SkillContentReference, consume: (directory: string) => Promise<T>): Promise<T> {
|
||||||
const { companyId, skillId, versionId } = reference;
|
const { companyId, skillId, versionId } = reference;
|
||||||
await skillAccessService(db, options.actor).assertSkill(companyId, skillId);
|
|
||||||
return withSkillVersionCacheLock(db, companyId, async () => {
|
return withSkillVersionCacheLock(db, companyId, async () => {
|
||||||
const version = await packages.version(companyId, skillId, versionId);
|
const version = await packages.version(companyId, skillId, versionId);
|
||||||
if (!version) throw notFound("Skill version not found");
|
if (!version) throw notFound("Skill version not found");
|
||||||
@@ -48,5 +51,5 @@ export function skillExecutionContent(db: Db, options: { actor?: AuthorizationAc
|
|||||||
return consume(directory);
|
return consume(directory);
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
return { withVersion };
|
return { withVersion, withReadableVersion };
|
||||||
}
|
}
|
||||||
Reference in new issue
Block a user