diff --git a/.env.example b/.env.example index a3a46ae..323fff3 100644 --- a/.env.example +++ b/.env.example @@ -71,3 +71,19 @@ MCP_COMMAND_ALLOWLIST= PLUGIN_OAUTH_REDIRECT_URI= # Authorization-server hostnames allowed for plugin OAuth (comma-separated). PLUGIN_OAUTH_ISSUER_ALLOWLIST=auth.exa.ai + +# ── REST webhook (ZEN-93) ─────────────────────────────────────────────────── +# Machine-to-machine trigger: POST /integrations/webhook +# +# WEBHOOK_BEARER_TOKEN: invent this yourself (`openssl rand -hex 32`). +# Empty disables the endpoint (404). Not created by the app or the database. +# +# WEBHOOK_USER_ID: UUID of an existing row in app_users. The webhook has no +# browser login, so it borrows that user's VCS token and LLM connection. +# On first boot there are no users yet — leave this empty. Sign in via the +# dashboard (GitHub App), connect VCS + a model, then: +# psql "$DATABASE_URL" -c "select id, login from app_users;" +# Paste the id here and restart the controller. A dedicated service account +# is not implemented yet. +WEBHOOK_BEARER_TOKEN= +WEBHOOK_USER_ID= diff --git a/apps/controller/README.md b/apps/controller/README.md index e41069a..03b47c1 100644 --- a/apps/controller/README.md +++ b/apps/controller/README.md @@ -20,6 +20,8 @@ Decides what runs and when, resolves connected identities, mints run credentials | `db/migrations/` | plain SQL, generated by `db:generate`, readable in review | | `http/app.ts` | route composition, CORS, error handling | | `http/manual.ts` | request parsing shared by the run and session routes | +| `http/webhook.ts` | machine-to-machine REST trigger (ZEN-93) | +| `integrations/` | ingress + report-back (memory, REST webhook, GitHub issue-comment adapter) | | `http/runs.ts` | the operator API, including the resumable SSE stream | | `http/sessions.ts` | durable chat sessions and guarded follow-up turns | | `http/plugins.ts` | marketplace catalog, install, configure, OAuth connect/callback, operator publish/review | diff --git a/apps/controller/config.ts b/apps/controller/config.ts index 9d115ce..ccabe3f 100644 --- a/apps/controller/config.ts +++ b/apps/controller/config.ts @@ -74,6 +74,11 @@ const schema = z.object({ PLUGIN_OAUTH_REDIRECT_URI: z.string().default(""), /** Comma-separated authorization-server hostnames allowed for plugin OAuth. */ PLUGIN_OAUTH_ISSUER_ALLOWLIST: z.string().default("auth.exa.ai"), + + /** Empty disables POST /integrations/webhook. */ + WEBHOOK_BEARER_TOKEN: z.string().default(""), + /** app_users.id whose VCS + LLM connections the webhook uses. Empty = unset. */ + WEBHOOK_USER_ID: z.union([z.literal(""), z.string().uuid()]).default(""), }); export type Env = Readonly>; @@ -107,6 +112,10 @@ export interface Config { oauthRedirectUri: string; oauthIssuerAllowlist: string[]; }; + webhook: { + bearerToken: string | null; + userId: string | null; + }; /** Handed to provider factories so they can read their own variables. */ env: Env; } @@ -178,6 +187,10 @@ function build(env: Env): Config { .map((host) => host.trim().toLowerCase()) .filter((host) => host.length > 0), }, + webhook: { + bearerToken: value.WEBHOOK_BEARER_TOKEN.trim() || null, + userId: value.WEBHOOK_USER_ID.trim() || null, + }, env, }; } diff --git a/apps/controller/db/migrations/0015_unusual_slipstream.sql b/apps/controller/db/migrations/0015_unusual_slipstream.sql new file mode 100644 index 0000000..10df56d --- /dev/null +++ b/apps/controller/db/migrations/0015_unusual_slipstream.sql @@ -0,0 +1 @@ +ALTER TABLE "runs" ADD COLUMN "surface_ref" jsonb; \ No newline at end of file diff --git a/apps/controller/db/migrations/meta/0015_snapshot.json b/apps/controller/db/migrations/meta/0015_snapshot.json new file mode 100644 index 0000000..1cac63a --- /dev/null +++ b/apps/controller/db/migrations/meta/0015_snapshot.json @@ -0,0 +1,1682 @@ +{ + "id": "f8b7daa1-676e-434a-8e3d-1c1e3ecebd57", + "prevId": "601cc7dc-b335-489d-a18e-416f04b5e42f", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.app_users": { + "name": "app_users", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "github_user_id": { + "name": "github_user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "login": { + "name": "login", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "display_name": { + "name": "display_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "avatar_url": { + "name": "avatar_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "app_users_github_user_idx": { + "name": "app_users_github_user_idx", + "columns": [ + { + "expression": "github_user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.llm_connections": { + "name": "llm_connections", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "display_name": { + "name": "display_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "auth_type": { + "name": "auth_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "api": { + "name": "api", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "base_url": { + "name": "base_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "model": { + "name": "model", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "models": { + "name": "models", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "context_window": { + "name": "context_window", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "max_tokens": { + "name": "max_tokens", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "credential": { + "name": "credential", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "is_default": { + "name": "is_default", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "llm_connections_user_updated_idx": { + "name": "llm_connections_user_updated_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "updated_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "llm_connections_user_id_app_users_id_fk": { + "name": "llm_connections_user_id_app_users_id_fk", + "tableFrom": "llm_connections", + "tableTo": "app_users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.oauth_states": { + "name": "oauth_states", + "schema": "", + "columns": { + "state": { + "name": "state", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "return_to": { + "name": "return_to", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "code_verifier": { + "name": "code_verifier", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "oauth_states_expiry_idx": { + "name": "oauth_states_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "oauth_states_user_id_app_users_id_fk": { + "name": "oauth_states_user_id_app_users_id_fk", + "tableFrom": "oauth_states", + "tableTo": "app_users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugin_audit_log": { + "name": "plugin_audit_log", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "actor_user_id": { + "name": "actor_user_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "plugin_name": { + "name": "plugin_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "action": { + "name": "action", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "detail": { + "name": "detail", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "plugin_audit_log_created_idx": { + "name": "plugin_audit_log_created_idx", + "columns": [ + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "plugin_audit_log_actor_user_id_app_users_id_fk": { + "name": "plugin_audit_log_actor_user_id_app_users_id_fk", + "tableFrom": "plugin_audit_log", + "tableTo": "app_users", + "columnsFrom": [ + "actor_user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugin_oauth_clients": { + "name": "plugin_oauth_clients", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "issuer": { + "name": "issuer", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "redirect_uri": { + "name": "redirect_uri", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "client_id": { + "name": "client_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "plugin_oauth_clients_issuer_redirect_idx": { + "name": "plugin_oauth_clients_issuer_redirect_idx", + "columns": [ + { + "expression": "issuer", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "redirect_uri", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugin_oauth_tokens": { + "name": "plugin_oauth_tokens", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "access_encrypted": { + "name": "access_encrypted", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "refresh_encrypted": { + "name": "refresh_encrypted", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "plugin_oauth_tokens_user_id_app_users_id_fk": { + "name": "plugin_oauth_tokens_user_id_app_users_id_fk", + "tableFrom": "plugin_oauth_tokens", + "tableTo": "app_users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "plugin_oauth_tokens_plugin_id_plugins_id_fk": { + "name": "plugin_oauth_tokens_plugin_id_plugins_id_fk", + "tableFrom": "plugin_oauth_tokens", + "tableTo": "plugins", + "columnsFrom": [ + "plugin_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "plugin_oauth_tokens_user_id_plugin_id_pk": { + "name": "plugin_oauth_tokens_user_id_plugin_id_pk", + "columns": [ + "user_id", + "plugin_id" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugin_settings": { + "name": "plugin_settings", + "schema": "", + "columns": { + "plugin_id": { + "name": "plugin_id", + "type": "uuid", + "primaryKey": true, + "notNull": true + }, + "install_mode": { + "name": "install_mode", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'default_off'" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "plugin_settings_plugin_id_plugins_id_fk": { + "name": "plugin_settings_plugin_id_plugins_id_fk", + "tableFrom": "plugin_settings", + "tableTo": "plugins", + "columnsFrom": [ + "plugin_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugin_user_state": { + "name": "plugin_user_state", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "override": { + "name": "override", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "installed_version_id": { + "name": "installed_version_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "plugin_user_state_user_id_app_users_id_fk": { + "name": "plugin_user_state_user_id_app_users_id_fk", + "tableFrom": "plugin_user_state", + "tableTo": "app_users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "plugin_user_state_plugin_id_plugins_id_fk": { + "name": "plugin_user_state_plugin_id_plugins_id_fk", + "tableFrom": "plugin_user_state", + "tableTo": "plugins", + "columnsFrom": [ + "plugin_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "plugin_user_state_installed_version_id_plugin_versions_id_fk": { + "name": "plugin_user_state_installed_version_id_plugin_versions_id_fk", + "tableFrom": "plugin_user_state", + "tableTo": "plugin_versions", + "columnsFrom": [ + "installed_version_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "plugin_user_state_user_id_plugin_id_pk": { + "name": "plugin_user_state_user_id_plugin_id_pk", + "columns": [ + "user_id", + "plugin_id" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugin_user_variables": { + "name": "plugin_user_variables", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "value_encrypted": { + "name": "value_encrypted", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "plugin_user_variables_user_id_app_users_id_fk": { + "name": "plugin_user_variables_user_id_app_users_id_fk", + "tableFrom": "plugin_user_variables", + "tableTo": "app_users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "plugin_user_variables_plugin_id_plugins_id_fk": { + "name": "plugin_user_variables_plugin_id_plugins_id_fk", + "tableFrom": "plugin_user_variables", + "tableTo": "plugins", + "columnsFrom": [ + "plugin_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "plugin_user_variables_user_id_plugin_id_name_pk": { + "name": "plugin_user_variables_user_id_plugin_id_name_pk", + "columns": [ + "user_id", + "plugin_id", + "name" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugin_versions": { + "name": "plugin_versions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "plugin_id": { + "name": "plugin_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "version": { + "name": "version", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "source": { + "name": "source", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "artifact_path": { + "name": "artifact_path", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "components": { + "name": "components", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{\"skills\":false,\"mcp\":false}'::jsonb" + }, + "review_status": { + "name": "review_status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'draft'" + }, + "manifest": { + "name": "manifest", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "plugin_versions_plugin_version_idx": { + "name": "plugin_versions_plugin_version_idx", + "columns": [ + { + "expression": "plugin_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "version", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "plugin_versions_status_idx": { + "name": "plugin_versions_status_idx", + "columns": [ + { + "expression": "review_status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "plugin_versions_plugin_id_plugins_id_fk": { + "name": "plugin_versions_plugin_id_plugins_id_fk", + "tableFrom": "plugin_versions", + "tableTo": "plugins", + "columnsFrom": [ + "plugin_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugins": { + "name": "plugins", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "publisher": { + "name": "publisher", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'Zen8'" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "plugins_name_idx": { + "name": "plugins_name_idx", + "columns": [ + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.run_events": { + "name": "run_events", + "schema": "", + "columns": { + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "seq": { + "name": "seq", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "type": { + "name": "type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "data": { + "name": "data", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "run_events_run_id_runs_id_fk": { + "name": "run_events_run_id_runs_id_fk", + "tableFrom": "run_events", + "tableTo": "runs", + "columnsFrom": [ + "run_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "run_events_run_id_seq_pk": { + "name": "run_events_run_id_seq_pk", + "columns": [ + "run_id", + "seq" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.runs": { + "name": "runs", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "session_id": { + "name": "session_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "turn_number": { + "name": "turn_number", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'queued'" + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "repo_full_name": { + "name": "repo_full_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "trigger": { + "name": "trigger", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "surface_ref": { + "name": "surface_ref", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "model": { + "name": "model", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "thinking_level": { + "name": "thinking_level", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'medium'" + }, + "model_connection_id": { + "name": "model_connection_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "plugins": { + "name": "plugins", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "callback_token": { + "name": "callback_token", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "sandbox_provider": { + "name": "sandbox_provider", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "sandbox_id": { + "name": "sandbox_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "sandbox_stopped_at": { + "name": "sandbox_stopped_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "event_seq": { + "name": "event_seq", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "last_event_at": { + "name": "last_event_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "attempt": { + "name": "attempt", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "claimed_at": { + "name": "claimed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "claim_expires_at": { + "name": "claim_expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "deadline_at": { + "name": "deadline_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "runs_status_created_idx": { + "name": "runs_status_created_idx", + "columns": [ + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "runs_created_idx": { + "name": "runs_created_idx", + "columns": [ + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "runs_sandbox_idx": { + "name": "runs_sandbox_idx", + "columns": [ + { + "expression": "sandbox_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"runs\".\"sandbox_id\" is not null and \"runs\".\"sandbox_stopped_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "runs_user_id_app_users_id_fk": { + "name": "runs_user_id_app_users_id_fk", + "tableFrom": "runs", + "tableTo": "app_users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + }, + "runs_session_id_sessions_id_fk": { + "name": "runs_session_id_sessions_id_fk", + "tableFrom": "runs", + "tableTo": "sessions", + "columnsFrom": [ + "session_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "runs_model_connection_id_llm_connections_id_fk": { + "name": "runs_model_connection_id_llm_connections_id_fk", + "tableFrom": "runs", + "tableTo": "llm_connections", + "columnsFrom": [ + "model_connection_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "restrict", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.sessions": { + "name": "sessions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "repo_full_name": { + "name": "repo_full_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "repo": { + "name": "repo", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "model": { + "name": "model", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "model_connection_id": { + "name": "model_connection_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "active_run_id": { + "name": "active_run_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "latest_run_id": { + "name": "latest_run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "turn_count": { + "name": "turn_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 1 + }, + "agent_checkpoint": { + "name": "agent_checkpoint", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "sandbox_provider": { + "name": "sandbox_provider", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "sandbox_id": { + "name": "sandbox_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "workspace_expires_at": { + "name": "workspace_expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "sessions_updated_idx": { + "name": "sessions_updated_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "sessions_workspace_expiry_idx": { + "name": "sessions_workspace_expiry_idx", + "columns": [ + { + "expression": "workspace_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"sessions\".\"sandbox_id\" is not null and \"sessions\".\"active_run_id\" is null", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "sessions_user_id_app_users_id_fk": { + "name": "sessions_user_id_app_users_id_fk", + "tableFrom": "sessions", + "tableTo": "app_users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + }, + "sessions_model_connection_id_llm_connections_id_fk": { + "name": "sessions_model_connection_id_llm_connections_id_fk", + "tableFrom": "sessions", + "tableTo": "llm_connections", + "columnsFrom": [ + "model_connection_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "restrict", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.vcs_connections": { + "name": "vcs_connections", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "account_name": { + "name": "account_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "access_token": { + "name": "access_token", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "refresh_token": { + "name": "refresh_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "vcs_connections_user_provider_idx": { + "name": "vcs_connections_user_provider_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "provider", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "vcs_connections_user_id_app_users_id_fk": { + "name": "vcs_connections_user_id_app_users_id_fk", + "tableFrom": "vcs_connections", + "tableTo": "app_users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.web_sessions": { + "name": "web_sessions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "user_id": { + "name": "user_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "token_hash": { + "name": "token_hash", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "web_sessions_token_hash_idx": { + "name": "web_sessions_token_hash_idx", + "columns": [ + { + "expression": "token_hash", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "web_sessions_user_expiry_idx": { + "name": "web_sessions_user_expiry_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "web_sessions_user_id_app_users_id_fk": { + "name": "web_sessions_user_id_app_users_id_fk", + "tableFrom": "web_sessions", + "tableTo": "app_users", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/apps/controller/db/migrations/meta/_journal.json b/apps/controller/db/migrations/meta/_journal.json index 47a405e..2458c9a 100644 --- a/apps/controller/db/migrations/meta/_journal.json +++ b/apps/controller/db/migrations/meta/_journal.json @@ -106,6 +106,13 @@ "when": 1786010881404, "tag": "0014_damp_serpent_society", "breakpoints": true + }, + { + "idx": 15, + "version": "7", + "when": 1786598055402, + "tag": "0015_unusual_slipstream", + "breakpoints": true } ] } \ No newline at end of file diff --git a/apps/controller/db/runs.ts b/apps/controller/db/runs.ts index 48f214e..8827e6f 100644 --- a/apps/controller/db/runs.ts +++ b/apps/controller/db/runs.ts @@ -51,6 +51,7 @@ export interface CreateRunInput { thinkingLevel?: ThinkingLevel; modelConnectionId?: string | null; callbackToken: string; + surfaceRef?: { kind: string; payload: Record } | null; } export async function createRun(database: Database, input: CreateRunInput): Promise { diff --git a/apps/controller/db/schema.ts b/apps/controller/db/schema.ts index 4bcd5de..e5ec497 100644 --- a/apps/controller/db/schema.ts +++ b/apps/controller/db/schema.ts @@ -169,6 +169,15 @@ export const runs = pgTable( /** The normalized request/event, verbatim, for replay and diagnosis. */ trigger: jsonb("trigger").notNull().$type(), + /** + * Return address for an ingress-started run (ZEN-93). Null for dashboard runs. + * Shape is the controller-local SurfaceRef envelope until protocol consult. + */ + surfaceRef: jsonb("surface_ref").$type<{ + kind: string; + payload: Record; + } | null>(), + /** Resolved at creation so a run is reproducible even if config changes. */ model: text("model").notNull(), thinkingLevel: text("thinking_level").notNull().default("medium").$type(), diff --git a/apps/controller/e2e.live.test.ts b/apps/controller/e2e.live.test.ts index 6e351c8..1fc4727 100644 --- a/apps/controller/e2e.live.test.ts +++ b/apps/controller/e2e.live.test.ts @@ -12,6 +12,7 @@ import { createSessionWithRun, getSession, } from "./db/sessions"; +import { createWebhookRegistry } from "./integrations"; import { createLogger } from "./logger"; import { createReconciler, type Reconciler } from "./reconcile/loop"; import { createCredentialBroker } from "./secrets/broker"; @@ -161,6 +162,7 @@ function newReconciler(): Reconciler { database, broker: createCredentialBroker(config, database, log), log, + integrations: createWebhookRegistry(config), }); } diff --git a/apps/controller/http/api.integration.test.ts b/apps/controller/http/api.integration.test.ts index a39bb5b..d0b7e33 100644 --- a/apps/controller/http/api.integration.test.ts +++ b/apps/controller/http/api.integration.test.ts @@ -19,6 +19,7 @@ import { seedRun, seedTestUser, silentLogger, + testAppDeps, testConfig, withTestModel, } from "../test-support"; @@ -306,11 +307,7 @@ describe("dashboard support", () => { describe("model connections", () => { it("stores credentials per user and snapshots the selected connection on a run", async () => { - const secureApp = createApp({ - config: testConfig({ APP_AUTH_REQUIRED: "true" }), - database, - log: silentLogger(), - }); + const secureApp = createApp(testAppDeps(database, { APP_AUTH_REQUIRED: "true" })); const user = await upsertAppUser(database, { githubUserId: "model-user", login: "model-user", @@ -385,11 +382,7 @@ describe("model connections", () => { }); it("lets a resumed session explicitly switch away from a deleted connection", async () => { - const secureApp = createApp({ - config: testConfig({ APP_AUTH_REQUIRED: "true" }), - database, - log: silentLogger(), - }); + const secureApp = createApp(testAppDeps(database, { APP_AUTH_REQUIRED: "true" })); const user = await upsertAppUser(database, { githubUserId: "session-model-user", login: "session-model-user", diff --git a/apps/controller/http/app.ts b/apps/controller/http/app.ts index a5e4cd1..6baa5ea 100644 --- a/apps/controller/http/app.ts +++ b/apps/controller/http/app.ts @@ -12,6 +12,7 @@ import { pluginRoutes } from "./plugins"; import { runRoutes } from "./runs"; import { sessionRoutes } from "./sessions"; import { vcsRoutes } from "./vcs"; +import { webhookRoutes } from "./webhook"; const UNSAFE_METHODS = new Set(["POST", "PUT", "PATCH", "DELETE"]); @@ -33,6 +34,7 @@ export function createApp(deps: Deps): Hono { c.set("config", deps.config); c.set("database", deps.database); c.set("log", deps.log); + c.set("integrations", deps.integrations); c.set( "user", await getAppUserForSession( @@ -55,7 +57,8 @@ export function createApp(deps: Deps): Hono { const publicPath = c.req.path === "/healthz" || c.req.path.startsWith("/auth/") || - c.req.path.startsWith("/internal/"); + c.req.path.startsWith("/internal/") || + c.req.path.startsWith("/integrations/"); if (deps.config.auth.requireUser && !c.get("user") && !publicPath) { return c.json({ error: "authentication required" }, 401); } @@ -90,6 +93,7 @@ export function createApp(deps: Deps): Hono { app.route("/sessions", sessionRoutes()); app.route("/plugins", pluginRoutes()); app.route("/internal", internalRoutes()); + app.route("/integrations/webhook", webhookRoutes()); app.route("/llm", llmRoutes(new OAuthFlowManager(deps.database, deps.config))); app.route("/vcs", vcsRoutes()); app.route("/", metaRoutes()); diff --git a/apps/controller/http/browser.integration.test.ts b/apps/controller/http/browser.integration.test.ts index 6419779..b474b8a 100644 --- a/apps/controller/http/browser.integration.test.ts +++ b/apps/controller/http/browser.integration.test.ts @@ -12,7 +12,7 @@ import { resetTables, seedSession, setupTestDatabase, - silentLogger, + testAppDeps, testConfig, } from "../test-support"; import { createApp } from "./app"; @@ -22,7 +22,7 @@ let app: ReturnType; beforeAll(() => { database = setupTestDatabase(); - app = createApp({ config: testConfig(), database, log: silentLogger() }); + app = createApp(testAppDeps(database)); }); beforeEach(async () => resetTables(database)); afterAll(async () => closeDatabase(database)); @@ -43,11 +43,7 @@ describe("browser boundary", () => { }); it("requires a session and scopes runs to the signed-in user", async () => { - const secureApp = createApp({ - config: testConfig({ APP_AUTH_REQUIRED: "true" }), - database, - log: silentLogger(), - }); + const secureApp = createApp(testAppDeps(database, { APP_AUTH_REQUIRED: "true" })); const first = await upsertAppUser(database, { githubUserId: "github-1", login: "first", @@ -155,11 +151,7 @@ describe("browser boundary", () => { }); it("returns a validation error when a selected model is stale or foreign", async () => { - const secureApp = createApp({ - config: testConfig({ APP_AUTH_REQUIRED: "true" }), - database, - log: silentLogger(), - }); + const secureApp = createApp(testAppDeps(database, { APP_AUTH_REQUIRED: "true" })); const user = await upsertAppUser(database, { githubUserId: "stale-model-user", login: "stale-model-user", diff --git a/apps/controller/http/deps.ts b/apps/controller/http/deps.ts index 00a615b..e3988d2 100644 --- a/apps/controller/http/deps.ts +++ b/apps/controller/http/deps.ts @@ -1,6 +1,7 @@ import type { Config } from "../config"; import type { Database } from "../db/client"; import type { AppUserRow } from "../db/schema"; +import type { IntegrationRegistry } from "../integrations"; import type { Logger } from "../logger"; /** What every route needs. Passed in rather than imported so tests can substitute. */ @@ -8,6 +9,7 @@ export interface Deps { config: Config; database: Database; log: Logger; + integrations: IntegrationRegistry; } export type AppEnv = { Variables: Deps & { user: AppUserRow | null } }; diff --git a/apps/controller/http/internal.ts b/apps/controller/http/internal.ts index 68d7af8..9e48d7d 100644 --- a/apps/controller/http/internal.ts +++ b/apps/controller/http/internal.ts @@ -10,6 +10,7 @@ import type { Database } from "../db/client"; import { appendEvent, completeRun, getRunByCallbackToken } from "../db/runs"; import type { RunRow } from "../db/schema"; import { getSessionForRun, saveSessionCheckpoint } from "../db/sessions"; +import { reportRunLifecycle } from "../integrations"; import type { AppEnv } from "./deps"; /** @@ -55,28 +56,7 @@ export function internalRoutes(): Hono { const parsed = runStatusReportSchema.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) return c.json({ error: "unrecognized status" }, 422); - let { status, detail } = parsed.data; - const database = c.get("database"); - - const session = await getSessionForRun(database, run); - if (status === "done" && session && !session.agentCheckpoint) { - status = "error"; - detail = "the turn completed without a durable Pi session checkpoint"; - } - - // Recorded as an event first, so the reason survives even if the transition - // below loses a race with the reconciler. - await appendEvent(database, run.id, "status", { status, detail: detail ?? null }); - - // The agent's own completion is authoritative. Telemetry never implies it. - const applied = await completeRun( - database, - run.id, - status === "done" ? "succeeded" : "failed", - status === "done" ? null : (detail ?? "the agent reported an error"), - ); - - c.get("log").info("terminal status from sandbox", { runId: run.id, status, applied }); + await applySandboxStatus(c, run, parsed.data.status, parsed.data.detail); return c.json({ ok: true }); }); @@ -109,6 +89,34 @@ async function requireRun(c: Context, runId: string): Promise, + run: RunRow, + reported: "done" | "error", + reportedDetail: string | null | undefined, +): Promise { + let status = reported; + let detail = reportedDetail; + const database = c.get("database"); + const session = await getSessionForRun(database, run); + if (status === "done" && session && !session.agentCheckpoint) { + status = "error"; + detail = "the turn completed without a durable Pi session checkpoint"; + } + + // Recorded as an event first, so the reason survives even if the transition + // below loses a race with the reconciler. + await appendEvent(database, run.id, "status", { status, detail: detail ?? null }); + + const outcome = status === "done" ? "succeeded" : "failed"; + const error = status === "done" ? null : (detail ?? "the agent reported an error"); + const applied = await completeRun(database, run.id, outcome, error); + if (applied) { + await reportRunLifecycle(c.get("integrations"), c.get("log"), run, outcome, error); + } + c.get("log").info("terminal status from sandbox", { runId: run.id, status, applied }); +} + async function authenticate( database: Database, runId: string, diff --git a/apps/controller/http/runs.ts b/apps/controller/http/runs.ts index 8f033db..0dddc74 100644 --- a/apps/controller/http/runs.ts +++ b/apps/controller/http/runs.ts @@ -10,6 +10,7 @@ import { streamSSE } from "hono/streaming"; import type { Database } from "../db/client"; import { completeRun, createRun, getRun, listEvents, listRuns } from "../db/runs"; import type { RunRow } from "../db/schema"; +import { reportRunLifecycle } from "../integrations"; import { requireAuthenticatedUser, userOwns } from "./auth"; import type { AppEnv } from "./deps"; import { readManualRouteRequest } from "./manual"; @@ -99,6 +100,13 @@ export function runRoutes(): Hono { // for crashes and timeouts — cancelling is the same shape, so it reuses the // same path instead of duplicating teardown. await completeRun(database, runId, "cancelled", "cancelled by an operator"); + await reportRunLifecycle( + c.get("integrations"), + c.get("log"), + run, + "cancelled", + "cancelled by an operator", + ); c.get("log").info("run cancelled", { runId }); return c.json({ status: "cancelled" satisfies RunStatus }); }); diff --git a/apps/controller/http/webhook.integration.test.ts b/apps/controller/http/webhook.integration.test.ts new file mode 100644 index 0000000..e3c5923 --- /dev/null +++ b/apps/controller/http/webhook.integration.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, it } from "vitest"; +import type { Database } from "../db/client"; +import { getRun } from "../db/runs"; +import { + HttpCallbackReportSink, + IntegrationRegistry, + RestWebhookIngressAdapter, +} from "../integrations"; +import { bindTestDatabase, seedTestUser, silentLogger, testConfig } from "../test-support"; +import { createApp } from "./app"; + +let database: Database; + +bindTestDatabase((db) => { + database = db; +}); + +describe("REST webhook ingress", () => { + it("is absent until a bearer token is configured", async () => { + const seeded = await seedTestUser(database, testConfig()); + const app = createApp({ + config: testConfig({ WEBHOOK_USER_ID: seeded.userId }), + database, + log: silentLogger(), + integrations: new IntegrationRegistry(), + }); + expect((await app.request("/integrations/webhook", { method: "POST" })).status).toBe(404); + }); + + it("queues a run with a return address and POSTs a queued callback", async () => { + const config = testConfig(); + const seeded = await seedTestUser(database, config); + const calls: { url: string; body: string }[] = []; + const integrations = new IntegrationRegistry(); + integrations.registerAdapter(new RestWebhookIngressAdapter("hook-secret")); + integrations.registerSink( + new HttpCallbackReportSink(async (url, init) => { + calls.push({ url, body: init.body }); + return { ok: true, status: 200 }; + }), + ); + const app = createApp({ + config: testConfig({ + WEBHOOK_BEARER_TOKEN: "hook-secret", + WEBHOOK_USER_ID: seeded.userId, + }), + database, + log: silentLogger(), + integrations, + }); + + const unauthorized = await app.request("/integrations/webhook", { + method: "POST", + headers: { "content-type": "application/json", authorization: "Bearer nope" }, + body: JSON.stringify({ + repo: "acme/demo", + prompt: "hi", + callbackUrl: "https://cb.test/", + }), + }); + expect(unauthorized.status).toBe(401); + + const response = await app.request("/integrations/webhook", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer hook-secret", + }, + body: JSON.stringify({ + repo: "acme/demo", + prompt: "fix the flaky test", + callbackUrl: "https://ci.example/hooks/pi", + modelConnectionId: seeded.modelConnectionId, + modelId: "test-model", + thinkingLevel: "off", + }), + }); + expect(response.status).toBe(202); + const body = (await response.json()) as { id: string; status: string }; + expect(body.status).toBe("queued"); + + const run = await getRun(database, body.id); + expect(run?.surfaceRef).toEqual({ + kind: "rest_webhook", + payload: { callbackUrl: "https://ci.example/hooks/pi" }, + }); + expect(calls).toHaveLength(1); + expect(JSON.parse(calls[0]?.body ?? "{}")).toMatchObject({ + runId: body.id, + status: "queued", + terminal: false, + }); + }); +}); diff --git a/apps/controller/http/webhook.ts b/apps/controller/http/webhook.ts new file mode 100644 index 0000000..c58dd2c --- /dev/null +++ b/apps/controller/http/webhook.ts @@ -0,0 +1,172 @@ +import { randomBytes } from "node:crypto"; +import type { Trigger } from "@pi-cloud-agent/protocol"; +import { thinkingLevelSchema } from "@pi-cloud-agent/protocol"; +import type { Context } from "hono"; +import { Hono } from "hono"; +import { z } from "zod"; +import type { Config } from "../config"; +import type { Database } from "../db/client"; +import { createRun } from "../db/runs"; +import { + type IngressAccept, + REST_WEBHOOK_SURFACE_KIND, + reportRunLifecycle, +} from "../integrations"; +import { getVcsProvider } from "../vcs/connections"; +import type { AppEnv } from "./deps"; +import { resolveRequestedLlmModel } from "./model-selection"; + +const webhookModelSchema = z.object({ + modelConnectionId: z.string().uuid(), + modelId: z.string().min(1), + thinkingLevel: thinkingLevelSchema.default("medium"), +}); + +type WebhookModel = z.infer; + +/** + * Machine-to-machine ingress (ZEN-93). + * + * Auth is the webhook bearer token, not a browser session. Disabled when + * WEBHOOK_BEARER_TOKEN is unset. + */ +export function webhookRoutes(): Hono { + const app = new Hono(); + + app.post("/", async (c) => { + const ready = webhookReady(c); + if (ready instanceof Response) return ready; + const parsed = await parseWebhookPost(c, ready.token); + if (parsed instanceof Response) return parsed; + return enqueueWebhookRun(c, ready.actorId, parsed.accepted, parsed.models); + }); + + return app; +} + +function webhookReady(c: Context): { token: string; actorId: string } | Response { + const token = c.get("config").webhook.bearerToken; + if (!token) return c.json({ error: "not found" }, 404); + const actorId = c.get("config").webhook.userId; + if (!actorId) return c.json({ error: "webhook actor is not configured" }, 503); + if (!c.get("integrations").getAdapter(REST_WEBHOOK_SURFACE_KIND)) { + return c.json({ error: "not found" }, 404); + } + return { token, actorId }; +} + +async function parseWebhookPost( + c: Context, + token: string, +): Promise<{ accepted: IngressAccept; models: WebhookModel } | Response> { + const authorization = c.req.header("authorization"); + if (!bearerMatches(authorization, token)) { + return c.json({ error: "unauthorized" }, 401); + } + const adapter = c.get("integrations").getAdapter(REST_WEBHOOK_SURFACE_KIND); + if (!adapter) return c.json({ error: "not found" }, 404); + const body = await c.req.json().catch(() => null); + const accepted = await adapter.accept({ authorizationHeader: authorization, body }); + if (!accepted) return c.json({ error: "invalid request" }, 422); + const models = webhookModelSchema.safeParse(body); + if (!models.success) { + return c.json({ error: "invalid request", issues: models.error.issues }, 422); + } + return { accepted, models: models.data }; +} + +async function enqueueWebhookRun( + c: Context, + actorId: string, + accepted: IngressAccept, + models: WebhookModel, +): Promise { + const config = c.get("config"); + const selected = await resolveRequestedLlmModel( + c.get("database"), + config, + actorId, + models.modelConnectionId, + models.modelId, + models.thinkingLevel, + ); + if (!selected.ok) return c.json({ error: selected.error }, 422); + + const trigger = await webhookTrigger(c.get("database"), config, actorId, accepted.trigger); + if (trigger instanceof Response) return trigger; + + const run = await createRun(c.get("database"), { + userId: actorId, + provider: trigger.repo.provider, + repoFullName: `${trigger.repo.owner}/${trigger.repo.name}`, + trigger, + model: `${selected.model.provider}/${selected.model.name}`, + modelConnectionId: selected.model.connectionId, + thinkingLevel: models.thinkingLevel, + callbackToken: randomBytes(32).toString("hex"), + surfaceRef: accepted.surface, + }); + + await reportRunLifecycle(c.get("integrations"), c.get("log"), run, "queued"); + c.get("log").info("webhook run queued", { runId: run.id, repo: run.repoFullName }); + return c.json({ id: run.id, status: run.status }, 202); +} + +async function webhookTrigger( + database: Database, + config: Config, + actorId: string, + trigger: Trigger, +): Promise { + try { + const repo = await resolveWebhookRepo( + database, + config, + actorId, + trigger.repo.provider, + `${trigger.repo.owner}/${trigger.repo.name}`, + trigger.repo.headBranch, + ); + return { ...trigger, repo }; + } catch (error) { + return new Response( + JSON.stringify({ error: error instanceof Error ? error.message : String(error) }), + { status: 422, headers: { "content-type": "application/json" } }, + ); + } +} + +function bearerMatches(header: string | undefined, expected: string): boolean { + if (!header) return false; + const match = /^Bearer\s+(.+)$/i.exec(header.trim()); + return match?.[1] === expected; +} + +async function resolveWebhookRepo( + database: Database, + config: Config, + userId: string, + provider: string, + repoFullName: string, + branch: string, +) { + const vcs = await getVcsProvider(database, config, provider, userId); + const repository = await vcs.getRepository(repoFullName); + if (!repository) { + throw new Error("repository is not available through the connected identity"); + } + const resolvedBranch = + branch.trim() || (await vcs.getDefaultBranch(repoFullName).catch(() => null)) || ""; + return { + provider: repository.provider, + host: repository.host, + owner: repository.owner, + name: repository.name, + cloneUrl: repository.cloneUrl, + defaultBranch: resolvedBranch || repository.defaultBranch || "main", + baseSha: "", + headSha: "", + headBranch: resolvedBranch, + prNumber: null, + }; +} diff --git a/apps/controller/index.ts b/apps/controller/index.ts index 1164048..da97c31 100644 --- a/apps/controller/index.ts +++ b/apps/controller/index.ts @@ -2,6 +2,7 @@ import { serve } from "@hono/node-server"; import { getConfig } from "./config"; import { db } from "./db/client"; import { createApp } from "./http/app"; +import { createWebhookRegistry } from "./integrations"; import { createLogger } from "./logger"; import { createReconciler } from "./reconcile/loop"; import { createCredentialBroker } from "./secrets/broker"; @@ -19,7 +20,8 @@ const config = getConfig(); const log = createLogger("controller", { level: config.logLevel }); const database = db(); -const app = createApp({ config, database, log }); +const integrations = createWebhookRegistry(config); +const app = createApp({ config, database, log, integrations }); const reconciler = createReconciler({ config, database, @@ -29,6 +31,7 @@ const reconciler = createReconciler({ createLogger("secrets", { level: config.logLevel }), ), log: createLogger("reconciler", { level: config.logLevel }), + integrations, }); const server = serve({ fetch: app.fetch, port: config.port }, (info) => { diff --git a/apps/controller/integrations/factory.ts b/apps/controller/integrations/factory.ts new file mode 100644 index 0000000..aa1dede --- /dev/null +++ b/apps/controller/integrations/factory.ts @@ -0,0 +1,12 @@ +import type { Config } from "../config"; +import { IntegrationRegistry } from "./registry"; +import { HttpCallbackReportSink, RestWebhookIngressAdapter } from "./rest-webhook"; + +export function createWebhookRegistry(config: Config): IntegrationRegistry { + const registry = new IntegrationRegistry(); + const token = config.webhook.bearerToken; + if (!token) return registry; + registry.registerAdapter(new RestWebhookIngressAdapter(token)); + registry.registerSink(new HttpCallbackReportSink()); + return registry; +} diff --git a/apps/controller/integrations/github-issue-comment.test.ts b/apps/controller/integrations/github-issue-comment.test.ts new file mode 100644 index 0000000..5cf6776 --- /dev/null +++ b/apps/controller/integrations/github-issue-comment.test.ts @@ -0,0 +1,96 @@ +import { createHmac } from "node:crypto"; +import { describe, expect, it } from "vitest"; +import { GITHUB_ISSUE_SURFACE_KIND, GitHubIssueCommentAdapter } from "./index"; + +const SECRET = "github-webhook-secret"; +const BOT = "pi-cloud-bot"; + +function sign(rawBody: string): string { + const digest = createHmac("sha256", SECRET).update(rawBody).digest("hex"); + return `sha256=${digest}`; +} + +function issueCommentPayload( + overrides: { + action?: string; + commentBody?: string; + commentLogin?: string; + issueBody?: string | null; + pullRequest?: boolean; + } = {}, +): string { + const issue: Record = { + number: 42, + title: "Flaky tests in CI", + body: overrides.issueBody === undefined ? "The suite fails on main." : overrides.issueBody, + }; + if (overrides.pullRequest) + issue.pull_request = { url: "https://api.github.com/repos/acme/demo/pulls/42" }; + return JSON.stringify({ + action: overrides.action ?? "created", + issue, + comment: { + body: overrides.commentBody ?? `@${BOT} please fix the flaky tests`, + user: { login: overrides.commentLogin ?? "alice" }, + }, + repository: { + name: "demo", + clone_url: "https://github.com/acme/demo.git", + default_branch: "main", + owner: { login: "acme" }, + }, + }); +} + +function ingress(rawBody: string, headers: { event?: string; signature?: string | null } = {}) { + return { + eventHeader: headers.event ?? "issue_comment", + signatureHeader: headers.signature === undefined ? sign(rawBody) : headers.signature, + rawBody, + }; +} + +describe("ZEN-94 1a GitHub issue comment adapter", () => { + const adapter = new GitHubIssueCommentAdapter(SECRET, BOT); + + it("accepts a signed issue comment that @mentions the bot", async () => { + const rawBody = issueCommentPayload(); + const accepted = await adapter.accept(ingress(rawBody)); + expect(accepted).not.toBeNull(); + expect(accepted?.taskSpec.repo.owner).toBe("acme"); + expect(accepted?.taskSpec.repo.name).toBe("demo"); + expect(accepted?.taskSpec.prompt).toContain("Flaky tests in CI"); + expect(accepted?.taskSpec.prompt).toContain("@pi-cloud-bot please fix"); + expect(accepted?.trigger.kind).toBe("manual"); + expect(accepted?.surface).toEqual({ + kind: GITHUB_ISSUE_SURFACE_KIND, + payload: { owner: "acme", repo: "demo", issueNumber: 42 }, + }); + }); + + it("rejects a bad or missing HMAC", async () => { + const rawBody = issueCommentPayload(); + expect(await adapter.accept(ingress(rawBody, { signature: "sha256=deadbeef" }))).toBeNull(); + expect(await adapter.accept(ingress(rawBody, { signature: null }))).toBeNull(); + }); + + it("ignores other GitHub events, non-created actions, and PR issue threads", async () => { + const mention = issueCommentPayload(); + expect(await adapter.accept(ingress(mention, { event: "push" }))).toBeNull(); + expect(await adapter.accept(ingress(issueCommentPayload({ action: "edited" })))).toBeNull(); + expect( + await adapter.accept(ingress(issueCommentPayload({ pullRequest: true }))), + ).toBeNull(); + }); + + it("ignores comments that do not mention the bot, and the bot's own comments", async () => { + expect( + await adapter.accept(ingress(issueCommentPayload({ commentBody: "anyone there?" }))), + ).toBeNull(); + expect( + await adapter.accept( + ingress(issueCommentPayload({ commentLogin: BOT, commentBody: `@${BOT} looping` })), + ), + ).toBeNull(); + }); +}); diff --git a/apps/controller/integrations/github-issue-comment.ts b/apps/controller/integrations/github-issue-comment.ts new file mode 100644 index 0000000..c744235 --- /dev/null +++ b/apps/controller/integrations/github-issue-comment.ts @@ -0,0 +1,158 @@ +import { createHmac, timingSafeEqual } from "node:crypto"; +import type { RepoRef, TaskSpec, Trigger } from "@pi-cloud-agent/protocol"; +import { z } from "zod"; +import type { IngressAdapter } from "./ingress"; +import type { IngressAccept, SurfaceRef } from "./types"; +import { surfaceRefSchema } from "./types"; + +/** Issue-thread surface (ZEN-94 phase 1a). Not PR review comments. */ +export const GITHUB_ISSUE_SURFACE_KIND = "github_issue" as const; + +const PROMPT_MAX_CHARS = 32_000; + +const issueCommentPayloadSchema = z.object({ + action: z.string(), + issue: z.object({ + number: z.number().int().positive(), + title: z.string(), + body: z.string().nullable(), + pull_request: z.unknown().optional(), + }), + comment: z.object({ + body: z.string(), + user: z.object({ login: z.string() }), + }), + repository: z.object({ + name: z.string().min(1), + clone_url: z.string().url(), + default_branch: z.string().optional(), + owner: z.object({ login: z.string().min(1) }), + }), +}); + +interface GitHubWebhookIngressInput { + eventHeader: string | null | undefined; + signatureHeader: string | null | undefined; + rawBody: string; +} + +/** + * GitHub `issue_comment` → TaskSpec + SurfaceRef. + * + * HMAC is GitHub's `X-Hub-Signature-256`. Mentions are `@botLogin`. + * PR conversation comments (`issue.pull_request` set) wait for a later phase. + */ +export class GitHubIssueCommentAdapter implements IngressAdapter { + readonly kind = GITHUB_ISSUE_SURFACE_KIND; + + constructor( + private readonly webhookSecret: string, + private readonly botLogin: string, + ) {} + + async accept(input: unknown): Promise { + if (!isGitHubWebhookIngressInput(input)) return null; + if (input.eventHeader !== "issue_comment") return null; + if (!verifyGitHubSignature(input.rawBody, input.signatureHeader, this.webhookSecret)) { + return null; + } + + const payload = parseIssueCommentPayload(input.rawBody); + if (!payload) return null; + if (!shouldHandleIssueComment(payload, this.botLogin)) return null; + + const repo = repoFromPayload(payload); + const prompt = buildIssueMentionPrompt(payload); + const trigger: Trigger = { kind: "manual", repo, prompt }; + const taskSpec: TaskSpec = { prompt, repo }; + const surface: SurfaceRef = surfaceRefSchema.parse({ + kind: GITHUB_ISSUE_SURFACE_KIND, + payload: { + owner: payload.repository.owner.login, + repo: payload.repository.name, + issueNumber: payload.issue.number, + }, + }); + return { trigger, taskSpec, surface }; + } +} + +function isGitHubWebhookIngressInput(input: unknown): input is GitHubWebhookIngressInput { + if (input === null || typeof input !== "object") return false; + const value = input as Record; + return ( + typeof value.rawBody === "string" && "eventHeader" in value && "signatureHeader" in value + ); +} + +function parseIssueCommentPayload(rawBody: string) { + let json: unknown; + try { + json = JSON.parse(rawBody); + } catch { + return null; + } + const parsed = issueCommentPayloadSchema.safeParse(json); + return parsed.success ? parsed.data : null; +} + +function shouldHandleIssueComment( + payload: z.infer, + botLogin: string, +): boolean { + if (payload.action !== "created") return false; + if ("pull_request" in payload.issue && payload.issue.pull_request !== undefined) { + return false; + } + if (payload.comment.user.login.toLowerCase() === botLogin.toLowerCase()) return false; + return commentMentionsBot(payload.comment.body, botLogin); +} + +function commentMentionsBot(body: string, botLogin: string): boolean { + const escaped = botLogin.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); + return new RegExp(`(^|\\W)@${escaped}(?=$|\\W)`, "i").test(body); +} + +function repoFromPayload(payload: z.infer): RepoRef { + const owner = payload.repository.owner.login; + const name = payload.repository.name; + const branch = payload.repository.default_branch ?? "main"; + return { + provider: "github", + host: "github.com", + owner, + name, + cloneUrl: payload.repository.clone_url, + defaultBranch: branch, + baseSha: "", + headSha: "", + headBranch: branch, + prNumber: null, + }; +} + +function buildIssueMentionPrompt(payload: z.infer): string { + const issueBody = payload.issue.body?.trim() || "(no issue body)"; + const text = [ + `GitHub issue #${payload.issue.number}: ${payload.issue.title}`, + "", + issueBody, + "", + `Comment from @${payload.comment.user.login}:`, + payload.comment.body, + ].join("\n"); + if (text.length <= PROMPT_MAX_CHARS) return text; + return `${text.slice(0, PROMPT_MAX_CHARS)}\n…(truncated)`; +} + +function verifyGitHubSignature( + rawBody: string, + signatureHeader: string | null | undefined, + secret: string, +): boolean { + if (!signatureHeader?.startsWith("sha256=")) return false; + const actual = signatureHeader.slice("sha256=".length); + const expected = createHmac("sha256", secret).update(rawBody).digest("hex"); + if (actual.length !== expected.length) return false; + return timingSafeEqual(Buffer.from(actual, "utf8"), Buffer.from(expected, "utf8")); +} diff --git a/apps/controller/integrations/index.ts b/apps/controller/integrations/index.ts new file mode 100644 index 0000000..2fa833f --- /dev/null +++ b/apps/controller/integrations/index.ts @@ -0,0 +1,18 @@ +export { createWebhookRegistry } from "./factory"; +export { + GITHUB_ISSUE_SURFACE_KIND, + GitHubIssueCommentAdapter, +} from "./github-issue-comment"; +export { + MEMORY_SURFACE_KIND, + MemoryIngressAdapter, + MemoryReportSink, +} from "./memory"; +export { reportRunLifecycle } from "./notify"; +export { IntegrationRegistry } from "./registry"; +export { + HttpCallbackReportSink, + REST_WEBHOOK_SURFACE_KIND, + RestWebhookIngressAdapter, +} from "./rest-webhook"; +export type { IngressAccept } from "./types"; diff --git a/apps/controller/integrations/ingress.ts b/apps/controller/integrations/ingress.ts new file mode 100644 index 0000000..450b773 --- /dev/null +++ b/apps/controller/integrations/ingress.ts @@ -0,0 +1,20 @@ +import type { IngressAccept } from "./types"; + +/** + * One inbound surface (REST webhook, GitHub, Teams, …). + * + * `accept` owns verification (signature / token) and normalization. + * Auth helpers stay per-adapter until a shared pattern emerges. + * + * TODO(BA): decide whether a shared HMAC/signature helper belongs in the + * controller or stays adapter-local. + */ +export interface IngressAdapter { + readonly kind: string; + + /** + * Verify and normalize a foreign request. + * Return null when this adapter does not handle the payload (or rejects it). + */ + accept(input: unknown): Promise; +} diff --git a/apps/controller/integrations/memory.test.ts b/apps/controller/integrations/memory.test.ts new file mode 100644 index 0000000..4f761e9 --- /dev/null +++ b/apps/controller/integrations/memory.test.ts @@ -0,0 +1,93 @@ +import { isTerminal } from "@pi-cloud-agent/protocol"; +import { describe, expect, it } from "vitest"; +import { + IntegrationRegistry, + MEMORY_SURFACE_KIND, + MemoryIngressAdapter, + MemoryReportSink, +} from "./index"; + +const sampleRepo = { + provider: "github", + host: "github.com", + owner: "acme", + name: "demo", + cloneUrl: "https://github.com/acme/demo.git", + defaultBranch: "main", + baseSha: "", + headSha: "", + headBranch: "main", + prNumber: null, +}; + +describe("ZEN-92 integration substrate", () => { + it("rejects a bad secret and accepts a verified payload into core shapes", async () => { + const adapter = new MemoryIngressAdapter("correct-secret"); + + expect( + await adapter.accept({ secret: "wrong", prompt: "hi", repo: sampleRepo }), + ).toBeNull(); + + const accepted = await adapter.accept({ + secret: "correct-secret", + prompt: "fix the flaky test", + repo: sampleRepo, + surfacePayload: { callbackId: "cb-1" }, + }); + + expect(accepted).not.toBeNull(); + expect(accepted?.taskSpec.prompt).toBe("fix the flaky test"); + expect(accepted?.trigger.kind).toBe("manual"); + expect(accepted?.surface).toEqual({ + kind: MEMORY_SURFACE_KIND, + payload: { callbackId: "cb-1" }, + }); + }); + + it("resolves a sink by surface kind and records lifecycle reports only", async () => { + const registry = new IntegrationRegistry(); + const adapter = new MemoryIngressAdapter("s"); + const sink = new MemoryReportSink(); + registry.registerAdapter(adapter); + registry.registerSink(sink); + + const accepted = await registry.getAdapter(MEMORY_SURFACE_KIND)?.accept({ + secret: "s", + prompt: "ship it", + repo: sampleRepo, + }); + expect(accepted).not.toBeNull(); + if (!accepted) return; + + const resolved = registry.resolveSink(accepted.surface); + expect(resolved).toBe(sink); + + // Lifecycle path we will hook later — not agent findings. + await resolved?.report(accepted.surface, { + runId: "run_1", + status: "running", + terminal: false, + }); + await resolved?.report(accepted.surface, { + runId: "run_1", + status: "succeeded", + detail: null, + terminal: isTerminal("succeeded"), + }); + + expect(sink.reports).toHaveLength(2); + expect(sink.reports[1]?.terminal).toBe(true); + expect(sink.reports.every((r) => r.status !== undefined)).toBe(true); + }); + + it("refuses duplicate adapter or sink registration", () => { + const registry = new IntegrationRegistry(); + registry.registerAdapter(new MemoryIngressAdapter("a")); + registry.registerSink(new MemoryReportSink()); + + expect(() => registry.registerAdapter(new MemoryIngressAdapter("b"))).toThrow( + /already registered/, + ); + expect(() => registry.registerSink(new MemoryReportSink())).toThrow(/already registered/); + }); +}); diff --git a/apps/controller/integrations/memory.ts b/apps/controller/integrations/memory.ts new file mode 100644 index 0000000..9d902f4 --- /dev/null +++ b/apps/controller/integrations/memory.ts @@ -0,0 +1,70 @@ +import type { RepoRef, TaskSpec, Trigger } from "@pi-cloud-agent/protocol"; +import type { IngressAdapter } from "./ingress"; +import type { ReportSink } from "./report"; +import type { IngressAccept, SurfaceRef, SurfaceReport } from "./types"; +import { surfaceRefSchema } from "./types"; + +/** Test / design-double surface kind. Not a product integration. */ +export const MEMORY_SURFACE_KIND = "memory" as const; + +interface MemoryIngressInput { + secret: string; + prompt: string; + repo: RepoRef; + /** Optional opaque payload stored on the SurfaceRef for wake/callback tests. */ + surfacePayload?: Record; +} + +/** + * In-memory ingress + sink for proving the ZEN-92 round-trip without forges. + * Product adapters (REST / GitHub / Teams) land in ZEN-93–95. + */ +export class MemoryIngressAdapter implements IngressAdapter { + readonly kind = MEMORY_SURFACE_KIND; + + constructor(private readonly expectedSecret: string) {} + + async accept(input: unknown): Promise { + if (!isMemoryIngressInput(input)) return null; + if (input.secret !== this.expectedSecret) return null; + + const trigger: Trigger = { + kind: "manual", + repo: input.repo, + prompt: input.prompt, + }; + const taskSpec: TaskSpec = { + prompt: input.prompt, + repo: input.repo, + }; + const surface: SurfaceRef = surfaceRefSchema.parse({ + kind: MEMORY_SURFACE_KIND, + payload: input.surfacePayload ?? {}, + }); + return { trigger, taskSpec, surface }; + } +} + +export class MemoryReportSink implements ReportSink { + readonly kind = MEMORY_SURFACE_KIND; + readonly reports: SurfaceReport[] = []; + + supports(surface: SurfaceRef): boolean { + return surface.kind === MEMORY_SURFACE_KIND; + } + + async report(_surface: SurfaceRef, report: SurfaceReport): Promise { + this.reports.push(report); + } +} + +function isMemoryIngressInput(input: unknown): input is MemoryIngressInput { + if (input === null || typeof input !== "object") return false; + const value = input as Record; + return ( + typeof value.secret === "string" && + typeof value.prompt === "string" && + value.repo !== null && + typeof value.repo === "object" + ); +} diff --git a/apps/controller/integrations/notify.ts b/apps/controller/integrations/notify.ts new file mode 100644 index 0000000..bb3a9f0 --- /dev/null +++ b/apps/controller/integrations/notify.ts @@ -0,0 +1,29 @@ +import { isTerminal, type RunStatus } from "@pi-cloud-agent/protocol"; +import type { RunRow } from "../db/schema"; +import type { Logger } from "../logger"; +import type { IntegrationRegistry } from "./registry"; + +/** + * Best-effort lifecycle ping. Never throws — a dead callback must not fail the run. + */ +export async function reportRunLifecycle( + registry: IntegrationRegistry, + log: Logger, + run: RunRow, + status: RunStatus, + detail?: string | null, +): Promise { + if (!run.surfaceRef) return; + const sink = registry.resolveSink(run.surfaceRef); + if (!sink) return; + try { + await sink.report(run.surfaceRef, { + runId: run.id, + status, + detail: detail ?? run.error, + terminal: isTerminal(status), + }); + } catch (error) { + log.warn("surface report failed", { runId: run.id, status, error }); + } +} diff --git a/apps/controller/integrations/registry.ts b/apps/controller/integrations/registry.ts new file mode 100644 index 0000000..c1ad4ba --- /dev/null +++ b/apps/controller/integrations/registry.ts @@ -0,0 +1,42 @@ +import type { IngressAdapter } from "./ingress"; +import type { ReportSink } from "./report"; +import type { SurfaceRef } from "./types"; + +/** + * In-process registry of ingress adapters and report sinks. + * + * TODO(BA): wire into HTTP dispatch + reconciler/internal status once + * `SurfaceRef` persistence on the run row is approved (schema migration). + */ +export class IntegrationRegistry { + private readonly adapters = new Map(); + private readonly sinks = new Map(); + + registerAdapter(adapter: IngressAdapter): void { + if (this.adapters.has(adapter.kind)) { + throw new Error(`ingress adapter already registered: ${adapter.kind}`); + } + this.adapters.set(adapter.kind, adapter); + } + + registerSink(sink: ReportSink): void { + if (this.sinks.has(sink.kind)) { + throw new Error(`report sink already registered: ${sink.kind}`); + } + this.sinks.set(sink.kind, sink); + } + + getAdapter(kind: string): IngressAdapter | null { + return this.adapters.get(kind) ?? null; + } + + /** Prefer a sink whose kind matches the surface; else the first that supports it. */ + resolveSink(surface: SurfaceRef): ReportSink | null { + const byKind = this.sinks.get(surface.kind); + if (byKind?.supports(surface)) return byKind; + for (const sink of this.sinks.values()) { + if (sink.supports(surface)) return sink; + } + return null; + } +} diff --git a/apps/controller/integrations/report.ts b/apps/controller/integrations/report.ts new file mode 100644 index 0000000..03acb00 --- /dev/null +++ b/apps/controller/integrations/report.ts @@ -0,0 +1,17 @@ +import type { SurfaceRef, SurfaceReport } from "./types"; + +/** + * One outbound surface for lifecycle reports. + * + * TODO(BA): confirm sinks never post semantic agent outcomes — only lifecycle. + * TODO(BA): decide progress source — status transitions (current) vs full + * `run_events` stream vs terminal-only. + */ +export interface ReportSink { + readonly kind: string; + + /** True when this sink can report for the given surface. */ + supports(surface: SurfaceRef): boolean; + + report(surface: SurfaceRef, report: SurfaceReport): Promise; +} diff --git a/apps/controller/integrations/rest-webhook.test.ts b/apps/controller/integrations/rest-webhook.test.ts new file mode 100644 index 0000000..3f6d66b --- /dev/null +++ b/apps/controller/integrations/rest-webhook.test.ts @@ -0,0 +1,101 @@ +import { isTerminal } from "@pi-cloud-agent/protocol"; +import { describe, expect, it } from "vitest"; +import { + HttpCallbackReportSink, + IntegrationRegistry, + REST_WEBHOOK_SURFACE_KIND, + RestWebhookIngressAdapter, +} from "./index"; + +describe("ZEN-93 REST webhook scaffold", () => { + const adapter = new RestWebhookIngressAdapter("test-token"); + + it("rejects missing/wrong bearer and accepts a valid JSON body", async () => { + const body = { + provider: "github", + repo: "acme/demo", + prompt: "fix flaky tests", + callbackUrl: "https://example.com/hooks/pi", + branch: "develop", + }; + + expect(await adapter.accept({ authorizationHeader: null, body })).toBeNull(); + expect( + await adapter.accept({ + authorizationHeader: "Bearer wrong", + body, + }), + ).toBeNull(); + expect( + await adapter.accept({ + authorizationHeader: "Bearer test-token", + body: { ...body, prompt: "" }, + }), + ).toBeNull(); + + const accepted = await adapter.accept({ + authorizationHeader: "Bearer test-token", + body, + }); + expect(accepted).not.toBeNull(); + expect(accepted?.trigger.kind).toBe("manual"); + expect(accepted?.taskSpec.prompt).toBe("fix flaky tests"); + expect(accepted?.taskSpec.repo.owner).toBe("acme"); + expect(accepted?.taskSpec.repo.headBranch).toBe("develop"); + expect(accepted?.surface).toEqual({ + kind: REST_WEBHOOK_SURFACE_KIND, + payload: { callbackUrl: "https://example.com/hooks/pi" }, + }); + }); + + it("POSTs lifecycle callbacks and does not throw when the callback fails", async () => { + const calls: { url: string; body: string }[] = []; + const sink = new HttpCallbackReportSink(async (url, init) => { + calls.push({ url, body: init.body }); + return { ok: false, status: 503 }; + }); + const registry = new IntegrationRegistry(); + registry.registerAdapter(adapter); + registry.registerSink(sink); + + const accepted = await adapter.accept({ + authorizationHeader: "Bearer test-token", + body: { + repo: "acme/demo", + prompt: "ship it", + callbackUrl: "https://example.com/cb", + }, + }); + expect(accepted).not.toBeNull(); + if (!accepted) return; + + const resolved = registry.resolveSink(accepted.surface); + expect(resolved).toBe(sink); + + await expect( + resolved?.report(accepted.surface, { + runId: "run_1", + status: "running", + terminal: false, + }), + ).resolves.toBeUndefined(); + await expect( + resolved?.report(accepted.surface, { + runId: "run_1", + status: "succeeded", + detail: null, + terminal: isTerminal("succeeded"), + }), + ).resolves.toBeUndefined(); + + expect(calls).toHaveLength(2); + expect(calls[0]?.url).toBe("https://example.com/cb"); + expect(JSON.parse(calls[1]?.body ?? "{}")).toEqual({ + runId: "run_1", + status: "succeeded", + terminal: true, + detail: null, + }); + expect(sink.failures).toEqual(["callback HTTP 503", "callback HTTP 503"]); + }); +}); diff --git a/apps/controller/integrations/rest-webhook.ts b/apps/controller/integrations/rest-webhook.ts new file mode 100644 index 0000000..a4c102f --- /dev/null +++ b/apps/controller/integrations/rest-webhook.ts @@ -0,0 +1,168 @@ +import type { RepoRef, TaskSpec, Trigger } from "@pi-cloud-agent/protocol"; +import { z } from "zod"; +import type { IngressAdapter } from "./ingress"; +import type { ReportSink } from "./report"; +import type { IngressAccept, SurfaceRef, SurfaceReport } from "./types"; +import { surfaceRefSchema } from "./types"; + +/** First real surface kind (ZEN-93). Not a forge-specific integration. */ +export const REST_WEBHOOK_SURFACE_KIND = "rest_webhook" as const; + +const restWebhookBodySchema = z.object({ + provider: z.string().min(1).default("github"), + /** Provider full name, e.g. `acme/demo`. */ + repo: z.string().min(3), + prompt: z.string().min(1), + callbackUrl: z.string().url(), + branch: z.string().min(1).optional(), + wallClockSeconds: z.number().int().positive().optional(), +}); + +/** + * What the future HTTP route passes into `accept`. + * Keep headers out of the body so auth stays explicit. + */ +interface RestWebhookIngressInput { + authorizationHeader: string | null | undefined; + body: unknown; +} + +type FetchLike = ( + url: string, + init: { method: string; headers: Record; body: string }, +) => Promise<{ ok: boolean; status: number }>; + +/** + * Authenticated JSON ingress for CI/cron/internal tools. + * + * The HTTP route resolves `repo` via VCS and loads the bearer token from config. + */ +export class RestWebhookIngressAdapter implements IngressAdapter { + readonly kind = REST_WEBHOOK_SURFACE_KIND; + + constructor(private readonly bearerToken: string) {} + + async accept(input: unknown): Promise { + if (!isRestWebhookIngressInput(input)) return null; + if (!bearerMatches(input.authorizationHeader, this.bearerToken)) return null; + + const parsed = restWebhookBodySchema.safeParse(input.body); + if (!parsed.success) return null; + + const body = parsed.data; + const repo = provisionalRepoRef(body.provider, body.repo, body.branch); + if (!repo) return null; + + const trigger: Trigger = { kind: "manual", repo, prompt: body.prompt }; + const taskSpec: TaskSpec = { + prompt: body.prompt, + repo, + wallClockSeconds: body.wallClockSeconds, + }; + const surface: SurfaceRef = surfaceRefSchema.parse({ + kind: REST_WEBHOOK_SURFACE_KIND, + payload: { callbackUrl: body.callbackUrl }, + }); + return { trigger, taskSpec, surface }; + } +} + +/** + * POSTs lifecycle reports to the caller-provided callback URL. + * + * Failures are swallowed so a dead callback cannot fail the run + * (ZEN-93 acceptance). Retries / signing / allowlists are TODO. + */ +export class HttpCallbackReportSink implements ReportSink { + readonly kind = REST_WEBHOOK_SURFACE_KIND; + /** Soft-failure log for tests / future metrics. */ + readonly failures: string[] = []; + + constructor(private readonly fetchImpl: FetchLike = defaultFetch) {} + + supports(surface: SurfaceRef): boolean { + return surface.kind === REST_WEBHOOK_SURFACE_KIND; + } + + async report(surface: SurfaceRef, report: SurfaceReport): Promise { + const callbackUrl = surface.payload.callbackUrl; + if (typeof callbackUrl !== "string" || callbackUrl.length === 0) { + this.failures.push("missing callbackUrl"); + return; + } + + const payload = { + runId: report.runId, + status: report.status, + terminal: report.terminal, + detail: report.detail ?? null, + }; + + try { + const response = await this.fetchImpl(callbackUrl, { + method: "POST", + headers: { "content-type": "application/json", accept: "application/json" }, + body: JSON.stringify(payload), + }); + if (!response.ok) { + this.failures.push(`callback HTTP ${response.status}`); + } + } catch (error) { + this.failures.push(error instanceof Error ? error.message : String(error)); + } + } +} + +function bearerMatches( + authorizationHeader: string | null | undefined, + expectedToken: string, +): boolean { + if (!authorizationHeader) return false; + const match = /^Bearer\s+(.+)$/i.exec(authorizationHeader.trim()); + return match?.[1] === expectedToken; +} + +function isRestWebhookIngressInput(input: unknown): input is RestWebhookIngressInput { + if (input === null || typeof input !== "object") return false; + return "body" in input && "authorizationHeader" in input; +} + +/** + * Best-effort RepoRef so TaskSpec validates in the scaffold. + * Production must replace this with `vcs.getRepository` + branch resolution. + */ +function provisionalRepoRef( + provider: string, + repoFullName: string, + branch?: string, +): RepoRef | null { + const [owner, name, ...rest] = repoFullName.split("/"); + if (!owner || !name || rest.length > 0) return null; + const headBranch = branch ?? "main"; + // TODO(wire): host/cloneUrl must come from the connected VCS provider. + const host = provider === "azure-devops" ? "dev.azure.com" : "github.com"; + const cloneUrl = + provider === "azure-devops" + ? `https://dev.azure.com/${owner}/_git/${name}` + : `https://github.com/${owner}/${name}.git`; + return { + provider, + host, + owner, + name, + cloneUrl, + defaultBranch: headBranch, + baseSha: "", + headSha: "", + headBranch, + prNumber: null, + }; +} + +async function defaultFetch( + url: string, + init: { method: string; headers: Record; body: string }, +): Promise<{ ok: boolean; status: number }> { + const response = await fetch(url, init); + return { ok: response.ok, status: response.status }; +} diff --git a/apps/controller/integrations/types.ts b/apps/controller/integrations/types.ts new file mode 100644 index 0000000..9db4765 --- /dev/null +++ b/apps/controller/integrations/types.ts @@ -0,0 +1,47 @@ +/** + * Integration substrate for ZEN-92 (ingress + report-back). + * + * Provisional home: the controller. Do not import this from `packages/runtime`. + * + * TODO(BA/maintainer): after confirmation, promote these contracts to + * `packages/protocol` (or a dedicated package) — AGENTS.md requires consult + * before widening protocol. + */ + +import type { RunStatus, TaskSpec, Trigger } from "@pi-cloud-agent/protocol"; +import { z } from "zod"; + +/** + * Where progress/final reports should go for a run started by an ingress. + * + * TODO(BA): tighten to a discriminated union per surface + * (`webhook` | `github_issue` | `teams_conversation` | …) once kinds are fixed. + */ +export const surfaceRefSchema = z.object({ + kind: z.string().min(1), + payload: z.record(z.string(), z.unknown()), +}); + +export type SurfaceRef = z.infer; + +/** Successful normalize step: core shapes only — no surface-specific leftovers. */ +export interface IngressAccept { + trigger: Trigger; + taskSpec: TaskSpec; + surface: SurfaceRef; +} + +/** + * Lifecycle-only report. Intentionally excludes agent findings / PR text. + * + * TODO(BA): confirm ReportSink stays lifecycle-only (VISION: no controller-side + * publish of semantic outcomes). If semantic posting is allowed later, extend + * this type deliberately — do not overload `detail` with parsed agent output. + */ +export interface SurfaceReport { + runId: string; + status: RunStatus; + /** Human-safe status detail from the controller (e.g. failure reason), never agent findings. */ + detail?: string | null; + terminal: boolean; +} diff --git a/apps/controller/reconcile/loop.ts b/apps/controller/reconcile/loop.ts index 73d3c0d..617ef2a 100644 --- a/apps/controller/reconcile/loop.ts +++ b/apps/controller/reconcile/loop.ts @@ -20,6 +20,8 @@ import { getSession, parkSession, } from "../db/sessions"; +import type { IntegrationRegistry } from "../integrations"; +import { reportRunLifecycle } from "../integrations"; import type { Logger } from "../logger"; import type { CredentialBroker } from "../secrets/broker"; import { type ProvisionDeps, provisionRun } from "./provision"; @@ -59,6 +61,7 @@ export interface ReconcilerOptions { maxConcurrentProvisions?: number; /** Fallback poll interval. NOTIFY makes the common case immediate. */ pollIntervalMs?: number; + integrations: IntegrationRegistry; } export interface Reconciler { @@ -90,10 +93,18 @@ export function createReconciler(options: ReconcilerOptions): Reconciler { maxConcurrentProvisions = 4, pollIntervalMs = 2000, createProvider = (name: string) => createSandboxProvider(name, config.env), + integrations, } = options; const sandbox = createProvider(config.sandbox.provider); - const provisionDeps: ProvisionDeps = { config, database, broker, sandbox, log }; + const provisionDeps: ProvisionDeps = { + config, + database, + broker, + sandbox, + log, + integrations, + }; let running = false; let timer: NodeJS.Timeout | null = null; @@ -198,6 +209,7 @@ export function createReconciler(options: ReconcilerOptions): Reconciler { if (!run) return; claimed.add(run.id); available -= 1; + await reportRunLifecycle(integrations, log, run, "provisioning"); // Detached on purpose: provisioning talks to a sandbox API and must not // hold up timeouts or teardown for other runs. `drain` waits for these. const pending = provisionRun(run, provisionDeps) @@ -217,7 +229,10 @@ export function createReconciler(options: ReconcilerOptions): Reconciler { ): Promise { for (const run of runs) { const changed = await completeRun(database, run.id, "failed", message); - if (changed) log.warn("run failed by the reconciler", { runId: run.id, reason }); + if (changed) { + log.warn("run failed by the reconciler", { runId: run.id, reason }); + await reportRunLifecycle(integrations, log, run, "failed", message); + } await park(run, reason); } } diff --git a/apps/controller/reconcile/provision.ts b/apps/controller/reconcile/provision.ts index 6e1463e..2160235 100644 --- a/apps/controller/reconcile/provision.ts +++ b/apps/controller/reconcile/provision.ts @@ -20,6 +20,8 @@ import { } from "../db/runs"; import type { RunRow } from "../db/schema"; import { clearSessionWorkspace, getSessionForRun } from "../db/sessions"; +import type { IntegrationRegistry } from "../integrations"; +import { reportRunLifecycle } from "../integrations"; import type { Logger } from "../logger"; import { buildTaskPrompt, resolvePluginsForRun } from "../plugins/catalog"; import type { CredentialBroker } from "../secrets/broker"; @@ -42,6 +44,7 @@ export interface ProvisionDeps { broker: CredentialBroker; sandbox: SandboxProvider; log: Logger; + integrations: IntegrationRegistry; } /** How many times a retryable provisioning failure is worth another attempt. */ @@ -122,6 +125,7 @@ export async function provisionRun(run: RunRow, deps: ProvisionDeps): Promise provider, + integrations: createWebhookRegistry(testConfig()), ...options, }); } @@ -429,6 +431,7 @@ describe("concurrency", () => { log: silentLogger(), createProvider: () => provider, maxConcurrentProvisions: 2, + integrations: createWebhookRegistry(testConfig()), }); await tick(loop); diff --git a/apps/controller/test-support.ts b/apps/controller/test-support.ts index 4265ef5..6cf5b59 100644 --- a/apps/controller/test-support.ts +++ b/apps/controller/test-support.ts @@ -9,6 +9,7 @@ import { createRun } from "./db/runs"; import type { RunRow, SessionRow } from "./db/schema"; import { createSessionWithRun } from "./db/sessions"; import { createApp } from "./http/app"; +import { createWebhookRegistry } from "./integrations"; import { saveApiKeyConnection } from "./llm/connections"; import { createLogger, type Logger } from "./logger"; @@ -55,7 +56,7 @@ export function bindTestApp( bindTestDatabase((database) => { assign({ database, - app: createApp({ config: testConfig(), database, log: silentLogger() }), + app: createApp(testAppDeps(database)), }); }); } @@ -130,6 +131,20 @@ export function silentLogger(): Logger { return createLogger("test", { level: "silent" }); } +/** Config + silent log + webhook registry for `createApp` in tests. */ +export function testAppDeps( + database: Database, + env: Record = {}, +): Parameters[0] { + const config = testConfig(env); + return { + config, + database, + log: silentLogger(), + integrations: createWebhookRegistry(config), + }; +} + export function manualTrigger(overrides: Partial = {}): Trigger { return { kind: "manual",