diff --git a/.changeset/quiet-hounds-attribute.md b/.changeset/quiet-hounds-attribute.md new file mode 100644 index 0000000..8913406 --- /dev/null +++ b/.changeset/quiet-hounds-attribute.md @@ -0,0 +1,5 @@ +--- +"@appx-org/agent-server": patch +--- + +Forward opaque per-prompt actor receipts into provider request metadata. diff --git a/packages/agent-server/src/http/sessionsRoutes.ts b/packages/agent-server/src/http/sessionsRoutes.ts index 149938e..6ccc27e 100644 --- a/packages/agent-server/src/http/sessionsRoutes.ts +++ b/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; +} diff --git a/packages/agent-server/src/runtime/projectRuntime.ts b/packages/agent-server/src/runtime/projectRuntime.ts index b4c4c87..74d52c2 100644 --- a/packages/agent-server/src/runtime/projectRuntime.ts +++ b/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; @@ -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; diff --git a/packages/agent-server/src/runtime/projectSession.ts b/packages/agent-server/src/runtime/projectSession.ts index 8bdd6b2..77a2e05 100644 --- a/packages/agent-server/src/runtime/projectSession.ts +++ b/packages/agent-server/src/runtime/projectSession.ts @@ -71,6 +71,8 @@ export type ProjectSessionDeps = { credentials: AgentCredentialsService; modelRegistry: Pick; logger: Pick; + withActorReceipt?: (receipt: string | undefined, callback: () => T | Promise) => Promise; + 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 { - 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 { + 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 { 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); diff --git a/packages/agent-server/src/runtime/requestAttribution.ts b/packages/agent-server/src/runtime/requestAttribution.ts new file mode 100644 index 0000000..bfec2fa --- /dev/null +++ b/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(); + private readonly queued = new Map(); + private readonly active = new Map(); + + 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(receipt: string | undefined, callback: () => T | Promise): Promise { + 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 = { ...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 { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/packages/agent-server/test/requestAttribution.test.ts b/packages/agent-server/test/requestAttribution.test.ts new file mode 100644 index 0000000..c8e4ed3 --- /dev/null +++ b/packages/agent-server/test/requestAttribution.test.ts @@ -0,0 +1,181 @@ +import assert from "node:assert/strict"; +import { describe, test } from "node:test"; +import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent"; +import { Hono } from "hono"; +import { createSessionsApp } from "../src/http/sessionsRoutes.js"; +import type { ProjectRuntime } from "../src/runtime/projectRuntime.js"; +import { + ActorReceiptContext, + APPX_ACTOR_RECEIPT_HEADER, + APPX_ACTOR_RECEIPT_METADATA_KEY, + applyActorReceipt, +} from "../src/runtime/requestAttribution.js"; + +type Handler = (event: any, context: ExtensionContext) => unknown; + +function bindReceiptContext(context: ActorReceiptContext) { + const handlers = new Map(); + const api = { + on(event: string, handler: Handler) { + handlers.set(event, handler); + }, + } as unknown as ExtensionAPI; + context.extensionFactory(api); + const extensionContext = (sessionId: string) => + ({ sessionManager: { getSessionId: () => sessionId } }) as unknown as ExtensionContext; + return { + input: (sessionId: string, receipt: string | undefined) => + context.run(receipt, () => + handlers.get("input")!({ type: "input", text: "prompt" }, extensionContext(sessionId)), + ), + failedInput: (sessionId: string, receipt: string | undefined) => + context.run(receipt, () => { + handlers.get("input")!({ type: "input", text: "prompt" }, extensionContext(sessionId)); + throw new Error("prompt rejected"); + }), + deliverUserMessage: (sessionId: string) => + handlers.get("message_start")!( + { type: "message_start", message: { role: "user", content: "prompt" } }, + extensionContext(sessionId), + ), + providerPayload: (sessionId: string) => + handlers.get("before_provider_request")!( + { + type: "before_provider_request", + payload: { metadata: { [APPX_ACTOR_RECEIPT_METADATA_KEY]: "forged", keep: "yes" } }, + }, + extensionContext(sessionId), + ), + providerDuringPrompt: (sessionId: string, receipt: string | undefined) => + context.run(receipt, () => + handlers.get("before_provider_request")!( + { type: "before_provider_request", payload: { metadata: { keep: "yes" } } }, + extensionContext(sessionId), + ), + ), + }; +} + +describe("AppX actor receipt forwarding", () => { + test("the prompt route passes the opaque receipt without adding it to the prompt body", async () => { + const calls: Array<{ text: string; receipt: string | undefined }> = []; + const runtime = { + async getSession(id: string) { + if (id !== "session-1") return null; + return { + sendPrompt(text: string, receipt?: string) { + calls.push({ text, receipt }); + return Promise.resolve(); + }, + }; + }, + } as unknown as ProjectRuntime; + const app = new Hono(); + app.route("/v1/projects/:projectId", createSessionsApp(runtime)); + + const response = await app.request("/v1/projects/project-1/sessions/session-1/prompt", { + method: "POST", + headers: { + "content-type": "application/json", + [APPX_ACTOR_RECEIPT_HEADER]: "v1.opaque.signature", + }, + body: JSON.stringify({ text: "build an app" }), + }); + + assert.equal(response.status, 200); + assert.deepEqual(calls, [{ text: "build an app", receipt: "v1.opaque.signature" }]); + }); + + test("activates a receipt only when its user message is delivered", async () => { + const state = bindReceiptContext(new ActorReceiptContext()); + await state.input("session-1", "v1.trusted.signature"); + assert.deepEqual(await state.providerPayload("session-1"), { metadata: { keep: "yes" } }); + + await state.deliverUserMessage("session-1"); + assert.deepEqual(await state.providerPayload("session-1"), { + metadata: { + [APPX_ACTOR_RECEIPT_METADATA_KEY]: "v1.trusted.signature", + keep: "yes", + }, + }); + }); + + test("switches a steering receipt at delivery without changing the active run early", async () => { + const state = bindReceiptContext(new ActorReceiptContext()); + await state.input("session-1", "v1.first.signature"); + await state.deliverUserMessage("session-1"); + await state.input("session-1", "v1.second.signature"); + + assert.equal( + ((await state.providerPayload("session-1")) as any).metadata[APPX_ACTOR_RECEIPT_METADATA_KEY], + "v1.first.signature", + ); + await state.deliverUserMessage("session-1"); + assert.equal( + ((await state.providerPayload("session-1")) as any).metadata[APPX_ACTOR_RECEIPT_METADATA_KEY], + "v1.second.signature", + ); + }); + + test("uses the current prompt receipt for extension-command provider calls", async () => { + const state = bindReceiptContext(new ActorReceiptContext()); + await state.input("session-1", "v1.active.signature"); + await state.deliverUserMessage("session-1"); + + assert.equal( + ((await state.providerDuringPrompt("session-1", "v1.command.signature")) as any).metadata[ + APPX_ACTOR_RECEIPT_METADATA_KEY + ], + "v1.command.signature", + ); + assert.equal( + ((await state.providerPayload("session-1")) as any).metadata[APPX_ACTOR_RECEIPT_METADATA_KEY], + "v1.active.signature", + ); + }); + + test("keeps concurrently delivered sessions isolated", async () => { + const state = bindReceiptContext(new ActorReceiptContext()); + await Promise.all([ + state.input("session-1", "v1.first.signature"), + state.input("session-2", "v1.second.signature"), + ]); + await state.deliverUserMessage("session-2"); + await state.deliverUserMessage("session-1"); + + assert.equal( + ((await state.providerPayload("session-1")) as any).metadata[APPX_ACTOR_RECEIPT_METADATA_KEY], + "v1.first.signature", + ); + assert.equal( + ((await state.providerPayload("session-2")) as any).metadata[APPX_ACTOR_RECEIPT_METADATA_KEY], + "v1.second.signature", + ); + }); + + test("removes a queued receipt when Pi rejects the prompt before delivery", async () => { + const state = bindReceiptContext(new ActorReceiptContext()); + await assert.rejects(state.failedInput("session-1", "v1.stale.signature"), /prompt rejected/); + await state.input("session-1", "v1.current.signature"); + await state.deliverUserMessage("session-1"); + + assert.equal( + ((await state.providerPayload("session-1")) as any).metadata[APPX_ACTOR_RECEIPT_METADATA_KEY], + "v1.current.signature", + ); + }); + + test("an unreceipted delivered prompt clears prior and forged attribution", async () => { + const state = bindReceiptContext(new ActorReceiptContext()); + await state.input("session-1", "v1.first.signature"); + await state.deliverUserMessage("session-1"); + await state.input("session-1", undefined); + await state.deliverUserMessage("session-1"); + + assert.deepEqual(await state.providerPayload("session-1"), { metadata: { keep: "yes" } }); + assert.deepEqual( + applyActorReceipt({ metadata: { [APPX_ACTOR_RECEIPT_METADATA_KEY]: "forged", keep: "yes" } }, undefined), + { metadata: { keep: "yes" } }, + ); + }); +});