feat(coding-agent): deliver the semantic-edge ledger through the agent-traces outbox

packages/coding-agent/.changes/acp-semantic-edges-delivery.md

@@ -0,0 +1 @@
+- Registered the per-session semantic-edge ledger with the agent-traces outbox as its own kind-tagged entry: durable upload intent at persist, an append-only byte cursor that never re-counts unchanged ledgers, startup catch-up counting, and pruning when a ledger is deleted with its session. No delivery endpoint exists yet, so pending ledgers are counted but never sent.

packages/coding-agent/src/core/agent-session-services.ts

@@ -16,6 +16,7 @@ import { ModelRegistry } from "./model-registry.js";
 import { DefaultResourceLoader, type DefaultResourceLoaderOptions, type ResourceLoader } from "./resource-loader.js";
 import type { SubagentRuntimeHost } from "./rlm-runtime.js";
 import { type CreateAgentSessionResult, createAgentSession } from "./sdk.js";
+import { semanticEdgeLedgerPath } from "./semantic-edges.js";
 import type { SessionManager } from "./session-manager.js";
 import { SettingsManager } from "./settings-manager.js";
 import { installAgentTelemetry, isTelemetryEnabled } from "./telemetry.js";
@@ -225,6 +226,10 @@ export async function createAgentSessionFromServices(
 	installAgentTraceUpload(options.sessionManager, {
 		authStorage: options.services.authStorage,
 		settingsManager: options.services.settingsManager,
+		semanticEdgesLedgerPath: semanticEdgeLedgerPath({
+			rlmSessionDir: options.rlmSessionDir,
+			sessionArtifactDir: options.sessionManager.getSessionArtifactDir(),
+		}),
 	});
 	const result = await createAgentSession({
 		cwd: options.services.cwd,

packages/coding-agent/src/core/agent-session.ts

@@ -233,8 +233,8 @@ import {
 } from "./rlm-runtime.js";
 import {
 	modelRequestHeaders,
-	SEMANTIC_EDGES_LEDGER_FILENAME,
 	SemanticEdgeRecorder,
+	semanticEdgeLedgerPath,
 	wrapStreamFnWithSemanticEdges,
 } from "./semantic-edges.js";
 import {
@@ -1284,9 +1284,11 @@ export class AgentSession {
 		this._rlmSessionDir = config.rlmSessionDir;
 		this._rlmParentNodeId = config.rlmParentNodeId;
 		this._rlmParentAgent = config.rlmParentAgent;
-		const semanticEdgesDir = this._rlmSessionDir ?? this.sessionManager.getSessionArtifactDir();
 		this._semanticEdges = new SemanticEdgeRecorder({
-			ledgerPath: semanticEdgesDir ? join(semanticEdgesDir, SEMANTIC_EDGES_LEDGER_FILENAME) : undefined,
+			ledgerPath: semanticEdgeLedgerPath({
+				rlmSessionDir: this._rlmSessionDir,
+				sessionArtifactDir: this.sessionManager.getSessionArtifactDir(),
+			}),
 			sessionId: this.sessionManager.getSessionId(),
 			parentSessionId: config.semanticParentSessionId,
 			spawnedByRequestId: config.semanticSpawnedByRequestId,

packages/coding-agent/src/core/agent-traces.ts

@@ -82,6 +82,8 @@ export interface AgentTraceUploadInstallOptions {
 	configPath?: string;
 	fetchFn?: typeof fetch;
 	requestTimeoutMs?: number;
+	/** The session's semantic-edge ledger; registered with the outbox as its own delivery kind. */
+	semanticEdgesLedgerPath?: string;
 }
 
 export type AgentTracePreviewResult =
@@ -639,8 +641,12 @@ interface AgentTraceUploadedSignature {
 	mtimeMs: number;
 }
 
+export const SEMANTIC_EDGES_OUTBOX_KIND = "semantic-edges";
+
 export interface AgentTraceCatchUpResult {
 	pruned: number;
+	/** Registered semantic-edge ledgers with bytes beyond their cursor; no delivery endpoint exists yet. */
+	semanticEdgeLedgersPending: number;
 	results: Array<{ sessionFile: string; result: AgentTraceUploadResult }>;
 }
 
@@ -654,9 +660,14 @@ function agentTraceOutboxEntryPath(sessionFile: string): string {
 	return join(getAgentTraceOutboxDir(), `${key}.json`);
 }
 
-function parseOutboxEntry(
-	raw: string,
-): { sessionFile: string; uploaded: AgentTraceUploadedSignature | null } | undefined {
+function parseOutboxEntry(raw: string):
+	| {
+			sessionFile: string;
+			kind?: string;
+			uploaded: AgentTraceUploadedSignature | null;
+			uploadedBytes?: number;
+	  }
+	| undefined {
 	const parsed = parseResponseObject(raw);
 	if (!parsed || typeof parsed.sessionFile !== "string") {
 		return undefined;
@@ -665,7 +676,12 @@ function parseOutboxEntry(
 		typeof parsed.size === "number" && typeof parsed.mtimeMs === "number"
 			? { size: parsed.size, mtimeMs: parsed.mtimeMs }
 			: null;
-	return { sessionFile: parsed.sessionFile, uploaded };
+	return {
+		sessionFile: parsed.sessionFile,
+		kind: typeof parsed.kind === "string" ? parsed.kind : undefined,
+		uploaded,
+		uploadedBytes: typeof parsed.uploadedBytes === "number" ? parsed.uploadedBytes : undefined,
+	};
 }
 
 /** `undefined` = no usable cursor; `null` = scheduled but never uploaded. */
@@ -688,15 +704,19 @@ function signatureEquals(a: AgentTraceUploadedSignature | null | undefined, b: A
 const locallyManagedSessionFiles = new Set<string>();
 
 /** Best-effort and synchronous: upload intent must be on disk the moment the transcript persist returns. */
-function markAgentTraceOutboxPendingSync(sessionFile: string): boolean {
+function markAgentTraceOutboxPendingSync(sessionFile: string, kind?: string): boolean {
 	try {
 		const entryPath = agentTraceOutboxEntryPath(sessionFile);
 		if (existsSync(entryPath)) {
 			return true;
 		}
 		mkdirSync(getAgentTraceOutboxDir(), { recursive: true });
 		const tempPath = `${entryPath}.${process.pid}.${randomUUID()}.tmp`;
-		writeFileSync(tempPath, `${JSON.stringify({ sessionFile })}\n`, "utf8");
+		writeFileSync(
+			tempPath,
+			`${JSON.stringify(kind === undefined ? { sessionFile } : { sessionFile, kind })}\n`,
+			"utf8",
+		);
 		renameSync(tempPath, entryPath);
 		return true;
 	} catch {
@@ -724,7 +744,7 @@ async function recordAgentTraceOutboxUpload(
 export async function catchUpAgentTraceUploads(
 	options: Omit<AgentTraceUploadOptions, "sessionFile">,
 ): Promise<AgentTraceCatchUpResult> {
-	const catchUp: AgentTraceCatchUpResult = { pruned: 0, results: [] };
+	const catchUp: AgentTraceCatchUpResult = { pruned: 0, semanticEdgeLedgersPending: 0, results: [] };
 	if (options.requireEnabled !== false && !(await getAgentTracesEnabled(options))) {
 		return catchUp;
 	}
@@ -756,6 +776,36 @@ export async function catchUpAgentTraceUploads(
 			catchUp.pruned += 1;
 			continue;
 		}
+		if (entry.kind === SEMANTIC_EDGES_OUTBOX_KIND) {
+			let ledgerStats: Awaited<ReturnType<typeof stat>>;
+			try {
+				ledgerStats = await stat(entry.sessionFile);
+			} catch (error) {
+				if ((error as NodeJS.ErrnoException).code === "ENOENT") {
+					await unlink(entryPath).catch(() => undefined);
+					catchUp.pruned += 1;
+				}
+				continue;
+			}
+			if (!ledgerStats.isFile()) {
+				await unlink(entryPath).catch(() => undefined);
+				catchUp.pruned += 1;
+				continue;
+			}
+			// Append-only byte cursor: a ledger whose size equals its delivered offset has nothing new.
+			if (ledgerStats.size === entry.uploadedBytes) {
+				continue;
+			}
+			// No delivery endpoint exists yet (verifiers#2449 consumes edges in-band over ACP
+			// metadata; the trace server has no semantic-edges route). The delta and cursor stay
+			// untouched so the first real sender delivers the whole backlog.
+			catchUp.semanticEdgeLedgersPending += 1;
+			continue;
+		}
+		if (entry.kind !== undefined) {
+			// A newer build may register kinds this one cannot deliver; leave their cursors alone.
+			continue;
+		}
 		if (locallyManagedSessionFiles.has(entry.sessionFile)) {
 			continue;
 		}
@@ -1018,13 +1068,17 @@ class AgentTraceUploadController {
 	private inFlight: Promise<void> | undefined;
 	private lastUploadStartedAt: number | undefined;
 	private notBeforeAt = 0;
+	private ledgerIntentMarked = false;
 
 	constructor(
 		private readonly sessionManager: SessionManager,
 		private options: AgentTraceUploadInstallOptions,
 	) {}
 
 	update(options: AgentTraceUploadInstallOptions): void {
+		if (options.semanticEdgesLedgerPath !== this.options.semanticEdgesLedgerPath) {
+			this.ledgerIntentMarked = false;
+		}
 		this.options = options;
 	}
 
@@ -1034,6 +1088,10 @@ class AgentTraceUploadController {
 		if (sessionFile && !locallyManagedSessionFiles.has(sessionFile) && markAgentTraceOutboxPendingSync(sessionFile)) {
 			locallyManagedSessionFiles.add(sessionFile);
 		}
+		const ledgerPath = this.options.semanticEdgesLedgerPath;
+		if (ledgerPath && !this.ledgerIntentMarked) {
+			this.ledgerIntentMarked = markAgentTraceOutboxPendingSync(ledgerPath, SEMANTIC_EDGES_OUTBOX_KIND);
+		}
 		this.arm();
 	};
 

packages/coding-agent/src/core/semantic-edges.ts

@@ -1,6 +1,6 @@
 import { createHash, randomUUID } from "node:crypto";
 import { appendFileSync, existsSync, mkdirSync, readFileSync, truncateSync } from "node:fs";
-import { dirname } from "node:path";
+import { dirname, join } from "node:path";
 import type { StreamFn } from "@earendil-works/pi-agent-core";
 
 /**
@@ -94,6 +94,15 @@ export interface SemanticEdge {
 	type: SemanticEdgeType;
 }
 
+/** The one derivation of where a session's ledger lives; recorder and outbox must agree. */
+export function semanticEdgeLedgerPath(options: {
+	rlmSessionDir?: string;
+	sessionArtifactDir?: string;
+}): string | undefined {
+	const dir = options.rlmSessionDir ?? options.sessionArtifactDir;
+	return dir ? join(dir, SEMANTIC_EDGES_LEDGER_FILENAME) : undefined;
+}
+
 export function modelRequestHeaders(requestId: string): Record<string, string> {
 	return {
 		[MODEL_REQUEST_ID_HEADER]: requestId,

packages/coding-agent/test/agent-traces.test.ts

@@ -90,17 +90,31 @@ function writeOutboxEntry(agentDir: string, sessionFile: string, signature?: { s
 function readOutboxEntry(
 	agentDir: string,
 	sessionFile: string,
-): { sessionFile: string; size?: number; mtimeMs?: number } | undefined {
+): { sessionFile: string; kind?: string; size?: number; mtimeMs?: number; uploadedBytes?: number } | undefined {
 	if (!existsSync(outboxEntryPath(agentDir, sessionFile))) {
 		return undefined;
 	}
 	return JSON.parse(readFileSync(outboxEntryPath(agentDir, sessionFile), "utf8")) as {
 		sessionFile: string;
+		kind?: string;
 		size?: number;
 		mtimeMs?: number;
+		uploadedBytes?: number;
 	};
 }
 
+function writeLedgerOutboxEntry(agentDir: string, ledgerFile: string, uploadedBytes?: number): void {
+	mkdirSync(join(agentDir, "agent-traces-outbox"), { recursive: true });
+	writeFileSync(
+		outboxEntryPath(agentDir, ledgerFile),
+		JSON.stringify({
+			sessionFile: ledgerFile,
+			kind: "semantic-edges",
+			...(uploadedBytes === undefined ? {} : { uploadedBytes }),
+		}),
+	);
+}
+
 async function advanceTimersUntil(condition: () => boolean): Promise<void> {
 	for (let step = 0; step < 200 && !condition(); step += 1) {
 		await stat(new URL(import.meta.url));
@@ -1202,12 +1216,99 @@ describe("agent trace upload", () => {
 			reloadConfig: false,
 		});
 
-		expect(result).toEqual({ pruned: 1, results: [] });
+		expect(result).toEqual({ pruned: 1, semanticEdgeLedgersPending: 0, results: [] });
 		expect(calls).toHaveLength(0);
 		expect(existsSync(outboxEntryPath(tempDir, deletedFile))).toBe(false);
 		expect(readOutboxEntry(tempDir, keptFile)).toEqual({ sessionFile: keptFile, ...keptSignature });
 	});
 
+	it("registers the semantic-edge ledger with a kind-tagged durable intent at persist", async () => {
+		vi.useFakeTimers();
+		const cwd = join(tempDir, "project");
+		const sessionDir = join(tempDir, "sessions");
+		mkdirSync(cwd, { recursive: true });
+		const sessionManager = SessionManager.create(cwd, sessionDir);
+		sessionManager.newSession({ id: "ledger-intent-session" });
+		const ledgerPath = join(tempDir, "artifacts", "semantic-edges.jsonl");
+
+		const calls: FetchCall[] = [];
+		installAgentTraceUpload(sessionManager, {
+			authStorage: AuthStorage.inMemory({
+				[PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" },
+			}),
+			settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }),
+			baseUrl: "https://api.example.test",
+			fetchFn: createFetchRecorder(calls),
+			semanticEdgesLedgerPath: ledgerPath,
+		});
+		sessionManager.appendMessage(createUserMessage("hello"));
+		sessionManager.appendMessage(createAssistantMessage("hi"));
+
+		// Synchronously durable, tagged with its own delivery kind, before any wire call.
+		expect(readOutboxEntry(tempDir, ledgerPath)).toEqual({ sessionFile: ledgerPath, kind: "semantic-edges" });
+		expect(calls).toHaveLength(0);
+
+		// A re-install with a DIFFERENT ledger path registers the new ledger at the next persist.
+		const movedLedgerPath = join(tempDir, "artifacts-moved", "semantic-edges.jsonl");
+		installAgentTraceUpload(sessionManager, {
+			authStorage: AuthStorage.inMemory({
+				[PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" },
+			}),
+			settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }),
+			baseUrl: "https://api.example.test",
+			fetchFn: createFetchRecorder(calls),
+			semanticEdgesLedgerPath: movedLedgerPath,
+		});
+		sessionManager.appendMessage(createAssistantMessage("after re-install"));
+		expect(readOutboxEntry(tempDir, movedLedgerPath)).toEqual({
+			sessionFile: movedLedgerPath,
+			kind: "semantic-edges",
+		});
+		expect(calls).toHaveLength(0);
+	});
+
+	it("counts appended ledger bytes as pending, stays quiet at the cursor, and prunes deleted ledgers", async () => {
+		const sessionDir = join(tempDir, "sessions");
+		mkdirSync(sessionDir, { recursive: true });
+		const ledgerFile = join(sessionDir, "semantic-edges.jsonl");
+		writeFileSync(ledgerFile, `${JSON.stringify({ type: "session_registered", session_id: "s" })}\n`);
+		writeLedgerOutboxEntry(tempDir, ledgerFile);
+		const deletedLedger = join(sessionDir, "gone", "semantic-edges.jsonl");
+		writeLedgerOutboxEntry(tempDir, deletedLedger);
+
+		const calls: FetchCall[] = [];
+		const options = {
+			authStorage: AuthStorage.inMemory({
+				[PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" },
+			}),
+			settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }),
+			baseUrl: "https://api.example.test",
+			fetchFn: createFetchRecorder(calls),
+			reloadConfig: false,
+		};
+
+		// Catch-up after a kill: the never-delivered ledger is pending; the deleted one is pruned.
+		const first = await catchUpAgentTraceUploads(options);
+		expect(first).toEqual({ pruned: 1, semanticEdgeLedgersPending: 1, results: [] });
+		expect(existsSync(outboxEntryPath(tempDir, deletedLedger))).toBe(false);
+		// The cursor stays untouched: the first real sender must deliver the whole backlog.
+		expect(readOutboxEntry(tempDir, ledgerFile)).toEqual({ sessionFile: ledgerFile, kind: "semantic-edges" });
+
+		// A cursor at the file size means unchanged: never re-counted, never resent.
+		const { size } = await stat(ledgerFile);
+		writeLedgerOutboxEntry(tempDir, ledgerFile, size);
+		const second = await catchUpAgentTraceUploads(options);
+		expect(second).toEqual({ pruned: 0, semanticEdgeLedgersPending: 0, results: [] });
+
+		// Appended bytes beyond the cursor become pending again.
+		writeFileSync(ledgerFile, `${JSON.stringify({ type: "request_started", request_id: "r", session_id: "s" })}\n`, {
+			flag: "a",
+		});
+		const third = await catchUpAgentTraceUploads(options);
+		expect(third).toEqual({ pruned: 0, semanticEdgeLedgersPending: 1, results: [] });
+		expect(calls).toHaveLength(0);
+	});
+
 	it("arms upload timers that never hold the process open", async () => {
 		const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
 		const cwd = join(tempDir, "project");
@@ -1381,7 +1482,7 @@ describe("agent trace upload", () => {
 		};
 
 		const result = await catchUpAgentTraceUploads(options);
-		expect(result).toEqual({ pruned: 1, results: [] });
+		expect(result).toEqual({ pruned: 1, semanticEdgeLedgersPending: 0, results: [] });
 		expect(calls).toHaveLength(0);
 		expect(existsSync(join(tempDir, "agent-traces-outbox", "deadbeef.json"))).toBe(false);
 		expect(await uploadAgentTraceFile({ ...options, sessionFile: keptFile })).toEqual({ status: "unchanged" });