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;
 		}
@@ -1030,9 +1080,23 @@ class AgentTraceUploadController {
 
 	schedule = (): void => {
 		this.pending = true;
-		const sessionFile = this.sessionManager.getSessionFile();
-		if (sessionFile && !locallyManagedSessionFiles.has(sessionFile) && markAgentTraceOutboxPendingSync(sessionFile)) {
-			locallyManagedSessionFiles.add(sessionFile);
+		// Intent is consent-gated at persist time: an entry created while sharing
+		// is off would turn a later enable into retroactive collection of
+		// opted-out sessions. Marking re-runs every persist (existsSync-cheap),
+		// so an entry pruned by a racing catch-up is re-registered.
+		if (this.options.settingsManager.getAgentTracesEnabled()) {
+			const sessionFile = this.sessionManager.getSessionFile();
+			if (
+				sessionFile &&
+				!locallyManagedSessionFiles.has(sessionFile) &&
+				markAgentTraceOutboxPendingSync(sessionFile)
+			) {
+				locallyManagedSessionFiles.add(sessionFile);
+			}
+			const ledgerPath = this.options.semanticEdgesLedgerPath;
+			if (ledgerPath) {
+				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,