Skip to content

Commit 6db955d

Browse files
TheGreatAxiosTheGreatAxios
committed
Gate primary evidence archive to sessions that request it
Workers omit the holder, so assemble keeps plain storage and skips admission and authorize recording wraps.
1 parent 424df45 commit 6db955d

3 files changed

Lines changed: 187 additions & 70 deletions

File tree

‎src/session/assemble-runtime.test.ts‎

Lines changed: 102 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,43 @@ function stubInferenceDeps(): ChatAgentWiring["inferenceDeps"] {
108108
};
109109
}
110110

111+
function stubAuthorize(): ChatAgentWiring["authorize"] {
112+
return async () => ({ effect: "allow", matchingGrants: [], resolvedBy: null });
113+
}
114+
115+
function stubChatAgentWiring(overrides: Partial<ChatAgentWiring> = {}): ChatAgentWiring {
116+
return {
117+
toolsId: "test/tools",
118+
agentId: "test/agent",
119+
systemPrompt: "prompt",
120+
authorize: stubAuthorize(),
121+
getDynamicRunner: () => {
122+
throw new Error("getDynamicRunner should not run at assemble or mocked build");
123+
},
124+
computeAdvertised: () => [],
125+
activateTools: () => false,
126+
inactivityTimeoutMs: 1_000,
127+
onTasksChange: () => {},
128+
requestContinuation: () => {},
129+
getProvider: () => ({ providerName: "test", model: "m" }),
130+
getWorkdir: () => "/build-dir",
131+
inferenceDeps: stubInferenceDeps(),
132+
getSources: () => [
133+
{
134+
id: "s",
135+
provider: "test",
136+
baseURL: "http://localhost",
137+
apiKey: "k",
138+
model: "m",
139+
},
140+
],
141+
getDefaultSource: () => "s",
142+
getCompactor: () => stubCompactor("build"),
143+
onBuilt: () => {},
144+
...overrides,
145+
};
146+
}
147+
111148
describe("assembleChatAgent", () => {
112149
test("getWorkdir and getCompactor run at buildAgent time, not assemble time", async () => {
113150
const storeDirs: string[] = [];
@@ -148,41 +185,18 @@ describe("assembleChatAgent", () => {
148185
let liveDir = "/assemble-dir";
149186
let liveCompactor = stubCompactor("assemble");
150187

151-
const { buildAgent } = assembleChatAgent({
152-
toolsId: "test/tools",
153-
agentId: "test/agent",
154-
systemPrompt: "prompt",
155-
authorize: async () => ({ effect: "allow", matchingGrants: [], resolvedBy: null }),
156-
getDynamicRunner: () => {
157-
throw new Error("getDynamicRunner should not run at assemble or mocked build");
158-
},
159-
computeAdvertised: () => [],
160-
activateTools: () => false,
161-
inactivityTimeoutMs: 1_000,
162-
onTasksChange: () => {},
163-
requestContinuation: () => {},
164-
getProvider: () => ({ providerName: "test", model: "m" }),
165-
getWorkdir: () => {
166-
workdirCalls.push(liveDir);
167-
return liveDir;
168-
},
169-
inferenceDeps: stubInferenceDeps(),
170-
getSources: () => [
171-
{
172-
id: "s",
173-
provider: "test",
174-
baseURL: "http://localhost",
175-
apiKey: "k",
176-
model: "m",
188+
const { buildAgent } = assembleChatAgent(
189+
stubChatAgentWiring({
190+
getWorkdir: () => {
191+
workdirCalls.push(liveDir);
192+
return liveDir;
177193
},
178-
],
179-
getDefaultSource: () => "s",
180-
getCompactor: () => {
181-
compactorCalls.push(liveCompactor.name);
182-
return liveCompactor;
183-
},
184-
onBuilt: () => {},
185-
});
194+
getCompactor: () => {
195+
compactorCalls.push(liveCompactor.name);
196+
return liveCompactor;
197+
},
198+
}),
199+
);
186200

187201
expect(workdirCalls).toEqual([]);
188202
expect(compactorCalls).toEqual([]);
@@ -205,4 +219,58 @@ describe("assembleChatAgent", () => {
205219
},
206220
);
207221
});
222+
223+
test("omits evidence archive when no holder is provided", async () => {
224+
const fakeStorage = {
225+
readBlob: async () => new Uint8Array(),
226+
} as unknown as ContextStore;
227+
const fakeAgent = { close: async () => {} } as unknown as Agent;
228+
const authorize = stubAuthorize();
229+
let capturedStorage: ContextStore | undefined;
230+
let capturedAuthorize: unknown;
231+
let builtAgent: Agent | undefined;
232+
let builtStorage: ContextStore | undefined;
233+
234+
await withMockedModuleDuring(
235+
import.meta.resolve("./optimized-context-store.js"),
236+
(real: typeof import("./optimized-context-store.js")) => ({
237+
...real,
238+
createOptimizedContextStore: async () => fakeStorage,
239+
}),
240+
async () => {
241+
await withMockedModuleDuring(
242+
import.meta.resolve("../agent/live-tool-dispatch.js"),
243+
(real: typeof import("../agent/live-tool-dispatch.js")) => ({
244+
...real,
245+
createAgentWithLiveToolDispatch: async (
246+
_def: unknown,
247+
env: { storage: ContextStore; authorize: unknown },
248+
) => {
249+
capturedStorage = env.storage;
250+
capturedAuthorize = env.authorize;
251+
return fakeAgent;
252+
},
253+
}),
254+
async () => {
255+
const { assembleChatAgent } = await import("./assemble-runtime.js");
256+
const { buildAgent } = assembleChatAgent(
257+
stubChatAgentWiring({
258+
authorize,
259+
onBuilt: (agent, storage) => {
260+
builtAgent = agent;
261+
builtStorage = storage;
262+
},
263+
}),
264+
);
265+
await buildAgent();
266+
},
267+
);
268+
},
269+
);
270+
271+
expect(capturedStorage).toBe(fakeStorage);
272+
expect(capturedAuthorize).toBe(authorize);
273+
expect(builtAgent).toBe(fakeAgent);
274+
expect(builtStorage).toBe(fakeStorage);
275+
});
208276
});

‎src/session/assemble-runtime.ts‎

Lines changed: 46 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -433,59 +433,68 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent {
433433
const buildAgent = async (): Promise<Agent> => {
434434
const workdir = wiring.getWorkdir();
435435
const storage = await createOptimizedContextStore(workdir);
436-
// workdir is …/<sessionId>/context for primary sessions.
437-
const sessionId = path.basename(path.dirname(workdir));
438-
const archive = createCompactionArchive({
439-
sessionId,
440-
contextDir: workdir,
441-
writeBlob: (key, bytes, contentType) => storage.writeBlob(key, bytes, contentType),
442-
readBlob: (key) => storage.readBlob(key),
443-
});
444-
if (wiring.evidenceArchiveHolder !== undefined) {
445-
wiring.evidenceArchiveHolder.current = archive;
436+
// Primary-only evidence archive. Workers never pass evidenceArchiveHolder, so
437+
// they keep plain storage and omit admission / authorize recording wraps.
438+
const archiveHolder = wiring.evidenceArchiveHolder;
439+
let primaryArchive: CompactionArchive | undefined;
440+
if (archiveHolder !== undefined) {
441+
const sessionId = path.basename(path.dirname(workdir));
442+
primaryArchive = createCompactionArchive({
443+
sessionId,
444+
contextDir: workdir,
445+
writeBlob: (key, bytes, contentType) => storage.writeBlob(key, bytes, contentType),
446+
readBlob: (key) => storage.readBlob(key),
447+
});
448+
archiveHolder.current = primaryArchive;
446449
}
447450

448-
const storageWithArchive: ContextStore = {
449-
...storage,
450-
async writeResponse(turn, signal) {
451-
const content = turn.content.map((block) => {
452-
if (block.type !== "text") return block;
453-
const text = applyRecordingPolicyToText(block.text);
454-
return text === block.text ? block : { ...block, text };
455-
});
456-
const admitted = { ...turn, content };
457-
for (const block of admitted.content) {
458-
if (block.type === "text" && block.text.length > 0) {
459-
await archive.recordAuthorizedPayload({
460-
kind: "assistant_text",
461-
payload: block.text,
462-
provenance: "writeResponse:post-policy",
463-
});
464-
}
465-
}
466-
return storage.writeResponse(admitted, signal);
467-
},
468-
};
451+
const storageForAgent: ContextStore =
452+
primaryArchive === undefined
453+
? storage
454+
: {
455+
...storage,
456+
async writeResponse(turn, signal) {
457+
const content = turn.content.map((block) => {
458+
if (block.type !== "text") return block;
459+
const text = applyRecordingPolicyToText(block.text);
460+
return text === block.text ? block : { ...block, text };
461+
});
462+
const admitted = { ...turn, content };
463+
for (const block of admitted.content) {
464+
if (block.type === "text" && block.text.length > 0) {
465+
await primaryArchive.recordAuthorizedPayload({
466+
kind: "assistant_text",
467+
payload: block.text,
468+
provenance: "writeResponse:post-policy",
469+
});
470+
}
471+
}
472+
return storage.writeResponse(admitted, signal);
473+
},
474+
};
469475

470476
const agent = await createAgentWithLiveToolDispatch(agentDef, {
471477
sources: wiring.getSources(),
472478
defaultSource: wiring.getDefaultSource(),
473-
storage: storageWithArchive,
479+
storage: storageForAgent,
474480
workdir,
475481
// contextTransforms ride deps: the published @intx/agent forwards deps
476482
// into reactor assembly verbatim, and the vendored assembly picks the
477483
// transforms up from there.
478484
deps: {
479485
...wiring.inferenceDeps,
480486
contextTransforms: [
481-
createAttachmentRehydrateTransform((key) => storageWithArchive.readBlob(key)),
487+
createAttachmentRehydrateTransform((key) => storageForAgent.readBlob(key)),
482488
],
483489
},
484490
audit: noopAuditStore(),
485491
// Gate-backed reactor authorization: ask-tier calls suspend via the
486492
// vendored approval-suspend primitive instead of parking on a closure.
487493
// Finalize evidence admission after guards resolve; never scrub exec args.
488-
authorize: wrapAuthorizeWithEvidenceArchive(wiring.authorize, () => archive),
494+
authorize:
495+
primaryArchive === undefined
496+
? wiring.authorize
497+
: wrapAuthorizeWithEvidenceArchive(wiring.authorize, () => primaryArchive),
489498
directors: createDirectorRegistry({
490499
factories: [chatDirectorDef.factory],
491500
defaultId: `${ID_PREFIX}/chat`,
@@ -494,8 +503,9 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent {
494503
"pruning-compactor": wiring.getCompactor(),
495504
},
496505
});
497-
const admittedAgent = createPrimaryDeliveryAdmission(agent, archive);
498-
wiring.onBuilt(admittedAgent, storageWithArchive);
506+
const admittedAgent =
507+
primaryArchive === undefined ? agent : createPrimaryDeliveryAdmission(agent, primaryArchive);
508+
wiring.onBuilt(admittedAgent, storageForAgent);
499509
return admittedAgent;
500510
};
501511

‎src/session/compaction-archive.test.ts‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,45 @@ describe("primary message admission", () => {
134134
expect(delivered[0]!.content).toContain(CREDENTIAL_REDACTION);
135135
expect(delivered[1]!.content).toContain(CREDENTIAL_REDACTION);
136136
});
137+
138+
test("history and archive share the same admitted representation", async () => {
139+
const dir = tempDir();
140+
const blobs = new Map<string, Uint8Array>();
141+
const archive = createCompactionArchive({
142+
sessionId: "sess-admit",
143+
contextDir: dir,
144+
writeBlob: async (key, bytes) => {
145+
blobs.set(key, bytes);
146+
},
147+
readBlob: async (key) => {
148+
const bytes = blobs.get(key);
149+
if (bytes === undefined) throw new Error(`missing blob ${key}`);
150+
return bytes;
151+
},
152+
});
153+
const delivered: InboundMessage[] = [];
154+
const agent = {
155+
deliver(message: InboundMessage) {
156+
delivered.push(message);
157+
},
158+
async send(message: InboundMessage) {
159+
delivered.push(message);
160+
return { ok: true as const };
161+
},
162+
};
163+
const wrapped = createPrimaryDeliveryAdmission(agent, archive);
164+
wrapped.deliver(inbound({ content: "constraint sk-abcdefghijklmnopqrstuvwxyz012345" }));
165+
await archive.awaitPendingWrites();
166+
expect(delivered).toHaveLength(1);
167+
const admitted = delivered[0]!.content!;
168+
expect(admitted).toContain(CREDENTIAL_REDACTION);
169+
expect(admitted).not.toContain("sk-abcdefghijklmnopqrstuvwxyz012345");
170+
const occurrences = await archive.listOccurrences();
171+
expect(occurrences).toHaveLength(1);
172+
expect(occurrences[0]!.kind).toBe("user_message");
173+
const archived = await archive.readAuthorizedPayload(occurrences[0]!.occurrenceId);
174+
expect(archived).toBe(admitted);
175+
});
137176
});
138177

139178
describe("compaction archive storage", () => {

0 commit comments

Comments
 (0)