feat(server): forward per-prompt actor receipts

.changeset/quiet-hounds-attribute.md

@@ -0,0 +1,5 @@
+---
+"@appx-org/agent-server": patch
+---
+
+Forward opaque per-prompt actor receipts into provider request metadata.

packages/agent-server/src/http/sessionsRoutes.ts

@@ -44,6 +44,7 @@ import {
 	SessionModelSettingsResponseSchema,
 } from "../contract/schemas.js";
 import type { ProjectRuntime } from "../runtime/projectRuntime.js";
+import { APPX_ACTOR_RECEIPT_HEADER } from "../runtime/requestAttribution.js";
 import { subscribe } from "./sseBroker.js";
 
 /** Heartbeat cadence for SSE keepalive. Keeps proxies / LBs from closing idle streams. */
@@ -383,8 +384,9 @@ export function createSessionsApp(runtime: ProjectRuntime | ProjectRuntimeResolv
 			const { text } = c.req.valid("json");
 			const session = await runtime.getSession(id);
 			if (!session) return c.json({ error: "session not found" }, 404);
+			const actorReceipt = normalizeActorReceipt(c.req.header(APPX_ACTOR_RECEIPT_HEADER));
 			// Fire-and-forget: events flow over SSE, errors surface there too.
-			session.sendPrompt(text).catch((err) => {
+			session.sendPrompt(text, actorReceipt).catch((err) => {
 				console.error("[agent-server] prompt failed:", err);
 			});
 			return c.json({ ok: true } as const, 200);
@@ -527,3 +529,9 @@ export function createSessionsApp(runtime: ProjectRuntime | ProjectRuntimeResolv
 
 	return app;
 }
+
+function normalizeActorReceipt(value: string | undefined): string | undefined {
+	const receipt = value?.trim();
+	if (!receipt || receipt.length > 4096) return undefined;
+	return receipt;
+}

packages/agent-server/src/runtime/projectRuntime.ts

@@ -54,6 +54,7 @@ import type { AgentCredentialsService } from "../credentials/credentialsService.
 import type { ThinkingLevel } from "../shared/thinking.js";
 import { buildDeploymentPromptSection, type Deployment } from "./deployment.js";
 import { ProjectSession } from "./projectSession.js";
+import { ActorReceiptContext } from "./requestAttribution.js";
 
 type SessionModel = NonNullable<CreateAgentSessionOptions["model"]>;
 
@@ -261,6 +262,11 @@ export class ProjectRuntime {
 			);
 		}
 
+		// Receipt state is private to the host. Its extension is appended after
+		// caller extensions so request-controlled project code cannot replace the
+		// value forwarded by the authenticated HTTP boundary.
+		const actorReceipts = new ActorReceiptContext();
+
 		// Build the services bundle. Pi creates ResourceLoader +
 		// SettingsManager here, runs reload() exactly once, and registers
 		// extension-provided custom providers into the (shared)
@@ -275,7 +281,7 @@ export class ProjectRuntime {
 				additionalSkillPaths: config.skillPaths,
 				additionalPromptTemplatePaths: config.promptTemplatePaths,
 				additionalThemePaths: config.themePaths,
-				extensionFactories: config.extensionFactories,
+				extensionFactories: [...(config.extensionFactories ?? []), actorReceipts.extensionFactory],
 				noExtensions: config.noExtensions,
 				noSkills: config.noSkills,
 				noPromptTemplates: config.noPromptTemplates,
@@ -330,10 +336,15 @@ export class ProjectRuntime {
 				logger,
 			},
 			services,
+			actorReceipts,
 		);
 	}
 
-	private constructor(fields: ProjectRuntimeFields, services: AgentSessionServices) {
+	private constructor(
+		fields: ProjectRuntimeFields,
+		services: AgentSessionServices,
+		private readonly actorReceipts: ActorReceiptContext,
+	) {
 		this.projectDir = fields.projectDir;
 		this.sessionsDir = fields.sessionsDir;
 		this.credentials = fields.credentials;
@@ -368,6 +379,8 @@ export class ProjectRuntime {
 			credentials: this.credentials,
 			modelRegistry: this.services.modelRegistry,
 			logger: this.logger,
+			withActorReceipt: (receipt, callback) => this.actorReceipts.run(receipt, callback),
+			clearActorReceipt: () => this.actorReceipts.clear(session.sessionId),
 		});
 		this.sessions.set(ps.sessionId, ps);
 		return ps;

packages/agent-server/src/runtime/projectSession.ts

@@ -71,6 +71,8 @@ export type ProjectSessionDeps = {
 	credentials: AgentCredentialsService;
 	modelRegistry: Pick<ModelRegistry, "find">;
 	logger: Pick<Console, "log" | "error">;
+	withActorReceipt?: <T>(receipt: string | undefined, callback: () => T | Promise<T>) => Promise<T>;
+	clearActorReceipt?: () => void;
 };
 
 export class ProjectSession {
@@ -194,18 +196,25 @@ export class ProjectSession {
 	 * Send a user prompt. Events flow over SSE to subscribers. Returns once
 	 * the prompt has been queued; the agent runs asynchronously.
 	 */
-	async sendPrompt(text: string): Promise<void> {
-		await this.extensionsReady;
-		if (this.session.isStreaming) {
-			// While the agent is streaming, prompt() requires a streamingBehavior.
-			// "steer" queues the message for delivery as soon as the current
-			// assistant turn's tool calls finish — i.e. it actually interrupts
-			// the agent's plan rather than waiting for it to fully stop
-			// ("followUp"). Equivalent to session.steer(text).
-			await this.session.prompt(text, { streamingBehavior: "steer" });
+	async sendPrompt(text: string, actorReceipt?: string): Promise<void> {
+		const send = async () => {
+			await this.extensionsReady;
+			if (this.session.isStreaming) {
+				// While the agent is streaming, prompt() requires a streamingBehavior.
+				// "steer" queues the message for delivery as soon as the current
+				// assistant turn's tool calls finish — i.e. it actually interrupts
+				// the agent's plan rather than waiting for it to fully stop
+				// ("followUp"). Equivalent to session.steer(text).
+				await this.session.prompt(text, { streamingBehavior: "steer" });
+				return;
+			}
+			await this.session.prompt(text);
+		};
+		if (this.deps.withActorReceipt) {
+			await this.deps.withActorReceipt(actorReceipt, send);
 			return;
 		}
-		await this.session.prompt(text);
+		await send();
 	}
 
 	/**
@@ -243,6 +252,7 @@ export class ProjectSession {
 	async dispose(): Promise<void> {
 		if (this.disposed) return;
 		this.disposed = true;
+		this.deps.clearActorReceipt?.();
 		this.unsubscribeEvents();
 		for (const pending of this.pendingExtensionUi.values()) {
 			if (pending.timer) clearTimeout(pending.timer);

packages/agent-server/src/runtime/requestAttribution.ts

@@ -0,0 +1,97 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+import type { ExtensionFactory } from "@earendil-works/pi-coding-agent";
+
+export const APPX_ACTOR_RECEIPT_HEADER = "x-appx-actor-receipt";
+export const APPX_ACTOR_RECEIPT_METADATA_KEY = "openorange_app_actor_receipt";
+
+type IncomingReceipt = {
+	receipt: string | undefined;
+	sessionId?: string;
+};
+
+/**
+ * Carries an opaque gateway receipt from one HTTP prompt to the provider calls
+ * caused by that exact user message. Steering prompts are queued separately and
+ * become active only when Pi delivers their user message to the running agent.
+ */
+export class ActorReceiptContext {
+	private readonly incoming = new AsyncLocalStorage<IncomingReceipt>();
+	private readonly queued = new Map<string, IncomingReceipt[]>();
+	private readonly active = new Map<string, string>();
+
+	readonly extensionFactory: ExtensionFactory = (pi) => {
+		pi.on("input", (_event, context) => {
+			const incoming = this.incoming.getStore();
+			if (!incoming) return;
+			const sessionId = context.sessionManager.getSessionId();
+			const queue = this.queued.get(sessionId) ?? [];
+			incoming.sessionId = sessionId;
+			queue.push(incoming);
+			this.queued.set(sessionId, queue);
+		});
+
+		pi.on("message_start", (event, context) => {
+			if (event.message.role !== "user") return;
+			const sessionId = context.sessionManager.getSessionId();
+			const queue = this.queued.get(sessionId);
+			const receipt = queue?.shift()?.receipt;
+			if (!queue?.length) this.queued.delete(sessionId);
+			if (receipt) this.active.set(sessionId, receipt);
+			else this.active.delete(sessionId);
+		});
+
+		pi.on("before_provider_request", (event, context) => {
+			const incoming = this.incoming.getStore();
+			const receipt = incoming ? incoming.receipt : this.active.get(context.sessionManager.getSessionId());
+			return applyActorReceipt(event.payload, receipt);
+		});
+
+		pi.on("session_shutdown", (_event, context) => {
+			this.clear(context.sessionManager.getSessionId());
+		});
+	};
+
+	async run<T>(receipt: string | undefined, callback: () => T | Promise<T>): Promise<T> {
+		const incoming: IncomingReceipt = { receipt };
+		try {
+			return await this.incoming.run(incoming, callback);
+		} catch (error) {
+			this.removeQueued(incoming);
+			throw error;
+		}
+	}
+
+	clear(sessionId: string): void {
+		this.queued.delete(sessionId);
+		this.active.delete(sessionId);
+	}
+
+	private removeQueued(incoming: IncomingReceipt): void {
+		if (!incoming.sessionId) return;
+		const queue = this.queued.get(incoming.sessionId);
+		if (!queue) return;
+		const index = queue.indexOf(incoming);
+		if (index !== -1) queue.splice(index, 1);
+		if (queue.length === 0) this.queued.delete(incoming.sessionId);
+	}
+}
+
+/**
+ * Replace request-controlled attribution metadata with the receipt accepted by
+ * the agent-server HTTP boundary. The internal extension using this helper is
+ * loaded after project extensions, so app code cannot forge or replace it.
+ */
+export function applyActorReceipt(payload: unknown, receipt: string | undefined): unknown {
+	if (!isRecord(payload)) return payload;
+
+	const existingMetadata = isRecord(payload.metadata) ? payload.metadata : {};
+	const metadata: Record<string, unknown> = { ...existingMetadata };
+	delete metadata[APPX_ACTOR_RECEIPT_METADATA_KEY];
+	if (receipt) metadata[APPX_ACTOR_RECEIPT_METADATA_KEY] = receipt;
+
+	return { ...payload, metadata };
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+	return typeof value === "object" && value !== null && !Array.isArray(value);
+}