src/server/webhook-lane/feature.ts
Rama main · réplica de lectura del repositorio original.
/**
* Webhook lane — the public extension point for wiring an EXTERNAL CHAT PRODUCT to this agent.
*
* The shape of the problem, name-neutrally: some other product holds the conversation UI. It wants
* to hand this agent one turn and render the answer as it is produced. It cannot hold an HTTP
* response open for the minutes an agent run takes, and it needs the answer authenticated when it
* comes back. So the lane is TWO independent requests:
*
* 1. INGRESS — the product POSTs one turn to a route THIS module mounts for you.
* 2. CALLBACK — this agent POSTs the answer back, progressively, to a URL the product supplied
* IN that turn, each delivery HMAC-signed with a secret the product supplied too.
*
* The core registers NOTHING here, exactly as in `features.ts`. It ships the machinery — the signed
* transport, the flush/coalescing policy, the segment wire shape, and the turn-ingress contract —
* and an add-on names its own product:
*
* ```ts
* import { registerFeature } from 'shraga/server/features';
* import { createWebhookLaneFeature } from 'shraga/server/webhook-lane/feature';
*
* registerFeature(createWebhookLaneFeature({
* name: 'acme',
* route: '/api/acme/turn',
* onTurnAccepted: (link) => rememberLink(link), // your persistence, your policy
* }));
* ```
*
* WHAT THE ADD-ON OWNS, and why the core cannot: where links are persisted and under what filename,
* which links may receive PROACTIVE posts (an owner-notice policy is a product decision — see
* `postNotice` in `streamer.ts` for the transport), the route path, and any CLI surface. None of
* that is generalizable without inventing policy, so the seam hands it back.
*
* TRUST. The ingress is authenticated with an ordinary shraga API key (`POST /api/api-keys`), so
* reaching the route requires a credential the owner minted. The callback URL + secret arrive IN
* that authenticated request — per turn — which is what makes a rotated webhook secret take effect
* on the very next message with no configuration on this side. The URL is still validated here
* (https, or loopback for development) rather than trusted because the request authenticated: we
* hold the owner's credential and will POST to whatever it names.
*/
import crypto from 'node:crypto';
import type { ServerFeature, FeatureContext } from '../features.ts';
import { streamChat } from '../claude.ts';
import { getMcpConfig } from '../mcp.ts';
import {
appendMessage, upsertSession, setRunStatus, acquireSessionLock, releaseSessionLock,
type ConvBlock,
} from '../sessions.ts';
import { apiKeyPrincipal, validateApiKey } from '../api-keys.ts';
import type { Principal } from '../security/principal.ts';
import { admitTurn, requestIp } from '../security/runtime.ts';
import { writeDenial } from '../security/guard.ts';
import { WebhookStreamer, type WebhookTarget } from './streamer.ts';
/** One receiver connection, as learned from a turn. Handed to `onTurnAccepted` so an add-on can
* persist it and later reach the same conversation proactively.
*
* `uid` is the shraga user whose API key opened this link, and `email` is that user's address,
* taken from `validateApiKey` and never from the request body. An add-on that fans PROACTIVE
* notices out to "owners" must join on the email — a uid does not join to an owner list. */
export type WebhookLaneLink = WebhookTarget & { convId: string; at: number; uid: string; email?: string };
export interface WebhookLaneOptions {
/** Feature name (must be unique in the registry). Also the default transcript `channel` and
* turn-context `source` tag, so a one-line registration is coherent. */
name: string;
/** Absolute ingress path to mount, e.g. `/api/acme/turn`. Named by the add-on: the core must not
* contain a product's route. */
route: string;
/** Transcript channel tag stamped on the user message. Defaults to `name`. */
channel?: string;
/** `context.source` handed to the turn-context seam. Defaults to `name`. */
source?: string;
/**
* Called once per ACCEPTED turn, before the run starts, with the connection this turn negotiated.
* This is the add-on's chance to persist the link for its proactive lane.
*
* NOTIFICATION, NOT A GATE — and deliberately so. Whether a `connId` may be re-pointed at a
* different user's callback is the add-on's model, so the add-on refuses the WRITE inside this
* hook; the turn itself still runs and still answers on the callback the caller supplied, because
* that caller authenticated with their own API key and is entitled to their own answer. Throwing
* is contained: a broken persistence layer must not cost the user their reply.
*/
onTurnAccepted?(link: WebhookLaneLink): void;
/** Streamer knobs, if the defaults in `streamer.ts` do not suit the receiver. */
streamer?: { flushInterval?: number; flushThreshold?: number; postTimeout?: number };
}
/** The ingress body, as it arrives on the wire. This IS the contract a receiver implements. */
export interface TurnRequestBody {
/** Stable id for the receiver connection. Inside the signed material of every callback. */
connId?: string;
/** The receiver's conversation id. Echoed on every callback. */
convId?: string;
/** Agent session to run in. Defaults to `convId`, so a receiver may omit it entirely. */
sessionId?: string;
/** The row to patch with the answer. The receiver mints it before calling. */
msgId?: string;
prompt?: string;
callback?: { url?: string; secret?: string };
/** Receiver capability negotiation. `'segments'` means "I can store and render structured tool
* segments" — see `WebhookStreamerOptions.sendSegments`. ABSENT means no, and the receiver gets
* the flattened in-text tool markers instead. A list, so it can grow without a new field. */
accepts?: unknown;
}
/** Run one turn, streaming into the receiver's row. Errors settle the row VISIBLY — a reader must
* never be left watching a "typing…" placeholder that will never resolve. Exported so an add-on
* with its own ingress (a queue consumer, say) can reuse the run half without the route half. */
export async function runWebhookTurn(args: {
callback: WebhookTarget; convId: string; msgId: string; sessionId: string; prompt: string;
uid: string; userEmail: string; sendSegments: boolean;
channel: string; source: string;
/** Who the turn runs for — the API key holder for the built-in ingress. */
principal: Principal;
streamer?: WebhookLaneOptions['streamer'];
}): Promise<void> {
const { callback, convId, msgId, sessionId, prompt, uid, userEmail, sendSegments, channel, source, principal } = args;
const streamer = new WebhookStreamer({ callback, convId, msgId, sendSegments, ...(args.streamer ?? {}) });
const abortController = new AbortController();
// Lock origin is 'api': the union in sessions.ts is a closed set ('web'|'slack'|'scheduler'|
// 'api') and this is an authenticated API caller. Widening it just to label the medium would
// touch recovery and status code paths for no behavioural gain.
if (!acquireSessionLock(sessionId, 'api', abortController)) {
// sessionId defaults to convId, so this is genuinely "you sent two messages into the same
// thread while the first was still running". Say so rather than dropping it silently.
await streamer.fail('That conversation is already processing a message — wait for it to finish.');
return;
}
upsertSession(sessionId, prompt, { uid, email: userEmail });
appendMessage(sessionId, { id: crypto.randomUUID(), role: 'user', blocks: [{ type: 'text', text: prompt }], channel });
setRunStatus(sessionId, 'running', 'web');
const blocks: ConvBlock[] = [];
let text = '';
try {
for await (const ev of streamChat({
principal, prompt, sessionId, uid, userEmail,
mcpServers: getMcpConfig(uid),
abortController,
context: { source, user: userEmail },
onPermissionRequest: async () => ({ allow: true }),
})) {
if (ev.type === 'text_delta') { text += ev.text; streamer.feed({ type: 'text_delta', text: ev.text }); }
else if (ev.type === 'tool_use') {
if (text) { blocks.push({ type: 'text', text }); text = ''; }
blocks.push({ type: 'tool_use', tool: ev.tool, toolUseId: ev.toolUseId, input: ev.input });
streamer.feed({ type: 'tool_use', tool: ev.tool, toolUseId: ev.toolUseId, input: ev.input });
}
else if (ev.type === 'tool_result') {
blocks.push({ type: 'tool_result', toolUseId: ev.toolUseId, output: ev.output });
// The `running` → `completed`/`error` transition. Fed unconditionally; the streamer ignores
// it unless the receiver negotiated segments.
streamer.feed({ type: 'tool_result', toolUseId: ev.toolUseId, output: ev.output, isError: ev.isError });
}
else if (ev.type === 'done') break;
else if (ev.type === 'error') {
if (text) { blocks.push({ type: 'text', text }); text = ''; }
blocks.push({ type: 'error', text: ev.message });
await streamer.fail(ev.message);
return;
}
}
if (text) blocks.push({ type: 'text', text });
await streamer.finish();
} catch (err) {
console.error(`[${channel}] turn failed:`, (err as Error).message);
await streamer.fail((err as Error).message || 'agent error');
} finally {
// The transcript is persisted whatever happened, so the shraga UI and the next turn's context
// see the same history the receiver saw.
if (blocks.length) appendMessage(sessionId, { id: crypto.randomUUID(), role: 'assistant', blocks });
if (releaseSessionLock(sessionId, abortController)) setRunStatus(sessionId, 'idle');
}
}
/**
* Build a `ServerFeature` that mounts one turn ingress. Register it with `registerFeature()`.
*
* No capability flag is declared: `flags` tells the CLIENT that a surface exists, and this lane is
* driven entirely by the external product calling in — nothing in the client gates on it. An add-on
* that DOES have a client surface can declare its own flag by spreading this feature.
*/
export function createWebhookLaneFeature(opts: WebhookLaneOptions): ServerFeature {
const channel = opts.channel ?? opts.name;
const source = opts.source ?? opts.name;
// Per-INSTANCE, not module-global: two lanes may be registered, and a shared guard would let the
// first one mounted silence the second.
let mounted = false;
return {
name: opts.name,
register(ctx: FeatureContext): void {
if (ctx.passive || mounted) return;
mounted = true;
ctx.app.post(opts.route, (req, res) => {
const bearer = /^Bearer\s+(.+)$/i.exec(req.get('authorization') ?? '')?.[1];
const caller = bearer ? validateApiKey(bearer) : null;
if (!caller) return void res.status(401).json({ error: 'unauthorized' });
const { connId, convId, sessionId, msgId, prompt, callback, accepts } = req.body as TurnRequestBody;
if (!connId || !convId || !msgId || !prompt) return void res.status(400).json({ error: 'connId, convId, msgId and prompt are required' });
if (!callback?.url || !callback?.secret) return void res.status(400).json({ error: 'callback.url and callback.secret are required' });
try {
const u = new URL(callback.url);
if (u.protocol !== 'https:' && u.hostname !== 'localhost' && u.hostname !== '127.0.0.1') throw new Error('https required');
} catch { return void res.status(400).json({ error: 'callback.url must be a valid HTTPS URL' }); }
// Guard before any spend: blocklist → rate → concurrency (shadow unless SECURITY_ENFORCE).
const principal = apiKeyPrincipal(caller); // carries the key's role cap, so guard rank and the turn's role agree
const admission = admitTurn(principal, { ip: requestIp(req), channel });
if (!admission.ok) return void writeDenial(res, admission);
const cb: WebhookTarget = { url: callback.url, secret: callback.secret, connId };
try {
opts.onTurnAccepted?.({ ...cb, convId, at: Date.now(), uid: caller.uid, email: caller.email });
} catch (err) {
console.warn(`[${opts.name}] onTurnAccepted threw:`, (err as Error).message);
}
// ACCEPT, then run. The answer arrives on the callback, so holding this response open would
// only give the caller's trigger a socket to time out on.
try { res.json({ status: 'accepted', sessionId: sessionId || convId }); }
catch (err) { admission.release(); throw err; }
void runWebhookTurn({
callback: cb, convId, msgId, sessionId: sessionId || convId, prompt,
uid: caller.uid, userEmail: caller.email, principal,
sendSegments: Array.isArray(accepts) && accepts.includes('segments'),
channel, source, streamer: opts.streamer,
}).finally(admission.release);
});
console.log(`[${opts.name}] turn ingress mounted at POST ${opts.route}`);
},
};
}