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>

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<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 () => {
     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" });

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<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">) {

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");

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<void>;
     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();

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 };

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,

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,
         });

server/src/services/run-evidence-migration.ts

@@ -160,7 +160,6 @@ export async function migrateRunEvidence(
       let writer: Awaited<ReturnType<typeof journal.claim>> | 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)

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<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;
-    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 };
 }