From 5e862b88f4acd808cf1eaff8aca956ec447baa0d Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Fri, 13 Nov 2026 01:14:14 +0300 Subject: [PATCH 1/3] feat(studio): R08b channel triggers served by the SDK in Studio Channel triggers (telegram, slack, github, teams, discord, http) now mount per agent through the SDK's own channel factories, so signature verification, sessions, replies and approval buttons behave exactly as under `lousho dev`. - server/triggers/channelTriggers.ts: CHANNEL_SECRET_KEYS aligned with the spec schema's required sets, channelFromTrigger secret resolution (github botName falls back to the spec name), one mountChannels call per agent, duplicate-kind and missing-secret errors surfaced via status(), remount on every TriggerHost reload via a new onReload hook - The channel middleware buffers the raw body itself and replays the exact bytes into the SDK handler, so platform signatures verify; registered before the /hooks express.raw mount, ahead of the JSON parser, with workspaceOf resolved earlier for the same reason - Abort listeners run before SDK handling on { channel, headers, body } (abortListeners.ts - the R03b core the webhook route will reuse); a non-true filter answers 200 { ok, skipped: 'filter' } without touching the SDK - Channel turns build the working-copy agent through buildRunnableAgent with trace-file + observe exporters; they land in Runs/Traces, never the test-pane queue - GET /agents/:id/triggers/status + runtimeClient.channelTriggerStatus; the Overview Triggers card shows each channel's webhook URL, Copy, mount errors, and the Runs/Traces note Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../server/__tests__/channelTriggers.test.ts | 452 ++++++++++++++++++ .../server/__tests__/hooksRoute.test.ts | 2 + .../server/__tests__/workspaceSeam.test.ts | 2 + apps/agent-forge/server/app.ts | 40 +- apps/agent-forge/server/routes/agentRoutes.ts | 11 + .../server/triggers/abortListeners.ts | 116 +++++ .../server/triggers/channelTriggers.ts | 440 +++++++++++++++++ apps/agent-forge/server/triggers/runtime.ts | 84 +++- .../server/triggers/triggerHost.ts | 11 + .../workspace/__tests__/middleware.test.ts | 2 + apps/agent-forge/server/workspace/context.ts | 27 +- apps/agent-forge/shared/wireTypes.ts | 10 + apps/agent-forge/src/runtime/runtimeClient.ts | 6 + .../screens/agentDetail/AgentOverviewTab.tsx | 66 ++- .../__tests__/AgentDetailRoute.test.tsx | 7 +- .../__tests__/AgentOverviewTab.test.tsx | 34 +- .../src/screens/agentDetail/agentOverview.css | 5 + 17 files changed, 1294 insertions(+), 21 deletions(-) create mode 100644 apps/agent-forge/server/__tests__/channelTriggers.test.ts create mode 100644 apps/agent-forge/server/triggers/abortListeners.ts create mode 100644 apps/agent-forge/server/triggers/channelTriggers.ts diff --git a/apps/agent-forge/server/__tests__/channelTriggers.test.ts b/apps/agent-forge/server/__tests__/channelTriggers.test.ts new file mode 100644 index 000000000..6cb3c97f9 --- /dev/null +++ b/apps/agent-forge/server/__tests__/channelTriggers.test.ts @@ -0,0 +1,452 @@ +/** + * R08b - channel triggers: `channelFromTrigger` secret mapping, per-agent + * mount/status, the `/hooks/:agentId/channel/` middleware (filter + * skip, abort listeners, raw-byte replay into the SDK handler) and the + * remount-on-spec-change contract. + */ +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; + +import { afterEach, describe, expect, it, vi } from 'vitest'; +import express, { type Express } from 'express'; +import request from 'supertest'; +import { createAgent, type AgentSpecV2, type SimpleAgent } from 'lousho'; +import { mockModel } from 'lousho/testing'; +import type { SecretRef, TriggerSpec } from 'lousho/spec'; + +import { AgentRepo } from '../storage/agentRepo'; +import { DraftStore } from '../storage/draftStore'; +import { SuggestionStore } from '../storage/suggestionStore'; +import { VersionStore } from '../storage/versionStore'; +import { IdleVersioner } from '../storage/idleVersioner'; +import { StudioEventBus } from '../events/eventBus'; +import { PassRunner } from '../passes/passRunner'; +import { createApp } from '../app'; +import type { LegacyAgentStore } from '../../src/persistence/AgentStore'; +import type { AgentHost } from '../host/agentHost'; +import type { RunManager } from '../runRegistry'; +import type { InboxService } from '../inbox/inboxService'; +import type { CronTriggers } from '../triggers/cronTriggers'; +import type { WorkspaceContext } from '../workspace/context'; +import { workspaceRefFor } from '../workspace/workspaceId'; +import { TriggerRegistry } from 'lousho/triggers'; +import { + CHANNEL_SECRET_KEYS, + ChannelTriggers, + channelFromTrigger, + type ChannelKind, + type ChannelTrigger, +} from '../triggers/channelTriggers'; +import type { HostRun, TriggerHost } from '../triggers/triggerHost'; + +const AGENT = 'a1'; + +const HTTP_TRIGGER: ChannelTrigger = { id: 't-http', type: 'channel', channel: 'http', secrets: {} }; + +function specWith(triggers: TriggerSpec[], extra: Partial = {}): AgentSpecV2 { + return { + version: 2, + name: 'chan-agent', + model: 'mock/mock-1', + instructions: 'Answer briefly.', + triggers, + ...extra, + } as AgentSpecV2; +} + +/** Resolves `{ secret: name }` from `secrets`, `{ env: name }` from the env, strings pass through. */ +function resolver(secrets: Record) { + return (ref: SecretRef | string): string | undefined => { + if (typeof ref === 'string') return ref; + if ('env' in ref) return process.env[ref.env]; + return secrets[ref.secret]; + }; +} + +const tmpDirs: string[] = []; + +function tmpDir(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'channel-triggers-')); + tmpDirs.push(dir); + return dir; +} + +afterEach(() => { + for (const dir of tmpDirs.splice(0)) fs.rmSync(dir, { recursive: true, force: true }); +}); + +interface Harness { + app: Express; + channels: ChannelTriggers; + repo: AgentRepo; + buildAgent: ReturnType>>; + host: { + runsOf: ReturnType>; + cancel: ReturnType>; + }; + baseDir: string; +} + +async function harness( + spec: AgentSpecV2, + options: { secrets?: Record; agent?: SimpleAgent; runs?: HostRun[] } = {} +): Promise { + const baseDir = tmpDir(); + const repo = new AgentRepo(baseDir); + await repo.writeSpec(AGENT, spec); + const buildAgent = vi.fn<[agentId: string], Promise>(async () => { + if (options.agent !== undefined) return options.agent; + return createAgent({ provider: mockModel(['Pong']) }) as unknown as SimpleAgent; + }); + const host = { + runsOf: vi.fn<[agentId: string], HostRun[]>(() => options.runs ?? []), + cancel: vi.fn<[runKey: string, reason: string], 'queued' | 'running' | undefined>(() => 'queued'), + }; + const channels = new ChannelTriggers({ + host: host as unknown as TriggerHost, + repo, + buildAgent, + resolve: resolver(options.secrets ?? {}), + log: vi.fn(), + }); + await channels.mount(AGENT); + const app = express(); + const handler = channels.handler(); + app.use((req, res, next) => handler(req, res, next)); + return { app, channels, repo, buildAgent, host, baseDir }; +} + +describe('channelFromTrigger', () => { + // Every SecretRef resolves to a factory-valid non-empty value (the refs + // are named `s-`, so discord's publicKey gets real hex). + const resolveAll = (ref: SecretRef | string): string | undefined => { + if (typeof ref === 'string') return ref; + if ('env' in ref) return process.env[ref.env]; + return ref.secret === 's-publicKey' ? 'ab'.repeat(32) : 'resolved'; + }; + + for (const kind of Object.keys(CHANNEL_SECRET_KEYS) as ChannelKind[]) { + it(`builds the ${kind} channel when every required secret resolves`, () => { + const trigger: ChannelTrigger = { + id: `t-${kind}`, + type: 'channel', + channel: kind, + secrets: Object.fromEntries( + [...CHANNEL_SECRET_KEYS[kind].required, ...CHANNEL_SECRET_KEYS[kind].optional].map((key) => [ + key, + { secret: `s-${key}` }, + ]) + ), + }; + const made = channelFromTrigger(trigger, resolveAll); + expect('channel' in made).toBe(true); + if ('channel' in made) expect(made.channel.name).toBe(kind); + }); + } + + it('reports the first missing required key by name only', () => { + const trigger: ChannelTrigger = { + id: 't-tg', + type: 'channel', + channel: 'telegram', + secrets: { botToken: { secret: 'bt' } }, + }; + expect(channelFromTrigger(trigger, resolver({ bt: 'x' }))).toEqual({ + error: 'Channel telegram needs secretToken', + }); + }); + + it('treats an unresolved optional key as absent, not an error', () => { + const trigger: ChannelTrigger = { + id: 't-tg', + type: 'channel', + channel: 'telegram', + secrets: { + botToken: { secret: 'bt' }, + secretToken: { secret: 'st' }, + botUsername: { secret: 'missing' }, + }, + }; + const made = channelFromTrigger(trigger, resolver({ bt: 'x', st: 'y' })); + expect('channel' in made).toBe(true); + }); + + it("falls back to the spec name for github's botName", () => { + const trigger: ChannelTrigger = { + id: 't-gh', + type: 'channel', + channel: 'github', + secrets: { webhookSecret: { secret: 'wh' }, token: { secret: 'tok' } }, + }; + const made = channelFromTrigger(trigger, resolver({ wh: 'x', tok: 'y' }), 'my-agent'); + expect('channel' in made).toBe(true); + // No botName secret and no specName -> still an error naming the key. + expect(channelFromTrigger(trigger, resolver({ wh: 'x', tok: 'y' }))).toEqual({ + error: 'Channel github needs botName', + }); + }); +}); + +describe('ChannelTriggers.mount / status', () => { + it('serves the http channel end to end through the SDK handler', async () => { + const replies = ['Pong', 'Again']; + const agent = createAgent({ provider: mockModel(replies) }); + const { app, channels, buildAgent } = await harness(specWith([HTTP_TRIGGER]), { + agent: agent as unknown as SimpleAgent, + }); + + const first = await request(app) + .post(`/hooks/${AGENT}/channel/http`) + .set('content-type', 'application/json') + .send({ sessionKey: 'u1', input: 'ping' }); + expect(first.status).toBe(200); + expect(first.body).toMatchObject({ text: 'Pong', finishReason: 'stop' }); + + // Same session continues the conversation; the next scripted reply lands. + const second = await request(app) + .post(`/hooks/${AGENT}/channel/http`) + .set('content-type', 'application/json') + .send({ sessionKey: 'u1', input: 'ping again' }); + expect(second.status).toBe(200); + expect(second.body.text).toBe('Again'); + expect(buildAgent).toHaveBeenCalledWith(AGENT); + + expect(channels.status(AGENT)).toEqual([ + { triggerId: 't-http', channel: 'http', url: `/hooks/${AGENT}/channel/http`, error: null }, + ]); + }); + + it('answers 404-shaped fallthrough for a channel path with no mounted trigger', async () => { + const { app } = await harness(specWith([HTTP_TRIGGER])); + const res = await request(app).post(`/hooks/${AGENT}/channel/telegram`).send('{}'); + // No telegram trigger mounted -> next() -> express default 404. + expect(res.status).toBe(404); + }); + + it('does not serve a trigger whose required secret is missing, and status names the key', async () => { + // The spec schema requires the secretToken KEY; the mount error covers + // the SecretRef that does not resolve to a value. + const tg: ChannelTrigger = { + id: 't-tg', + type: 'channel', + channel: 'telegram', + secrets: { botToken: { secret: 'bt' }, secretToken: { secret: 'st' } }, + }; + const { app, channels, buildAgent } = await harness(specWith([tg]), { secrets: { bt: 'x' } }); + + expect(channels.status(AGENT)).toEqual([ + { + triggerId: 't-tg', + channel: 'telegram', + url: `/hooks/${AGENT}/channel/telegram`, + error: 'Channel telegram needs secretToken', + }, + ]); + expect(buildAgent).not.toHaveBeenCalled(); + const res = await request(app).post(`/hooks/${AGENT}/channel/telegram`).send('{}'); + expect(res.status).toBe(404); + }); + + it('serves only the first of two triggers on the same channel kind', async () => { + const mk = (id: string): ChannelTrigger => ({ + id, + type: 'channel', + channel: 'telegram', + secrets: { botToken: { secret: 'bt' }, secretToken: { secret: 'st' } }, + }); + const { channels } = await harness(specWith([mk('t1'), mk('t2')]), { + secrets: { bt: 'x', st: 'y' }, + agent: createAgent({ provider: mockModel(['ok']) }) as unknown as SimpleAgent, + }); + + const rows = channels.status(AGENT); + expect(rows.find((r) => r.triggerId === 't1')?.error).toBeNull(); + expect(rows.find((r) => r.triggerId === 't2')?.error).toBe('Only one telegram channel per agent'); + }); + + it('remounts on spec change: the next request runs the rebuilt agent', async () => { + const first = createAgent({ provider: mockModel(['First']) }); + const { app, repo, channels, buildAgent } = await harness(specWith([HTTP_TRIGGER]), { + agent: first as unknown as SimpleAgent, + }); + + const before = await request(app) + .post(`/hooks/${AGENT}/channel/http`) + .set('content-type', 'application/json') + .send({ sessionKey: 's', input: 'hi' }); + expect(before.body.text).toBe('First'); + + // Edit the working copy (new instructions -> the host reloads -> onReload remounts). + await repo.writeSpec(AGENT, specWith([HTTP_TRIGGER], { instructions: 'Answer differently.' })); + buildAgent.mockImplementation(async () => createAgent({ provider: mockModel(['Second']) }) as unknown as SimpleAgent); + await channels.mount(AGENT); + + const after = await request(app) + .post(`/hooks/${AGENT}/channel/http`) + .set('content-type', 'application/json') + .send({ sessionKey: 's2', input: 'hi' }); + expect(after.body.text).toBe('Second'); + expect(buildAgent).toHaveBeenCalledTimes(2); + }); +}); + +describe('channel request pipeline', () => { + it('skips the SDK handler with 200 { ok, skipped: filter } when the filter is not true', async () => { + const filtered: ChannelTrigger = { ...HTTP_TRIGGER, filter: 'event.body.input === "go"' }; + const agent = createAgent({ provider: mockModel(['Pong', 'Pong']) }); + const { app } = await harness(specWith([filtered]), { agent: agent as unknown as SimpleAgent }); + + const skipped = await request(app) + .post(`/hooks/${AGENT}/channel/http`) + .set('content-type', 'application/json') + .send({ sessionKey: 'u1', input: 'stop' }); + expect(skipped.status).toBe(200); + expect(skipped.body).toEqual({ ok: true, skipped: 'filter' }); + + const kept = await request(app) + .post(`/hooks/${AGENT}/channel/http`) + .set('content-type', 'application/json') + .send({ sessionKey: 'u1', input: 'go' }); + expect(kept.status).toBe(200); + expect(kept.body.text).toBe('Pong'); + }); + + it('replays the exact raw bytes to the SDK handler (signature verification sees them)', async () => { + // Slack's signing secret is checked over the raw body; a body whose + // bytes would change under re-serialization (key order/whitespace) + // must still verify. Sign it by hand with the real algorithm. + const { createHmac } = await import('node:crypto'); + const body = '{"token":"x","team_id":"T1","type":"event_callback","event":{"type":"message","text":"hi"}}'; + const signingSecret = 'slack-secret'; + const timestamp = String(Math.floor(Date.now() / 1000)); + const basestring = `v0:${timestamp}:${body}`; + const signature = `v0=${createHmac('sha256', signingSecret).update(basestring).digest('hex')}`; + + const slack: ChannelTrigger = { + id: 't-slack', + type: 'channel', + channel: 'slack', + secrets: { signingSecret: { secret: 'ss' }, botToken: { secret: 'bt' } }, + }; + // A slack event the SDK parses but whose session turn we do not need to + // inspect - a wrong/altered body fails signature first (401), the exact + // bytes pass it (any non-401 proves the replay preserved them). + const { app } = await harness(specWith([slack]), { + secrets: { ss: signingSecret, bt: 'xoxb-fake' }, + agent: createAgent({ provider: mockModel(['ok']) }) as unknown as SimpleAgent, + }); + + // A tampered signature over the same bytes must still 401 - proving the + // SDK verified the replayed raw body, not a re-serialized one. + const bad = await request(app) + .post(`/hooks/${AGENT}/channel/slack`) + .set('content-type', 'application/json') + .set('x-slack-request-timestamp', timestamp) + .set('x-slack-signature', 'v0=deadbeef') + .send(body); + expect(bad.status).toBe(401); + + const res = await request(app) + .post(`/hooks/${AGENT}/channel/slack`) + .set('content-type', 'application/json') + .set('x-slack-request-timestamp', timestamp) + .set('x-slack-signature', signature) + .send(body); + + expect(res.status).not.toBe(401); + }); + + it('runs abort listeners before SDK handling and does not block the request', async () => { + const run: HostRun = { + runKey: 'rk-old', + agentId: AGENT, + triggerId: 't-http', + event: { input: 'old request' }, + input: 'old request', + status: 'queued', + enqueuedAt: '2026-01-01T00:00:00.000Z', + }; + const spec = specWith( + [HTTP_TRIGGER], + { + abortListeners: [ + { + id: 'l1', + triggerId: 't-http', + when: 'event.channel == "http"', + match: 'run.event.input == "old request"', + scope: 'both', + }, + ], + } as Partial + ); + const agent = createAgent({ provider: mockModel(['Pong']) }); + const { app, host } = await harness(spec, { + agent: agent as unknown as SimpleAgent, + runs: [run], + }); + + const res = await request(app) + .post(`/hooks/${AGENT}/channel/http`) + .set('content-type', 'application/json') + .send({ sessionKey: 'u1', input: 'hi' }); + + expect(res.status).toBe(200); + expect(res.body.text).toBe('Pong'); + expect(host.cancel).toHaveBeenCalledWith('rk-old', 'abort_listener:l1'); + // The event the listener matched on carries channel + lower-cased header names + parsed body. + expect(host.runsOf).toHaveBeenCalledWith(AGENT); + }); +}); + +describe('channel route inside createApp', () => { + const emptyStore: LegacyAgentStore = { + list: async () => [], + load: async () => undefined, + save: async () => {}, + remove: async () => {}, + }; + + it('serves the channel before the /hooks raw-body middleware can consume the stream', async () => { + const baseDir = tmpDir(); + const repo = new AgentRepo(baseDir); + await repo.writeSpec(AGENT, specWith([HTTP_TRIGGER])); + const channels = new ChannelTriggers({ + host: { runsOf: () => [], cancel: () => undefined } as unknown as TriggerHost, + repo, + buildAgent: async () => createAgent({ provider: mockModel(['ViaApp']) }) as unknown as SimpleAgent, + resolve: resolver({}), + log: vi.fn(), + }); + await channels.mount(AGENT); + + const workspace: WorkspaceContext = { + ref: workspaceRefFor(baseDir), + repo, + agentStore: emptyStore, + versions: new VersionStore(repo), + drafts: new DraftStore(repo), + idleVersioner: new IdleVersioner({ capture: async () => {} }), + suggestions: new SuggestionStore(repo), + host: {} as AgentHost, + runs: {} as RunManager, + chatTriggers: new TriggerRegistry(), + events: new StudioEventBus(), + passes: new PassRunner({ baseDir }), + inbox: {} as InboxService, + triggers: {} as TriggerHost, + cron: {} as CronTriggers, + channels, + }; + const app = createApp({ workspace, baseDir }); + + const res = await request(app) + .post(`/hooks/${AGENT}/channel/http`) + .set('content-type', 'application/json') + .send({ sessionKey: 'u9', input: 'ping' }); + + expect(res.status).toBe(200); + expect(res.body.text).toBe('ViaApp'); + }); +}); diff --git a/apps/agent-forge/server/__tests__/hooksRoute.test.ts b/apps/agent-forge/server/__tests__/hooksRoute.test.ts index 4876fd5ae..1ec840bd5 100644 --- a/apps/agent-forge/server/__tests__/hooksRoute.test.ts +++ b/apps/agent-forge/server/__tests__/hooksRoute.test.ts @@ -20,6 +20,7 @@ import { StudioEventBus } from '../events/eventBus'; import { PassRunner } from '../passes/passRunner'; import type { InboxService } from '../inbox/inboxService'; import type { TriggerHost } from '../triggers/triggerHost'; +import type { ChannelTriggers } from '../triggers/channelTriggers'; import type { CronTriggers } from '../triggers/cronTriggers'; import type { WorkspaceContext } from '../workspace/context'; import { workspaceRefFor } from '../workspace/workspaceId'; @@ -94,6 +95,7 @@ function appWith( inbox: {} as InboxService, triggers: host, cron: {} as CronTriggers, + channels: {} as ChannelTriggers, }; const secrets = new SecretsStore(baseDir); secrets.setSecret('wh', SECRET); diff --git a/apps/agent-forge/server/__tests__/workspaceSeam.test.ts b/apps/agent-forge/server/__tests__/workspaceSeam.test.ts index 069880d50..6aae40b1e 100644 --- a/apps/agent-forge/server/__tests__/workspaceSeam.test.ts +++ b/apps/agent-forge/server/__tests__/workspaceSeam.test.ts @@ -23,6 +23,7 @@ import { StudioEventBus } from '../events/eventBus'; import { PassRunner } from '../passes/passRunner'; import type { InboxService } from '../inbox/inboxService'; import type { TriggerHost } from '../triggers/triggerHost'; +import type { ChannelTriggers } from '../triggers/channelTriggers'; import type { CronTriggers } from '../triggers/cronTriggers'; import type { WorkspaceContext } from '../workspace/context'; @@ -133,6 +134,7 @@ describe('workspace seam', () => { inbox: {} as InboxService, triggers: {} as TriggerHost, cron: {} as CronTriggers, + channels: {} as ChannelTriggers, }; const app = createApp({ workspace, baseDir }); diff --git a/apps/agent-forge/server/app.ts b/apps/agent-forge/server/app.ts index 90409df90..c4b014ef6 100644 --- a/apps/agent-forge/server/app.ts +++ b/apps/agent-forge/server/app.ts @@ -26,7 +26,8 @@ * GET /inbox/count -> InboxCount * POST /inbox/alerts/:alertId/dismiss -> 204 | 404 * POST /hooks/:agentId/ -> webhook trigger fire (triggers/webhookRoute.ts, R08a); - * public - HMAC or loopback-only, no studio token + * ALL /hooks/:agentId/channel/ -> channel trigger (triggers/channelTriggers.ts, R08b); + * public - platform-signed, no studio token * GET /workspace -> { baseDir, workspace } - the workspace Studio was launched in * GET/POST /workspaces -> WorkspaceList · register a folder as a workspace (workspace/workspaceRoutes.ts) * DELETE /workspaces/:ws -> drop a registry entry (the folder is never touched; the launch workspace stays) @@ -1016,12 +1017,21 @@ function legacyWorkspaceContext( }); inbox.start(); // The legacy pair gets the same trigger stack the real context builds - // (R08a host + R08c cron) - webhook routes read it through workspaceOf(req). - const { triggers, cron } = createTriggerRuntime({ + // (R08a host + R08c cron + R08b channels) - webhook routes read it + // through workspaceOf(req). + const { triggers, cron, channels } = createTriggerRuntime({ baseDir: options.baseDir, repo, runManager: options.runManager, events, + studio: { + baseDir: options.baseDir, + checkpointStore: new FileCheckpointStore(options.baseDir), + approvalStore: new FileApprovalStore(options.baseDir), + secretsStore: options.secretsStore, + settings: autoApproval.settings, + spend: autoApproval.spendLog, + }, }); return { ref: workspaceRefFor(options.baseDir), @@ -1049,6 +1059,7 @@ function legacyWorkspaceContext( inbox, triggers, cron, + channels, }; } @@ -1127,16 +1138,29 @@ export function createApp(options: CreateAppOptions): Express { } next(); }); + // Resolve every request's workspace before the first route so handlers + // read their per-workspace stores from workspaceOf(req), never a + // closed-over singleton. Mounted ahead of the body parsers: it reads + // only headers/query, and the channel route below needs workspaceOf + // before `express.raw` can consume the stream (R08b). Studio serves + // exactly one workspace today - the launch folder - so this changes + // nothing user-visible. + app.use(workspaceMiddleware(workspaceResolver)); + // Channel triggers (triggers/channelTriggers.ts, R08b): `ALL + // /hooks/:agentId/channel/` is served by the SDK's own channel + // handlers, which read the raw request stream for platform signature + // checks - so this route MUST sit before the `express.raw` mount on + // `/hooks` (and before express.json). Channel requests carry the + // platform's signature, not a studio token; the /hooks gate exemption + // already covers them. + app.all('/hooks/:agentId/channel/*splat', (req, res, next) => { + workspaceOf(req).channels.handler()(req, res, next); + }); // Raw bytes for `/hooks/**` BEFORE the JSON parser: webhook HMACs sign // the exact body, and `express.raw` sets `req._body` so the JSON parser // below leaves it alone. Every other route keeps `express.json`. app.use('/hooks', express.raw({ type: () => true, limit: '1mb' })); app.use(express.json({ limit: '2mb' })); - // Resolve every request's workspace before the first route so handlers - // read their per-workspace stores from workspaceOf(req), never a - // closed-over singleton. Studio serves exactly one workspace today - - // the launch folder - so this changes nothing user-visible. - app.use(workspaceMiddleware(workspaceResolver)); app.get('/health', (_req, res) => res.status(200).send('ok')); diff --git a/apps/agent-forge/server/routes/agentRoutes.ts b/apps/agent-forge/server/routes/agentRoutes.ts index faf78870a..56fb13a82 100644 --- a/apps/agent-forge/server/routes/agentRoutes.ts +++ b/apps/agent-forge/server/routes/agentRoutes.ts @@ -384,6 +384,17 @@ export function registerAgentRoutes(app: Express, deps: AgentRouteDeps): void { }) ); + // The agent's channel-trigger mount status (R08b): one row per + // `channel` trigger with its public URL and mount error (missing + // secret, duplicate kind), for the Overview Triggers card. + app.get( + '/agents/:id/triggers/status', + asyncRoute(async (req, res) => { + const ws = workspaceOf(req); + res.json(ws.channels.status(paramId(req))); + }) + ); + app.put( '/agents/:id', asyncRoute(async (req, res) => { diff --git a/apps/agent-forge/server/triggers/abortListeners.ts b/apps/agent-forge/server/triggers/abortListeners.ts new file mode 100644 index 000000000..2f22eece8 --- /dev/null +++ b/apps/agent-forge/server/triggers/abortListeners.ts @@ -0,0 +1,116 @@ +/** + * Abort listeners (R03b core, consumed by the channel path in R08b and by + * the webhook route in R03b): when an inbound webhook/channel event + * arrives, every `spec.abortListeners` entry whose `when` safeExpression + * is strictly `true` over `{ event }` cancels the matching queued or + * running runs of the trigger it watches (`match` pairs the event with + * each candidate run's own event, also strictly `true`). + * + * All of an agent's listeners evaluate on every inbound event, + * independently of which trigger delivered it - the listener's + * `triggerId` selects which RUNS are candidates, not which events count. + * + * Truthiness is strict `=== true`: a listener never fires on a non-empty + * string or any other truthy value. An `ExpressionError` at run time logs + * `Abort listener failed: ` and skips that listener - it + * never blocks the dispatch. Event bodies are never logged (they may + * carry secrets); only listener ids and expressions are. + */ +import type { AgentSpecV2 } from 'lousho'; +import { evaluateSafeExpression, ExpressionError } from 'lousho/flows'; + +import type { HostRun, TriggerHost } from './triggerHost'; + +export interface AbortOutcome { + listenerId: string; + cancelled: Array<{ runKey: string; status: 'queued' | 'running' }>; +} + +/** What {@link applyAbortListeners} needs: the spec, the delivering trigger, the event, and the host's run views. */ +export interface AbortListenerInput { + spec: AgentSpecV2; + agentId: string; + /** The trigger that delivered the event (its own run is not queued yet, so it cannot self-cancel). */ + triggerId: string; + event: unknown; + host: Pick; + log: (agentId: string, message: string, level?: 'info' | 'warn' | 'error') => void; +} + +/** The run statuses a listener `scope` may cancel. */ +function scopeAllows(scope: string, run: HostRun): boolean { + if (scope === 'both') return true; + return run.status === scope; +} + +/** + * `evaluateSafeExpression(...) === true`, or `undefined` when the + * expression itself fails - an {@link ExpressionError} logs + * `Abort listener failed: ` and never blocks the dispatch. + */ +function evalListener( + expression: string, + scope: Record, + input: { agentId: string; listenerId: string; log: (agentId: string, message: string, level?: 'info' | 'warn' | 'error') => void } +): boolean | undefined { + try { + return evaluateSafeExpression(expression, scope) === true; + } catch (error) { + if (error instanceof ExpressionError) { + input.log(input.agentId, `Abort listener ${input.listenerId} failed: ${error.reason}`, 'warn'); + return undefined; + } + throw error; + } +} + +/** + * Cancels every candidate run of `listener` whose `match` evaluates true; + * appends them to `outcome.cancelled`. An expression error stops the scan + * early (`evalListener` already logged it). + */ +function cancelMatchingRuns( + listener: NonNullable[number], + input: AbortListenerInput, + outcome: AbortOutcome +): void { + const ctx = { agentId: input.agentId, log: input.log, listenerId: listener.id }; + for (const run of input.host.runsOf(input.agentId)) { + if (run.triggerId !== listener.triggerId || !scopeAllows(listener.scope, run)) continue; + const matches = evalListener( + listener.match, + { event: input.event, run: { event: run.event, triggerId: run.triggerId } }, + ctx + ); + if (matches === undefined) return; + if (!matches) continue; + const cancelled = input.host.cancel(run.runKey, `abort_listener:${listener.id}`); + if (cancelled === undefined) continue; + outcome.cancelled.push({ runKey: run.runKey, status: cancelled }); + input.log(input.agentId, `Run cancelled by abort listener ${listener.id}: ${listener.when}`, 'warn'); + } +} + +/** + * Runs before dispatch for every inbound webhook/channel event of + * `agentId`. `triggerId` names the trigger that delivered the event (the + * run the event will itself start is not yet queued, so it can never + * cancel itself). Returns one {@link AbortOutcome} per listener that + * cancelled at least one run. + */ +export function applyAbortListeners(input: AbortListenerInput): AbortOutcome[] { + const outcomes: AbortOutcome[] = []; + for (const listener of input.spec.abortListeners ?? []) { + const fires = evalListener( + listener.when, + { event: input.event }, + { agentId: input.agentId, listenerId: listener.id, log: input.log } + ); + if (fires !== true) continue; + + const outcome: AbortOutcome = { listenerId: listener.id, cancelled: [] }; + cancelMatchingRuns(listener, input, outcome); + if (outcome.cancelled.length > 0) outcomes.push(outcome); + } + return outcomes; +} diff --git a/apps/agent-forge/server/triggers/channelTriggers.ts b/apps/agent-forge/server/triggers/channelTriggers.ts new file mode 100644 index 000000000..b4461ea63 --- /dev/null +++ b/apps/agent-forge/server/triggers/channelTriggers.ts @@ -0,0 +1,440 @@ +/** + * Channel triggers (R08b, PLAN §3.1/§15): a `{ type: 'channel' }` trigger + * makes Studio serve that chat platform at + * `POST /hooks//channel/` through the SDK's own channel + * implementation (`mountChannels` + the per-platform factories), so + * signature verification, session handling, replies and approval buttons + * behave exactly as under `lousho dev`. + * + * Per agent, all its channel triggers mount under one + * `mountChannels(agent, channels, { basePath })` handler; channel names + * are the platform kinds, so two triggers of one kind on an agent are an + * error ("Only one telegram channel per agent") and only the first is + * served. + * + * The Express middleware buffers the raw body itself (the SDK handler + * reads the stream, so this route must run BEFORE app.ts's + * `express.raw` on `/hooks`), evaluates the spec's abort listeners and + * the trigger's `filter` on `{ channel, headers, body }`, then replays + * the exact bytes to the SDK handler - platform signatures are checked + * over those bytes inside the SDK. + * + * Channel turns run through `agent.session()` inside the SDK: they land + * in traces (the file + observe exporters `buildAgent` wires) but not in + * the test pane, the per-agent trigger queue or the failure streak. + */ +import { Readable } from 'node:stream'; +import type * as http from 'node:http'; + +import type { Request, Response, NextFunction } from 'express'; +import { + discordChannel, + githubChannel, + httpChannel, + mountChannels, + slackChannel, + teamsChannel, + telegramChannel, + type AgentSpecV2, + type Channel, + type ChannelsHandler, + type SimpleAgent, +} from 'lousho'; +import type { SecretRef, TriggerSpec } from 'lousho/spec'; +import { evaluateSafeExpression } from 'lousho/flows'; + +import type { ChannelTriggerStatus } from '../../shared/wireTypes'; +import type { AgentRepo } from '../storage/agentRepo'; +import { applyAbortListeners } from './abortListeners'; +import type { TriggerHost } from './triggerHost'; + +export type ChannelKind = Extract['channel']; + +/** A `channel` trigger entry. */ +export type ChannelTrigger = Extract; + +/** + * Option keys each factory takes from `trigger.secrets` (values resolved + * to strings). `required` mirrors the SDK spec schema's own + * `CHANNEL_SECRET_KEYS` (src/spec/v2/triggers.ts - `writeSpec` already + * refuses a spec missing those keys, so a miss here means the SecretRef + * did not resolve); `optional` are factory options the schema does not + * require. `github.botName` is the odd one: it names the bot, not a + * credential, so the host supplies the spec's `name` when `secrets` + * carries no `botName` ref. + */ +export const CHANNEL_SECRET_KEYS: Record = { + telegram: { required: ['botToken', 'secretToken'], optional: ['botUsername'] }, + slack: { required: ['signingSecret', 'botToken'], optional: [] }, + github: { required: ['webhookSecret', 'token', 'botName'], optional: ['botLogin'] }, + teams: { required: ['appId', 'appPassword'], optional: ['tenantId'] }, + discord: { required: ['publicKey', 'applicationId'], optional: ['botToken'] }, + http: { required: [], optional: [] }, +}; + +/** Resolves a SecretRef (or a plain string, which passes through) to its value; `undefined` when unset. */ +export type SecretResolver = (ref: SecretRef | string) => string | undefined; + +/** + * Resolves every required+optional key of {@link CHANNEL_SECRET_KEYS} for + * one trigger. The first required miss returns `{ error: "Channel + * needs " }`; optional misses are simply absent from the map. + * `github.botName` falls back to `specName` (it names the bot, not a + * credential). + */ +function collectSecretValues( + trigger: ChannelTrigger, + resolve: SecretResolver, + specName?: string +): { values: Map } | { error: string } { + const keys = CHANNEL_SECRET_KEYS[trigger.channel]; + const values = new Map(); + for (const key of [...keys.required, ...keys.optional]) { + const ref = trigger.secrets?.[key]; + const isRequired = keys.required.includes(key); + if (ref === undefined && isRequired && trigger.channel === 'github' && key === 'botName' && specName !== undefined) { + values.set(key, specName); + continue; + } + if (ref === undefined) { + if (isRequired) return { error: `Channel ${trigger.channel} needs ${key}` }; + continue; + } + const value = resolve(ref); + if (value === undefined || value === '') { + if (isRequired) return { error: `Channel ${trigger.channel} needs ${key}` }; + continue; + } + values.set(key, value); + } + return { values }; +} + +/** + * Builds the SDK `Channel` for one channel trigger. Every required key of + * {@link CHANNEL_SECRET_KEYS} must resolve (github's `botName` falls back + * to `specName`); the first miss returns `{ error: "Channel needs + * " }`. Optional keys that do not resolve are simply absent. + */ +export function channelFromTrigger( + trigger: ChannelTrigger, + resolve: SecretResolver, + specName?: string +): { channel: Channel } | { error: string } { + const collected = collectSecretValues(trigger, resolve, specName); + if ('error' in collected) return collected; + const { values } = collected; + const v = (key: string): string => values.get(key) ?? ''; + const opt = (key: string): Record => (values.has(key) ? { [key]: v(key) } : {}); + try { + switch (trigger.channel) { + case 'telegram': + return { + channel: telegramChannel({ + botToken: v('botToken'), + secretToken: v('secretToken'), + ...opt('botUsername'), + }), + }; + case 'slack': + return { channel: slackChannel({ signingSecret: v('signingSecret'), botToken: v('botToken') }) }; + case 'github': + return { + channel: githubChannel({ + webhookSecret: v('webhookSecret'), + botName: v('botName'), + token: v('token'), + ...opt('botLogin'), + }), + }; + case 'teams': + return { + channel: teamsChannel({ appId: v('appId'), appPassword: v('appPassword'), ...opt('tenantId') }), + }; + case 'discord': + return { + channel: discordChannel({ + publicKey: v('publicKey'), + applicationId: v('applicationId'), + ...opt('botToken'), + }), + }; + case 'http': + return { channel: httpChannel() }; + } + } catch (error) { + // A factory ConfigurationError (empty credential, bad option combo) + // surfaces as this trigger's mount error, same as a missing secret. + return { error: `Channel ${trigger.channel}: ${error instanceof Error ? error.message : String(error)}` }; + } +} + +/** + * Builds every servable channel of one agent's channel triggers: the first + * trigger of each kind wins (a second is an "Only one channel per + * agent" error), a trigger whose secrets do not resolve or whose factory + * rejects the options lands in `errors` under its id. Both outcomes are + * logged once per mount. + */ +function collectAgentChannels( + agentId: string, + triggers: ChannelTrigger[], + specName: string | undefined, + deps: { resolve: SecretResolver; log: TriggerHost['log'] } +): { channels: Channel[]; errors: Map } { + const errors = new Map(); + const seen = new Set(); + const channels: Channel[] = []; + for (const trigger of triggers) { + let error: string | undefined; + if (seen.has(trigger.channel)) { + error = `Only one ${trigger.channel} channel per agent`; + } else { + seen.add(trigger.channel); + const made = channelFromTrigger(trigger, deps.resolve, specName); + if ('channel' in made) { + channels.push(made.channel); + continue; + } + error = made.error; + } + errors.set(trigger.id, error); + deps.log(agentId, `Channel trigger ${trigger.id}: ${error}`, 'warn'); + } + return { channels, errors }; +} + +/** One agent's mounted channel surface. */ +interface AgentMount { + handler: ChannelsHandler; + spec: AgentSpecV2; + triggers: ChannelTrigger[]; +} + +/** `/hooks//channel/` out of a request url; `undefined` for anything else. */ +function channelRoute(url: string | undefined): { agentId: string; name: string } | undefined { + const pathname = (url ?? '/').split('?')[0].split('#')[0]; + const match = /^\/hooks\/([^/]+)\/channel\/([^/?#]+)/.exec(pathname); + if (match === null) return undefined; + try { + return { agentId: decodeURIComponent(match[1]), name: decodeURIComponent(match[2]) }; + } catch { + return undefined; + } +} + +/** The request body as a Buffer, capped at the SDK's own 1 MB channel limit. */ +function readBody(req: http.IncomingMessage, limit = 1024 * 1024): Promise { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + let bytes = 0; + req.on('data', (chunk: Buffer | string) => { + const data = typeof chunk === 'string' ? Buffer.from(chunk) : chunk; + bytes += data.length; + if (bytes <= limit) chunks.push(data); + }); + req.on('end', () => { + if (bytes > limit) { + reject(new Error('Request body exceeds the 1MB channel limit')); + return; + } + resolve(Buffer.concat(chunks)); + }); + req.on('error', reject); + }); +} + +/** `body` parsed as JSON, or `null` for anything else (the abort/filter event's `body`). */ +function jsonOrNull(body: Buffer): unknown { + try { + return JSON.parse(body.toString('utf8')); + } catch { + return null; + } +} + +/** + * A fresh `IncomingMessage` carrying `req`'s method/url/headers and + * `body` as its stream content - the SDK channel handler reads the raw + * stream for signature checks, so the buffered bytes must replay + * exactly. + */ +function replayRequest(req: http.IncomingMessage, body: Buffer): http.IncomingMessage { + const replay = new Readable({ read() {} }) as unknown as http.IncomingMessage; + replay.method = req.method; + replay.url = req.url; + replay.headers = req.headers; + replay.httpVersion = req.httpVersion; + replay.httpVersionMajor = req.httpVersionMajor; + replay.httpVersionMinor = req.httpVersionMinor; + process.nextTick(() => { + replay.push(body); + replay.push(null); + }); + return replay; +} + +export class ChannelTriggers { + /** agentId -> its mounted channel handler. */ + private readonly mounts = new Map(); + /** agentId -> the spec `status()` reports (kept even when nothing mounts). */ + private readonly specs = new Map(); + /** agentId -> triggerId -> mount error, reset on every mount. */ + private readonly errors = new Map>(); + /** Serializes mounts per agent so a reload during a build cannot interleave. */ + private readonly mounting = new Map>(); + + constructor( + private readonly deps: { + host: TriggerHost; + repo: AgentRepo; + /** Builds the agent a channel turn runs on (the working copy, rebuilt per mount). */ + buildAgent: (agentId: string) => Promise; + resolve: SecretResolver; + log: TriggerHost['log']; + } + ) {} + + /** + * (Re)mounts the channel triggers of one agent - called from + * `TriggerHost.reload` on every spec change. A spec with no channel + * triggers (or an unreadable one) unmounts; a build failure unmounts + * too, logged, so a removed trigger is never served stale. + */ + mount(agentId: string): Promise { + const prev = this.mounting.get(agentId) ?? Promise.resolve(); + const next = prev.then(() => this.mountNow(agentId)); + this.mounting.set( + agentId, + next.catch(() => {}) + ); + return next; + } + + private async mountNow(agentId: string): Promise { + const spec = (await this.deps.repo.readSpec(agentId).catch(() => undefined)) as AgentSpecV2 | undefined; + if (spec === undefined) { + this.specs.delete(agentId); + this.mounts.delete(agentId); + this.errors.delete(agentId); + return; + } + this.specs.set(agentId, spec); + const triggers = (spec.triggers ?? []).filter((t): t is ChannelTrigger => t.type === 'channel'); + if (triggers.length === 0) { + this.mounts.delete(agentId); + this.errors.delete(agentId); + return; + } + + const { channels, errors } = collectAgentChannels(agentId, triggers, spec.name, { + resolve: this.deps.resolve, + log: this.deps.log, + }); + this.errors.set(agentId, errors); + if (channels.length === 0) { + this.mounts.delete(agentId); + return; + } + + let agent: SimpleAgent; + try { + agent = await this.deps.buildAgent(agentId); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.deps.log(agentId, `Channel mount failed: ${message}`, 'error'); + // `errors` is already stored - surface the build failure on every + // trigger that had no error of its own, so status() never shows a + // served-looking row for a channel that is not mounted. + for (const trigger of triggers) { + if (!errors.has(trigger.id)) errors.set(trigger.id, `Channel mount failed: ${message}`); + } + this.mounts.delete(agentId); + return; + } + const handler = mountChannels(agent, channels, { + basePath: `/hooks/${agentId}/channel`, + onError: (error, context) => + this.deps.log( + agentId, + `Channel ${context.channel} failed at ${context.stage}: ${error instanceof Error ? error.message : String(error)}`, + 'error' + ), + }); + this.mounts.set(agentId, { handler, spec, triggers }); + } + + /** + * Express middleware for `/hooks/:agentId/channel/*`: non-channel paths + * `next()`; channel paths buffer the body, run abort listeners and the + * trigger `filter`, then hand the replayed request to the SDK handler. + */ + handler(): (req: Request, res: Response, next: NextFunction) => void { + return (req, res, next) => { + void this.handle(req, res, next).catch(next); + }; + } + + private async handle(req: Request, res: Response, next: NextFunction): Promise { + const route = channelRoute(req.url); + if (route === undefined) { + next(); + return; + } + const mount = this.mounts.get(route.agentId); + const trigger = mount?.triggers.find((t) => t.channel === route.name); + if (mount === undefined || trigger === undefined) { + next(); + return; + } + + let body: Buffer; + try { + body = await readBody(req); + } catch { + res.status(413).json({ error: 'Request body exceeds the 1MB channel limit' }); + return; + } + + // The event abort listeners and the filter see: channel kind, the + // lower-cased header NAMES (values may carry secrets, so they are + // never exposed) and the parsed JSON body (null when not JSON). + const event = { + channel: route.name, + headers: Object.fromEntries(Object.keys(req.headers).map((h) => [h.toLowerCase(), true])), + body: jsonOrNull(body), + }; + applyAbortListeners({ + spec: mount.spec, + agentId: route.agentId, + triggerId: trigger.id, + event, + host: this.deps.host, + log: this.deps.log, + }); + if (trigger.filter !== undefined && evaluateSafeExpression(trigger.filter, { event }) !== true) { + res.status(200).json({ ok: true, skipped: 'filter' }); + return; + } + + // The SDK handler answers the request itself; `false` means the path + // was not a channel route after all (e.g. a GET) - keep looking. + const handled = await mount.handler(replayRequest(req, body), res); + if (!handled) next(); + } + + /** This agent's channel triggers with their mount URL and error, for the Overview card. */ + status(agentId: string): ChannelTriggerStatus[] { + const spec = this.specs.get(agentId); + if (spec === undefined) return []; + const errors = this.errors.get(agentId); + return (spec.triggers ?? []) + .filter((t): t is ChannelTrigger => t.type === 'channel') + .map((t) => ({ + triggerId: t.id, + channel: t.channel, + url: `/hooks/${agentId}/channel/${t.channel}`, + error: errors?.get(t.id) ?? null, + })); + } +} diff --git a/apps/agent-forge/server/triggers/runtime.ts b/apps/agent-forge/server/triggers/runtime.ts index bdd6390d7..3f7b8dec4 100644 --- a/apps/agent-forge/server/triggers/runtime.ts +++ b/apps/agent-forge/server/triggers/runtime.ts @@ -1,16 +1,24 @@ /** - * The per-workspace trigger stack (R08a host + R08c cron scheduler), - * built and started together so the workspace context and the legacy - * app context behave identically: webhook/cron dispatches queue through - * the workspace's own RunManager, and trigger-phase log lines land on - * the same run-log stream RunManager's own entries use. + * The per-workspace trigger stack (R08a host + R08c cron scheduler + + * R08b channel mounts), built and started together so the workspace + * context and the legacy app context behave identically: webhook/cron + * dispatches queue through the workspace's own RunManager, trigger-phase + * log lines land on the same run-log stream RunManager's own entries + * use, and channel triggers mount the SDK's channel handlers per agent. */ import { randomUUID } from 'node:crypto'; +import type { SimpleAgent, TraceExporter } from 'lousho'; +import { fanOutExporters, fileTraceExporter } from 'lousho/traces'; + import type { LogEntry } from '../../shared/wireTypes'; +import { buildRunnableAgent, type StudioRunDeps } from '../buildAgentV2'; +import { observeReporting, type ObserveReporting } from '../observeReporting'; +import { agentTraceDir } from '../traceStore'; import type { AgentRepo } from '../storage/agentRepo'; import type { StudioEventBus } from '../events/eventBus'; import type { RunManager } from '../runRegistry'; +import { ChannelTriggers, type SecretResolver } from './channelTriggers'; import { CronTriggers } from './cronTriggers'; import { TriggerHost } from './triggerHost'; @@ -20,18 +28,73 @@ export interface TriggerRuntimeDeps { repo: AgentRepo; runManager: RunManager; events: StudioEventBus; + /** + * The stores a channel turn's agent builds through (R08b) - the same + * `StudioRunDeps` the workspace's `InProcessAgentHost` gets. + */ + studio: StudioRunDeps; + /** Report channel-turn spans to the observe daemon, like RunManager's runs. */ + observe?: boolean; } export interface TriggerRuntime { triggers: TriggerHost; cron: CronTriggers; + channels: ChannelTriggers; +} + +/** Channel turns trace to the agent's trace folder (+ observe), never the test pane's live feed. */ +function channelExporter(baseDir: string, agentId: string, reporting: ObserveReporting | undefined): TraceExporter { + const spans = [fileTraceExporter({ dir: agentTraceDir(baseDir, agentId) })]; + if (reporting !== undefined) spans.push(reporting.exporter()); + return fanOutExporters(spans) as TraceExporter; +} + +/** `trigger.secrets` values: `{ env }` reads the process env, `{ secret }` the Studio store, strings pass through. */ +function secretResolver(studio: StudioRunDeps): SecretResolver { + return (ref) => { + if (typeof ref === 'string') return ref; + if ('env' in ref) return process.env[ref.env]; + return studio.secretsStore?.getSecret(ref.secret); + }; } export function createTriggerRuntime(deps: TriggerRuntimeDeps): TriggerRuntime { + // Observe reporters are per-agent and reused across remounts, like + // RunManager's own `reporting` map. + const reporting = new Map(); + const reportingFor = (agentId: string): ObserveReporting | undefined => { + if (deps.observe !== true) return undefined; + let reporter = reporting.get(agentId); + if (reporter === undefined) { + reporter = observeReporting({ agentName: agentId, baseDir: deps.baseDir }); + reporting.set(agentId, reporter); + } + return reporter; + }; + // A channel turn builds the working-copy spec through the same + // buildRunnableAgent runs use; its wiring only differs in the sinks - + // spans go to the trace file + observe reporter, agent events to + // nowhere (channel sessions live inside the SDK, not RunManager). + const buildAgent = async (agentId: string): Promise => { + const spec = await deps.repo.readSpec(agentId); + if (spec === undefined) throw new Error(`No spec for agent '${agentId}'`); + const built = await buildRunnableAgent(spec, agentId, deps.studio, { + onEvent: () => {}, + exporter: channelExporter(deps.baseDir, agentId, reportingFor(agentId)), + }); + return built.agent; + }; + const triggers = new TriggerHost({ repo: deps.repo, runManager: deps.runManager, events: deps.events, + onReload: (agentId) => { + // Channel mounts follow every spec reload (R08b); mount() serializes + // per agent and never throws back into the host. + void channels.mount(agentId).catch(() => {}); + }, log: (agentId, message, level = 'info') => deps.runManager.emit('log', agentId, { id: randomUUID(), @@ -42,10 +105,19 @@ export function createTriggerRuntime(deps: TriggerRuntimeDeps): TriggerRuntime { message, } satisfies LogEntry), }); + // Channel mounts must exist before start(): start()'s initial reload of + // every agent fires onReload, which is what mounts them. + const channels = new ChannelTriggers({ + host: triggers, + repo: deps.repo, + buildAgent, + resolve: secretResolver(deps.studio), + log: (agentId, message, level) => triggers.log(agentId, message, level), + }); void triggers.start(); // The cron scheduler fires `cron` triggers through the host on its own // interval; cursors persist under each agent's sidecar dir. const cron = new CronTriggers({ baseDir: deps.baseDir, host: triggers }); cron.start(); - return { triggers, cron }; + return { triggers, cron, channels }; } diff --git a/apps/agent-forge/server/triggers/triggerHost.ts b/apps/agent-forge/server/triggers/triggerHost.ts index 49d72d75b..1e3cea37f 100644 --- a/apps/agent-forge/server/triggers/triggerHost.ts +++ b/apps/agent-forge/server/triggers/triggerHost.ts @@ -67,6 +67,12 @@ export interface TriggerHostDeps { log: (agentId: string, message: string, level?: 'info' | 'warn' | 'error') => void; /** Queued-request cap per agent; the in-flight run does not count. Default 20. */ queueCap?: number; + /** + * Called once per agent at the end of every `reload()` pass over it + * (R08b: the channel mount follows spec changes). Fire-and-forget - + * the hook's own errors are its caller's problem, never the reload's. + */ + onReload?: (agentId: string) => void; } interface AgentTriggers { @@ -140,6 +146,11 @@ export class TriggerHost { } } // A vanished trigger cannot be the in-flight one (its run already started); leave it. + try { + this.deps.onReload?.(id); + } catch { + // A reload hook must never break the host's own reload. + } } } diff --git a/apps/agent-forge/server/workspace/__tests__/middleware.test.ts b/apps/agent-forge/server/workspace/__tests__/middleware.test.ts index 2eb922d4d..581088801 100644 --- a/apps/agent-forge/server/workspace/__tests__/middleware.test.ts +++ b/apps/agent-forge/server/workspace/__tests__/middleware.test.ts @@ -17,6 +17,7 @@ import { PassRunner } from '../../passes/passRunner'; import type { InboxService } from '../../inbox/inboxService'; import type { TriggerHost } from '../../triggers/triggerHost'; import type { CronTriggers } from '../../triggers/cronTriggers'; +import type { ChannelTriggers } from '../../triggers/channelTriggers'; import { requestedWorkspaceId, workspaceMiddleware, workspaceOf } from '../middleware'; const emptyStore: LegacyAgentStore = { @@ -44,6 +45,7 @@ function ctx(id: string): WorkspaceContext { inbox: {} as InboxService, triggers: {} as TriggerHost, cron: {} as CronTriggers, + channels: {} as ChannelTriggers, }; } diff --git a/apps/agent-forge/server/workspace/context.ts b/apps/agent-forge/server/workspace/context.ts index aaa6f6de4..797a02559 100644 --- a/apps/agent-forge/server/workspace/context.ts +++ b/apps/agent-forge/server/workspace/context.ts @@ -40,6 +40,7 @@ import { attachFailureStreak } from '../lifecycle/failureStreak'; import { attachVersionSummaries } from '../passes/versionSummary'; import { attachDraftReviews } from '../passes/draftReview'; import { createTriggerRuntime } from '../triggers/runtime'; +import type { ChannelTriggers } from '../triggers/channelTriggers'; import type { CronTriggers } from '../triggers/cronTriggers'; import type { TriggerHost } from '../triggers/triggerHost'; @@ -107,6 +108,14 @@ export interface WorkspaceContext { * as `triggers` for non-launch contexts until W02a dispose). */ cron: CronTriggers; + /** + * This workspace's channel mounts (triggers/channelTriggers.ts, R08b): + * serves each agent's `channel` triggers at + * `/hooks//channel/` through the SDK's channel + * handlers. Mounted by the host's reload hook; passive like the host + * (no stop() - handlers are per-request lookups). + */ + channels: ChannelTriggers; } /** @@ -247,7 +256,22 @@ export function createWorkspaceContext(ref: WorkspaceRef, services: WorkspaceSer // The trigger host (R08a) + cron scheduler (R08c): webhook/cron // dispatches queue through this workspace's RunManager; trigger-phase // log lines land on the same run-log stream RunManager's entries use. - const { triggers, cron } = createTriggerRuntime({ baseDir: ref.root, repo, runManager: runs, events }); + const { triggers, cron, channels } = createTriggerRuntime({ + baseDir: ref.root, + repo, + runManager: runs, + events, + // Channel turns build through the same stores the run host gets. + studio: { + baseDir: ref.root, + checkpointStore, + approvalStore, + secretsStore: services.secretsStore, + settings: services.settingsStore, + spend: services.spendLog, + }, + ...(services.observe ? { observe: true as const } : {}), + }); return { ref, repo, @@ -264,5 +288,6 @@ export function createWorkspaceContext(ref: WorkspaceRef, services: WorkspaceSer inbox, triggers, cron, + channels, }; } diff --git a/apps/agent-forge/shared/wireTypes.ts b/apps/agent-forge/shared/wireTypes.ts index 352ecf162..c67682004 100644 --- a/apps/agent-forge/shared/wireTypes.ts +++ b/apps/agent-forge/shared/wireTypes.ts @@ -1288,3 +1288,13 @@ export const RATING_LABELS = ['Poor', 'Fair', 'Good', 'Great', 'Excellent'] as c /** Ratings at or below this value raise a `low_rating` inbox alert (R04a). */ export const LOW_RATING_MAX = 2; + +/** `GET /agents/:id/triggers/status` row (R08b): one channel trigger's public URL and mount error. */ +export interface ChannelTriggerStatus { + triggerId: string; + channel: 'telegram' | 'slack' | 'github' | 'teams' | 'discord' | 'http'; + /** Path (origin-relative) to paste into the platform's webhook settings. */ + url: string; + /** Mount failure (`Channel needs `, duplicate kind), `null` when served. */ + error: string | null; +} diff --git a/apps/agent-forge/src/runtime/runtimeClient.ts b/apps/agent-forge/src/runtime/runtimeClient.ts index 6b759b054..40292d2e4 100644 --- a/apps/agent-forge/src/runtime/runtimeClient.ts +++ b/apps/agent-forge/src/runtime/runtimeClient.ts @@ -37,6 +37,7 @@ import type { InboxItem, McpProbeRequest, McpProbeResponse, + ChannelTriggerStatus, PassSettings, PassSettingsPatch, PassSpendSummary, @@ -205,6 +206,11 @@ class RuntimeClient implements AgentApi { return this.request(`/agents/${encodeURIComponent(agentId)}/requirements`, { method: 'GET' }); } + /** The agent's channel-trigger mount rows (`GET /agents/:id/triggers/status`, R08b) - URL + mount error per channel trigger. */ + async channelTriggerStatus(agentId: string): Promise { + return this.request(`/agents/${encodeURIComponent(agentId)}/triggers/status`, { method: 'GET' }); + } + /** Writes `.lousho/agents/.yaml` from a spec v2 working copy (`PUT /agents/:id` with a `{ spec }` body). */ async saveAgent(agentId: string, spec: AgentSpecV2): Promise { await this.request(`/agents/${encodeURIComponent(agentId)}`, { diff --git a/apps/agent-forge/src/screens/agentDetail/AgentOverviewTab.tsx b/apps/agent-forge/src/screens/agentDetail/AgentOverviewTab.tsx index 3d8b4e0da..b44038fc6 100644 --- a/apps/agent-forge/src/screens/agentDetail/AgentOverviewTab.tsx +++ b/apps/agent-forge/src/screens/agentDetail/AgentOverviewTab.tsx @@ -1,8 +1,8 @@ import { ChevronRight, TriangleAlert } from 'lucide-react'; -import type { ReactNode } from 'react'; +import { useEffect, useState, type ReactNode } from 'react'; import type { AgentSpecV2 } from 'lousho/spec'; -import type { TraceSummaryPayload } from '../../../shared/wireTypes'; +import type { ChannelTriggerStatus, TraceSummaryPayload } from '../../../shared/wireTypes'; import { agentTabPath, BUILT_AGENT_TABS } from '../agents/agentTabs'; import { fmtAge, fmtDateTime, fmtDuration } from '../../observe/format'; import { Card } from '../../ui/Card'; @@ -10,6 +10,7 @@ import { SkeletonRows } from '../../ui/Skeleton'; import { StatusDot } from '../../ui/StatusBadge'; import { statusKey } from '../../ui/status'; import { useAgentDetail } from './AgentDetailContext'; +import { runtimeClient } from '../../runtime/runtimeClient'; import { capabilityRows, missingRequirements, @@ -28,6 +29,58 @@ const INSTRUCTIONS_LINES = 12; /** One `spec.triggers` entry, narrowed off the spec. */ type TriggerSpec = NonNullable[number]; type WebhookTrigger = Extract; +type ChannelTrigger = Extract; + +/** The public URL a channel trigger listens on: `/hooks//channel/` (R08b). */ +function channelUrl(agentId: string, trigger: ChannelTrigger): string { + return `${window.location.origin}/hooks/${agentId}/channel/${trigger.channel}`; +} + +/** A channel trigger's webhook-settings URL with Copy, plus its mount error when the host could not serve it. */ +function ChannelUrl({ + agentId, + trigger, + status, +}: { + agentId: string; + trigger: ChannelTrigger; + status: ChannelTriggerStatus | undefined; +}): JSX.Element { + const url = channelUrl(agentId, trigger); + return ( + + {url} + + {status?.error !== null && status?.error !== undefined && ( + {status.error} + )} + + ); +} + +/** The agent's channel-trigger mount rows (`GET /agents/:id/triggers/status`), keyed by trigger id; empty until loaded. */ +function useChannelStatuses(agentId: string): Map { + const [statuses, setStatuses] = useState([]); + useEffect(() => { + let cancelled = false; + runtimeClient + .channelTriggerStatus(agentId) + .then((rows) => { + if (!cancelled) setStatuses(rows); + }) + .catch(() => {}); + return () => { + cancelled = true; + }; + }, [agentId]); + return new Map(statuses.map((row) => [row.triggerId, row])); +} /** The public URL a webhook trigger listens on: `/hooks/` + the trigger's own `path`. */ function webhookUrl(agentId: string, trigger: WebhookTrigger): string { @@ -105,8 +158,9 @@ function CapabilitiesCard({ rows }: { rows: CapabilityRow[] }): JSX.Element { ); } -/** The Triggers card: one icon + line per trigger (webhooks also show their URL), or the empty copy. */ +/** The Triggers card: one icon + line per trigger (webhooks and channels also show their URL), or the empty copy. */ function TriggersCard({ agentId, triggers }: { agentId: string; triggers: TriggerSpec[] }): JSX.Element { + const channelStatuses = useChannelStatuses(agentId); return ( {triggers.length > 0 ? ( @@ -118,6 +172,9 @@ function TriggersCard({ agentId, triggers }: { agentId: string; triggers: Trigge ); } diff --git a/apps/agent-forge/src/screens/agentDetail/__tests__/AgentDetailRoute.test.tsx b/apps/agent-forge/src/screens/agentDetail/__tests__/AgentDetailRoute.test.tsx index e9cc027ad..47e3188f7 100644 --- a/apps/agent-forge/src/screens/agentDetail/__tests__/AgentDetailRoute.test.tsx +++ b/apps/agent-forge/src/screens/agentDetail/__tests__/AgentDetailRoute.test.tsx @@ -26,7 +26,12 @@ const mocks = vi.hoisted(() => { }); vi.mock('../../../runtime/runtimeClient', () => ({ - runtimeClient: { loadAgent: mocks.loadAgent, listTraces: mocks.listTraces, getDraft: vi.fn(async () => undefined) }, + runtimeClient: { + loadAgent: mocks.loadAgent, + listTraces: mocks.listTraces, + getDraft: vi.fn(async () => undefined), + channelTriggerStatus: vi.fn(async () => []), + }, RuntimeApiError: mocks.MockRuntimeApiError, })); diff --git a/apps/agent-forge/src/screens/agentDetail/__tests__/AgentOverviewTab.test.tsx b/apps/agent-forge/src/screens/agentDetail/__tests__/AgentOverviewTab.test.tsx index 75c79bf04..e144fb097 100644 --- a/apps/agent-forge/src/screens/agentDetail/__tests__/AgentOverviewTab.test.tsx +++ b/apps/agent-forge/src/screens/agentDetail/__tests__/AgentOverviewTab.test.tsx @@ -5,10 +5,10 @@ import type { AgentRecord, Requirement, TraceSummaryPayload } from '../../../../ import { render } from '../../../ui/__tests__/render'; import type { AgentSpecV2 } from 'lousho/spec'; -const mocks = vi.hoisted(() => ({ listTraces: vi.fn() })); +const mocks = vi.hoisted(() => ({ listTraces: vi.fn(), channelTriggerStatus: vi.fn() })); vi.mock('../../../runtime/runtimeClient', () => ({ - runtimeClient: { listTraces: mocks.listTraces }, + runtimeClient: { listTraces: mocks.listTraces, channelTriggerStatus: mocks.channelTriggerStatus }, })); import { AgentDetailContext } from '../AgentDetailContext'; @@ -101,6 +101,8 @@ function rowText(cardEl: HTMLElement, label: string): string { beforeEach(() => { mocks.listTraces.mockReset(); mocks.listTraces.mockResolvedValue([]); + mocks.channelTriggerStatus.mockReset(); + mocks.channelTriggerStatus.mockResolvedValue([]); }); describe('AgentOverviewTab', () => { @@ -177,6 +179,34 @@ describe('AgentOverviewTab', () => { ); }); + it('shows a channel trigger URL, its mount error, and the Runs/Traces note (R08b)', async () => { + mocks.channelTriggerStatus.mockResolvedValue([ + { + triggerId: 't4', + channel: 'github', + url: '/hooks/support-bot/channel/github', + error: 'Channel github needs token', + }, + ]); + const triggers: AgentSpecV2['triggers'] = [ + { id: 't4', type: 'channel', channel: 'github', secrets: {} }, + ]; + const host = renderTab(record({ spec: { triggers } })); + await flush(); + await flush(); + + expect(mocks.channelTriggerStatus).toHaveBeenCalledWith('support-bot'); + const triggersCard = card(host, 'Triggers'); + const item = triggersCard.querySelector('.agent-overview-list li')!; + expect(item.querySelector('.agent-overview-webhook code')?.textContent).toBe( + `${window.location.origin}/hooks/support-bot/channel/github`, + ); + expect(item.querySelector('.agent-overview-channel-error')?.textContent).toBe( + 'Channel github needs token', + ); + expect(triggersCard.textContent).toContain('Channel runs show up in Runs and Traces.'); + }); + it('shows permission-rule counts by action in the Safety card', async () => { const host = renderTab( record({ diff --git a/apps/agent-forge/src/screens/agentDetail/agentOverview.css b/apps/agent-forge/src/screens/agentDetail/agentOverview.css index 1b4f2690b..bc7b16937 100644 --- a/apps/agent-forge/src/screens/agentDetail/agentOverview.css +++ b/apps/agent-forge/src/screens/agentDetail/agentOverview.css @@ -160,3 +160,8 @@ gap: 2px; } } + +.agent-overview-channel-error { + color: var(--danger); + font-size: 12px; +} From ef4d3ca7b8d85c4817ecd57c82fbe2d03ecc8ea3 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Fri, 13 Nov 2026 02:02:44 +0300 Subject: [PATCH 2/3] feat(studio): R03b evaluate abort listeners on webhook events The webhook route now runs applyAbortListeners after signature verification and before the filter/dispatch (the channel path already did): a `when: true` listener cancels matching queued/running runs of the watched trigger with reason abort_listener:, and the 202 response lists the cancelled runKeys. Stopped runs end 'stopped', so the L12 failure streak never counts them. Unit coverage for scopes, strict-true when, expression-error skip and trigger matching, plus two route tests (cancel on matching PR, no cancel on a different one). Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../server/__tests__/abortListeners.test.ts | 153 ++++++++++++++++++ .../server/__tests__/hooksRoute.test.ts | 105 +++++++++++- .../server/triggers/webhookRoute.ts | 27 +++- docs/triggers.md | 6 +- llms-full.txt | 6 +- 5 files changed, 289 insertions(+), 8 deletions(-) create mode 100644 apps/agent-forge/server/__tests__/abortListeners.test.ts diff --git a/apps/agent-forge/server/__tests__/abortListeners.test.ts b/apps/agent-forge/server/__tests__/abortListeners.test.ts new file mode 100644 index 000000000..36bc25a23 --- /dev/null +++ b/apps/agent-forge/server/__tests__/abortListeners.test.ts @@ -0,0 +1,153 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { AgentSpecV2 } from 'lousho'; + +import { applyAbortListeners, type AbortListenerInput } from '../triggers/abortListeners'; +import type { HostRun } from '../triggers/triggerHost'; + +const AGENT = 'agent1'; + +function run(overrides: Partial = {}): HostRun { + return { + runKey: 't1:r1', + agentId: AGENT, + triggerId: 't1', + event: { pr: 7, action: 'opened' }, + input: 'review pr 7', + status: 'queued', + enqueuedAt: '2026-11-12T00:00:00.000Z', + ...overrides, + }; +} + +function specWith(listeners: NonNullable): AgentSpecV2 { + return { + version: 2, + name: 'reviewer', + model: 'anthropic/claude-sonnet-5-5', + instructions: 'Review.', + triggers: [{ id: 't1', type: 'webhook', path: '/review', promptTemplate: 'Review {{event.pr}}' }], + abortListeners: listeners, + } as AgentSpecV2; +} + +/** A fake host: `runs` are what runsOf returns; `cancel` records calls, reports the run's live status and drops cancelled queued runs like the real host. */ +function fakeHost(runs: HostRun[]) { + const cancelled: Array<{ runKey: string; reason: string }> = []; + const host = { + runsOf: () => runs, + cancel: (runKey: string, reason: string) => { + const found = runs.find((r) => r.runKey === runKey); + if (found === undefined) return undefined; + cancelled.push({ runKey, reason }); + if (found.status === 'queued') runs.splice(runs.indexOf(found), 1); + return found.status; + }, + }; + return { host, cancelled }; +} + +function inputFor( + spec: AgentSpecV2, + host: AbortListenerInput['host'], + event: unknown, + log = vi.fn() +): { input: AbortListenerInput; log: ReturnType } { + return { + input: { spec, agentId: AGENT, triggerId: 't1', event, host, log }, + log, + }; +} + +describe('applyAbortListeners', () => { + const closed = { pr: 7, action: 'closed' }; + const listener = { + id: 'l1', + triggerId: 't1', + when: "event.action == 'closed'", + match: 'run.event.pr == event.pr', + scope: 'both' as const, + }; + + it('cancels the matching queued run and reports the outcome', () => { + const { host, cancelled } = fakeHost([run()]); + const { input } = inputFor(specWith([listener]), host, closed); + + const outcomes = applyAbortListeners(input); + + expect(cancelled).toEqual([{ runKey: 't1:r1', reason: 'abort_listener:l1' }]); + expect(outcomes).toEqual([{ listenerId: 'l1', cancelled: [{ runKey: 't1:r1', status: 'queued' }] }]); + }); + + it('leaves a run for a different PR alone (match scope)', () => { + const { host, cancelled } = fakeHost([run({ event: { pr: 8, action: 'opened' } })]); + const { input } = inputFor(specWith([listener]), host, closed); + + expect(applyAbortListeners(input)).toEqual([]); + expect(cancelled).toEqual([]); + }); + + it('scope queued never stops a running run; scope running never drops a queued one', () => { + const running = run({ runKey: 't1:r2', status: 'running' }); + const queuedOnly = { ...listener, id: 'lq', scope: 'queued' as const }; + const { host: h1, cancelled: c1 } = fakeHost([running]); + expect(applyAbortListeners(inputFor(specWith([queuedOnly]), h1, closed).input)).toEqual([]); + expect(c1).toEqual([]); + + const queued = run({ runKey: 't1:r3' }); + const runningOnly = { ...listener, id: 'lr', scope: 'running' as const }; + const { host: h2, cancelled: c2 } = fakeHost([queued]); + expect(applyAbortListeners(inputFor(specWith([runningOnly]), h2, closed).input)).toEqual([]); + expect(c2).toEqual([]); + }); + + it('only cancels runs of the watched trigger', () => { + const other = run({ runKey: 't2:r1', triggerId: 't2', event: { pr: 7 } }); + const { host, cancelled } = fakeHost([other]); + const { input } = inputFor(specWith([listener]), host, closed); + + expect(applyAbortListeners(input)).toEqual([]); + expect(cancelled).toEqual([]); + }); + + it('does not fire when `when` is not strictly true', () => { + const truthy = { ...listener, when: "'yes'" }; + const { host, cancelled } = fakeHost([run()]); + const { input } = inputFor(specWith([truthy]), host, closed); + + expect(applyAbortListeners(input)).toEqual([]); + expect(cancelled).toEqual([]); + }); + + it('logs and skips a listener whose expression fails at run time', () => { + const bad = { ...listener, when: 'event..broken' }; + const { host, cancelled } = fakeHost([run()]); + const { input, log } = inputFor(specWith([bad]), host, closed); + + expect(applyAbortListeners(input)).toEqual([]); + expect(cancelled).toEqual([]); + expect(log).toHaveBeenCalledWith(AGENT, expect.stringContaining('Abort listener l1 failed:'), 'warn'); + }); + + it('evaluates every listener independently and logs each cancel', () => { + const second = { ...listener, id: 'l2', when: "event.action == 'closed' && event.pr == 7" }; + const { host } = fakeHost([run()]); + const { input, log } = inputFor(specWith([listener, second]), host, closed); + + const outcomes = applyAbortListeners(input); + + // The first listener's cancel drops the queued run, so the second + // listener finds no candidates and reports nothing. + expect(outcomes.map((o) => o.listenerId)).toEqual(['l1']); + expect(log).toHaveBeenCalledWith(AGENT, 'Run cancelled by abort listener l1: event.action == \'closed\'', 'warn'); + }); + + it('does nothing for an agent without listeners', () => { + const { host, cancelled } = fakeHost([run()]); + const spec = specWith([]); + delete spec.abortListeners; + const { input } = inputFor(spec, host, closed); + + expect(applyAbortListeners(input)).toEqual([]); + expect(cancelled).toEqual([]); + }); +}); diff --git a/apps/agent-forge/server/__tests__/hooksRoute.test.ts b/apps/agent-forge/server/__tests__/hooksRoute.test.ts index 1ec840bd5..7b9ad411c 100644 --- a/apps/agent-forge/server/__tests__/hooksRoute.test.ts +++ b/apps/agent-forge/server/__tests__/hooksRoute.test.ts @@ -19,7 +19,7 @@ import { IdleVersioner } from '../storage/idleVersioner'; import { StudioEventBus } from '../events/eventBus'; import { PassRunner } from '../passes/passRunner'; import type { InboxService } from '../inbox/inboxService'; -import type { TriggerHost } from '../triggers/triggerHost'; +import type { HostRun, TriggerHost } from '../triggers/triggerHost'; import type { ChannelTriggers } from '../triggers/channelTriggers'; import type { CronTriggers } from '../triggers/cronTriggers'; import type { WorkspaceContext } from '../workspace/context'; @@ -71,13 +71,18 @@ function sign(body: string, secret = SECRET): string { function appWith( triggers: TriggerSpec[], - dispatch: Mock = vi.fn(() => ({ ok: true as const, runKey: 'rk1', started: true })) + dispatch: Mock = vi.fn(() => ({ ok: true as const, runKey: 'rk1', started: true })), + hostExtra: Record = {} ) { const baseDir = `/tmp/hooks-${Math.random().toString(36).slice(2)}`; const repo = new AgentRepo(baseDir); const host = { triggersOf: () => triggers, dispatch, + runsOf: () => [], + cancel: () => undefined, + log: () => {}, + ...hostExtra, } as unknown as TriggerHost; const workspace: WorkspaceContext = { ref: workspaceRefFor(baseDir), @@ -100,7 +105,7 @@ function appWith( const secrets = new SecretsStore(baseDir); secrets.setSecret('wh', SECRET); const app = createApp({ workspace, baseDir, secretsStore: secrets }); - return { app: app as Express, dispatch }; + return { app: app as Express, dispatch, repo }; } describe('POST /hooks/:agentId/', () => { @@ -114,7 +119,7 @@ describe('POST /hooks/:agentId/', () => { .send(body); expect(res.status).toBe(202); - expect(res.body).toEqual({ queued: true, runKey: 'rk1', started: true }); + expect(res.body).toEqual({ queued: true, runKey: 'rk1', started: true, cancelled: [] }); expect(dispatch).toHaveBeenCalledWith({ agentId: AGENT, triggerId: 't1', @@ -209,4 +214,96 @@ describe('POST /hooks/:agentId/', () => { expect(res.status).toBe(429); expect(res.body.code).toBe('LOUSHO_TRIGGER_QUEUE_FULL'); }); + + it('runs abort listeners before dispatch and reports the cancelled runKeys (R03b)', async () => { + // The queued run event 1 started; event 2's `when`/`match` pair it. + const runs: HostRun[] = [ + { + runKey: 't2:run1', + agentId: AGENT, + triggerId: 't2', + event: { pr: 7, action: 'opened' }, + input: 'Got {"pr":7,"action":"opened"}', + status: 'queued', + enqueuedAt: new Date().toISOString(), + }, + ]; + const cancelled: Array<{ runKey: string; reason: string }> = []; + const { app, repo } = appWith([OPEN], undefined, { + runsOf: () => runs, + cancel: (runKey: string, reason: string) => { + const found = runs.find((r) => r.runKey === runKey); + if (found === undefined) return undefined; + cancelled.push({ runKey, reason }); + runs.splice(runs.indexOf(found), 1); + return found.status; + }, + }); + await repo.writeSpec(AGENT, { + version: 2, + name: 'agent1', + model: 'anthropic/claude-sonnet-5-5', + instructions: 'Help.', + triggers: [{ id: 't2', type: 'webhook', path: '/open', promptTemplate: 'Got {{event.body}}' }], + abortListeners: [ + { + id: 'l1', + triggerId: 't2', + when: "event.action == 'closed'", + match: 'run.event.pr == event.pr', + scope: 'both', + }, + ], + }); + + const res = await request(app) + .post(`/hooks/${AGENT}/hooks/open`) + .set('host', '127.0.0.1:4750') + .send(JSON.stringify({ pr: 7, action: 'closed' })); + + expect(res.status).toBe(202); + expect(res.body.cancelled).toEqual(['t2:run1']); + expect(cancelled).toEqual([{ runKey: 't2:run1', reason: 'abort_listener:l1' }]); + }); + + it('does not cancel a run for a different PR (R03b match scope)', async () => { + const runs: HostRun[] = [ + { + runKey: 't2:run1', + agentId: AGENT, + triggerId: 't2', + event: { pr: 8, action: 'opened' }, + input: 'x', + status: 'running', + enqueuedAt: new Date().toISOString(), + }, + ]; + const cancel = vi.fn(() => 'running' as const); + const { app, repo } = appWith([OPEN], undefined, { runsOf: () => runs, cancel }); + await repo.writeSpec(AGENT, { + version: 2, + name: 'agent1', + model: 'anthropic/claude-sonnet-5-5', + instructions: 'Help.', + triggers: [{ id: 't2', type: 'webhook', path: '/open', promptTemplate: 'Got {{event.body}}' }], + abortListeners: [ + { + id: 'l1', + triggerId: 't2', + when: "event.action == 'closed'", + match: 'run.event.pr == event.pr', + scope: 'both', + }, + ], + }); + + const res = await request(app) + .post(`/hooks/${AGENT}/hooks/open`) + .set('host', '127.0.0.1:4750') + .send(JSON.stringify({ pr: 7, action: 'closed' })); + + expect(res.status).toBe(202); + expect(res.body.cancelled).toEqual([]); + expect(cancel).not.toHaveBeenCalled(); + }); }); diff --git a/apps/agent-forge/server/triggers/webhookRoute.ts b/apps/agent-forge/server/triggers/webhookRoute.ts index 6ec8ef881..7525b2aa2 100644 --- a/apps/agent-forge/server/triggers/webhookRoute.ts +++ b/apps/agent-forge/server/triggers/webhookRoute.ts @@ -1,10 +1,13 @@ /** * Webhook trigger routes (R08a, PLAN §5): * - * POST /hooks/:agentId/ -> 202 { queued, runKey, started } + * POST /hooks/:agentId/ -> 202 { queued, runKey, started, cancelled } * | 202 { skipped: 'filter' } * | 400/401/403/404/422/429 * + * `cancelled` lists the runKeys abort listeners (R03b) stopped for this + * event, evaluated after signature verification and before the filter. + * * The request path after `/hooks/:agentId/` must equal a webhook * trigger's `path` (the canvas default is `/hooks/`, so the * full URL is `/hooks//hooks/`). Raw bytes are read @@ -31,6 +34,7 @@ import { checkWebhookAuth } from 'lousho/triggers'; import { isLoopbackHost } from '../auth'; import { asyncRoute, sendError } from '../http/routeUtils'; import { workspaceOf } from '../workspace/middleware'; +import { applyAbortListeners } from './abortListeners'; import { renderTemplate } from './template'; type WebhookTrigger = Extract; @@ -114,6 +118,25 @@ export function registerWebhookRoutes(app: Express, deps: WebhookRouteDeps): voi if (!(await checkSignature(req, res, trigger, rawBody, deps.resolveSecret))) return; const event = eventOf(rawBody); + + // Abort listeners run after signature verification but before the + // filter and the dispatch (R03b): a `when: true` listener cancels + // matching queued/running runs of the trigger it watches. The run + // this event is about to queue does not exist yet, so it can never + // cancel itself. + const spec = await ws.repo.readSpec(agentId); + const cancelled = + spec === undefined + ? [] + : applyAbortListeners({ + spec, + agentId, + triggerId: trigger.id, + event, + host: ws.triggers, + log: (id, message, level) => ws.triggers.log(id, message, level), + }).flatMap((outcome) => outcome.cancelled.map((c) => c.runKey)); + if (trigger.filter !== undefined) { const keep = evaluateSafeExpression(trigger.filter, { event }); if (keep !== true) { @@ -141,7 +164,7 @@ export function registerWebhookRoutes(app: Express, deps: WebhookRouteDeps): voi sendError(res, 404, `No webhook trigger '${trigger.id}' for agent '${agentId}'`); return; } - res.status(202).json({ queued: true, runKey: result.runKey, started: result.started }); + res.status(202).json({ queued: true, runKey: result.runKey, started: result.started, cancelled }); }) ); } diff --git a/docs/triggers.md b/docs/triggers.md index 5963ad610..aa82b2629 100644 --- a/docs/triggers.md +++ b/docs/triggers.md @@ -150,4 +150,8 @@ abortListeners: `when` and `match` use the same safe expression grammar as flow conditions (see [Flows](./flows.md#flow-expressions)); variable references are checked when the spec parses and unknown names resolve to `undefined` at runtime. -At most 20 listeners are allowed per spec. +At most 20 listeners are allowed per spec. Studio's trigger host evaluates +them on every inbound webhook and channel event, before the event's own +filter and dispatch, so the delivering run can never cancel itself; a +cancelled run logs `Run cancelled by abort listener ` and its webhook +response lists the cancelled run keys under `cancelled`. diff --git a/llms-full.txt b/llms-full.txt index 7517c98e0..06dd87b07 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -23437,7 +23437,11 @@ abortListeners: `when` and `match` use the same safe expression grammar as flow conditions (see [Flows](https://github.com/LinuxDevil/lousho/blob/main/docs/flows.md#flow-expressions)); variable references are checked when the spec parses and unknown names resolve to `undefined` at runtime. -At most 20 listeners are allowed per spec. +At most 20 listeners are allowed per spec. Studio's trigger host evaluates +them on every inbound webhook and channel event, before the event's own +filter and dispatch, so the delivering run can never cancel itself; a +cancelled run logs `Run cancelled by abort listener ` and its webhook +response lists the cancelled run keys under `cancelled`. # Troubleshooting From f4b0e4976e72a9a7d3d47746e3ab66fc224ecde9 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Fri, 13 Nov 2026 02:35:06 +0300 Subject: [PATCH 3/3] feat(studio): R09 trigger host respects meta.enabled TriggerHost caches meta.enabled per agent (refreshed on every reload, which agent.changed 'meta' already triggers), rejects dispatch with 'disabled' for turned-off agents, and drops + logs queued runs on the enabled->disabled transition while a running run finishes. Webhook answers 409 LOUSHO_AGENT_DISABLED right after auth (no event processing beyond it); channels answer 200 { ok, skipped: 'disabled' } before the SDK handler; cron's existing disabled branch logs the contract skip line and the cursor still advances; chain dispatches inherit the same reason. Manual runs are untouched. UI: the Overview Triggers card shows the off note (auto_failure_streak gets the streak copy) and non-chat trigger nodes dim with node-trigger-off + the contract tooltip. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../server/__tests__/channelTriggers.test.ts | 23 ++++++- .../server/__tests__/cronTriggers.test.ts | 17 +++++ .../server/__tests__/hooksRoute.test.ts | 29 +++++++++ .../server/__tests__/triggerHost.test.ts | 64 ++++++++++++++++++- .../server/triggers/channelTriggers.ts | 8 +++ .../server/triggers/triggerHost.ts | 36 ++++++++++- .../server/triggers/webhookRoute.ts | 13 +++- apps/agent-forge/src/canvas/NodeCard.tsx | 11 +++- .../agent-forge/src/components/CanvasArea.tsx | 13 +++- .../screens/agentDetail/AgentOverviewTab.tsx | 29 ++++++++- .../__tests__/AgentOverviewTab.test.tsx | 41 +++++++++++- .../src/screens/agentDetail/agentDetail.css | 5 +- .../src/screens/agentDetail/agentOverview.css | 1 + .../agent-forge/src/screens/agents/agents.css | 4 ++ 14 files changed, 275 insertions(+), 19 deletions(-) diff --git a/apps/agent-forge/server/__tests__/channelTriggers.test.ts b/apps/agent-forge/server/__tests__/channelTriggers.test.ts index 6cb3c97f9..f860bc703 100644 --- a/apps/agent-forge/server/__tests__/channelTriggers.test.ts +++ b/apps/agent-forge/server/__tests__/channelTriggers.test.ts @@ -84,13 +84,14 @@ interface Harness { host: { runsOf: ReturnType>; cancel: ReturnType>; + isEnabled: ReturnType>; }; baseDir: string; } async function harness( spec: AgentSpecV2, - options: { secrets?: Record; agent?: SimpleAgent; runs?: HostRun[] } = {} + options: { secrets?: Record; agent?: SimpleAgent; runs?: HostRun[]; enabled?: boolean } = {} ): Promise { const baseDir = tmpDir(); const repo = new AgentRepo(baseDir); @@ -102,6 +103,7 @@ async function harness( const host = { runsOf: vi.fn<[agentId: string], HostRun[]>(() => options.runs ?? []), cancel: vi.fn<[runKey: string, reason: string], 'queued' | 'running' | undefined>(() => 'queued'), + isEnabled: vi.fn<[agentId: string], boolean>(() => options.enabled ?? true), }; const channels = new ChannelTriggers({ host: host as unknown as TriggerHost, @@ -217,6 +219,23 @@ describe('ChannelTriggers.mount / status', () => { ]); }); + it('answers 200 skipped disabled for a turned-off agent and never calls the SDK handler (R09)', async () => { + const agent = createAgent({ provider: mockModel(['Pong']) }); + const { app } = await harness(specWith([HTTP_TRIGGER]), { + agent: agent as unknown as SimpleAgent, + enabled: false, + }); + + const res = await request(app) + .post(`/hooks/${AGENT}/channel/http`) + .set('content-type', 'application/json') + .send({ sessionKey: 'u1', input: 'ping' }); + + expect(res.status).toBe(200); + // The SDK handler's `{ text }` reply never happens - platforms see the skip. + expect(res.body).toEqual({ ok: true, skipped: 'disabled' }); + }); + it('answers 404-shaped fallthrough for a channel path with no mounted trigger', async () => { const { app } = await harness(specWith([HTTP_TRIGGER])); const res = await request(app).post(`/hooks/${AGENT}/channel/telegram`).send('{}'); @@ -413,7 +432,7 @@ describe('channel route inside createApp', () => { const repo = new AgentRepo(baseDir); await repo.writeSpec(AGENT, specWith([HTTP_TRIGGER])); const channels = new ChannelTriggers({ - host: { runsOf: () => [], cancel: () => undefined } as unknown as TriggerHost, + host: { runsOf: () => [], cancel: () => undefined, isEnabled: () => true } as unknown as TriggerHost, repo, buildAgent: async () => createAgent({ provider: mockModel(['ViaApp']) }) as unknown as SimpleAgent, resolve: resolver({}), diff --git a/apps/agent-forge/server/__tests__/cronTriggers.test.ts b/apps/agent-forge/server/__tests__/cronTriggers.test.ts index 607ee88ff..08ca2d6be 100644 --- a/apps/agent-forge/server/__tests__/cronTriggers.test.ts +++ b/apps/agent-forge/server/__tests__/cronTriggers.test.ts @@ -173,6 +173,23 @@ describe('CronTriggers', () => { expect(log).toHaveBeenCalledWith('a', 'Cron trigger t1 fired'); }); + it('skips a disabled agent, logs the contract line and still advances the cursor (R09)', async () => { + const { cron, dispatch, log } = make({ a: [CRON] }, { ok: false, reason: 'disabled' }); + await cron.tick(); + now = new Date('2026-01-01T10:05:00.000Z'); + await cron.tick(); + + expect(dispatch).toHaveBeenCalledTimes(1); + expect(log).toHaveBeenCalledWith('a', 'Cron trigger t1 skipped: the agent is turned off', 'warn'); + // The claim moved the cursor before the dispatch, so re-enabling later + // never bursts the missed window. + expect(readState(baseDir, 'a').t1.lastRunAt).toBe('2026-01-01T10:05:00.000Z'); + + now = new Date('2026-01-01T10:10:00.000Z'); + await cron.tick(); + expect(dispatch).toHaveBeenCalledTimes(2); + }); + it('deletes state entries for triggers removed from the spec', async () => { const specs: Record = { a: [CRON, CRON2] }; const { cron, triggersOf } = make(specs); diff --git a/apps/agent-forge/server/__tests__/hooksRoute.test.ts b/apps/agent-forge/server/__tests__/hooksRoute.test.ts index 7b9ad411c..bbc0881d7 100644 --- a/apps/agent-forge/server/__tests__/hooksRoute.test.ts +++ b/apps/agent-forge/server/__tests__/hooksRoute.test.ts @@ -81,6 +81,7 @@ function appWith( dispatch, runsOf: () => [], cancel: () => undefined, + isEnabled: () => true, log: () => {}, ...hostExtra, } as unknown as TriggerHost; @@ -306,4 +307,32 @@ describe('POST /hooks/:agentId/', () => { expect(res.body.cancelled).toEqual([]); expect(cancel).not.toHaveBeenCalled(); }); + + it('answers 409 LOUSHO_AGENT_DISABLED for a turned-off agent after auth (R09)', async () => { + const { app, dispatch } = appWith([SIGNED], undefined, { isEnabled: () => false }); + const body = JSON.stringify({ action: 'deploy' }); + const res = await request(app) + .post(`/hooks/${AGENT}/hooks/t1`) + .set('content-type', 'application/json') + .set('x-signature-256', sign(body)) + .send(body); + + expect(res.status).toBe(409); + expect(res.body.code).toBe('LOUSHO_AGENT_DISABLED'); + expect(res.body.error).toBe('Agent is turned off'); + expect(dispatch).not.toHaveBeenCalled(); + }); + + it('still rejects an unauthenticated call on a disabled agent before the enabled gate (R09)', async () => { + // Auth runs first: an unsigned remote call gets 403, never the 409 that + // would confirm the agent exists. + const { app } = appWith([OPEN], undefined, { isEnabled: () => false }); + const res = await request(app) + .post(`/hooks/${AGENT}/hooks/open`) + .set('host', 'staging.example.com') + .send('x'); + + expect(res.status).toBe(403); + expect(res.body.error).toBe('This webhook has no secret, so it only accepts local calls.'); + }); }); diff --git a/apps/agent-forge/server/__tests__/triggerHost.test.ts b/apps/agent-forge/server/__tests__/triggerHost.test.ts index 697787c11..d01ad9458 100644 --- a/apps/agent-forge/server/__tests__/triggerHost.test.ts +++ b/apps/agent-forge/server/__tests__/triggerHost.test.ts @@ -46,21 +46,26 @@ class FakeRuns extends EventEmitter { } } -function repoWith(specs: Record): AgentRepo { +function repoWith(specs: Record, metas: Record = {}): AgentRepo { return { listIds: () => Object.keys(specs), readSpec: async (id: string) => specs[id], + readMeta: async (id: string) => ({ enabled: metas[id]?.enabled ?? true }) as never, } as unknown as AgentRepo; } const made: TriggerHost[] = []; -function makeHost(specs: Record, queueCap?: number) { +function makeHost( + specs: Record, + queueCap?: number, + metas: Record = {} +) { const runs = new FakeRuns(); const events = new StudioEventBus(); const logs: string[] = []; const host = new TriggerHost({ - repo: repoWith(specs), + repo: repoWith(specs, metas), runManager: runs as unknown as RunManager, events, log: (_agentId, message) => logs.push(message), @@ -243,4 +248,57 @@ describe('TriggerHost', () => { expect(d2.ok).toBe(true); expect(host.cancel('missing:key', 'x')).toBeUndefined(); }); + + it('rejects dispatches for a disabled agent and isEnabled reflects meta (R09)', async () => { + const metas = { a: { enabled: false } }; + const { host, runs } = makeHost({ a: specWith([WEBHOOK]) }, undefined, metas); + await host.start(); + + expect(host.isEnabled('a')).toBe(false); + expect(host.isEnabled('missing')).toBe(true); + expect(host.dispatch({ agentId: 'a', triggerId: 't1', input: 'x', event: {} })).toEqual({ + ok: false, + reason: 'disabled', + }); + expect(host.runsOf('a')).toHaveLength(0); + expect(runs.calls).toHaveLength(0); + }); + + it('drops queued runs and logs the count when the agent turns off; the running run finishes (R09)', async () => { + const metas = { a: { enabled: true } }; + const { host, runs, logs } = makeHost({ a: specWith([WEBHOOK, CRON]) }, undefined, metas); + await host.start(); + + host.dispatch({ agentId: 'a', triggerId: 't1', input: 'one', event: {} }); + host.dispatch({ agentId: 'a', triggerId: 'c1', input: 'two', event: {} }); + host.dispatch({ agentId: 'a', triggerId: 'c1', input: 'three', event: {} }); + await flush(); + expect(host.runsOf('a').map((r) => r.status)).toEqual(['running', 'queued', 'queued']); + + metas.a.enabled = false; + await host.reload('a'); + + expect(host.isEnabled('a')).toBe(false); + // Only the in-flight run survives; both queued requests are gone. + expect(host.runsOf('a').map((r) => r.status)).toEqual(['running']); + expect(logs).toContain('Dropped 2 queued runs: the agent was turned off'); + expect(runs.calls).toHaveLength(1); + + // The running run still ends normally and reports its RunEnd. + const ends: RunEnd[] = []; + host.onRunEnd((end) => ends.push(end)); + runs.emitStatus('a', 'done', { resultText: 'ok' }); + expect(ends).toHaveLength(1); + expect(ends[0].status).toBe('done'); + }); + + it('does not re-drop or re-log while the agent stays disabled (R09)', async () => { + const metas = { a: { enabled: false } }; + const { host, logs } = makeHost({ a: specWith([WEBHOOK]) }, undefined, metas); + await host.start(); + await host.reload('a'); + await host.reload('a'); + + expect(logs.filter((m) => m.includes('turned off'))).toHaveLength(0); + }); }); diff --git a/apps/agent-forge/server/triggers/channelTriggers.ts b/apps/agent-forge/server/triggers/channelTriggers.ts index b4461ea63..4a56be6d8 100644 --- a/apps/agent-forge/server/triggers/channelTriggers.ts +++ b/apps/agent-forge/server/triggers/channelTriggers.ts @@ -388,6 +388,14 @@ export class ChannelTriggers { return; } + // R09: a turned-off agent's channels answer 200 skipped so platforms + // do not retry; the SDK handler (which would run the agent) is never + // called, and abort listeners do not run either. + if (!this.deps.host.isEnabled(route.agentId)) { + res.status(200).json({ ok: true, skipped: 'disabled' }); + return; + } + let body: Buffer; try { body = await readBody(req); diff --git a/apps/agent-forge/server/triggers/triggerHost.ts b/apps/agent-forge/server/triggers/triggerHost.ts index 1e3cea37f..59289bca3 100644 --- a/apps/agent-forge/server/triggers/triggerHost.ts +++ b/apps/agent-forge/server/triggers/triggerHost.ts @@ -98,6 +98,8 @@ const END_STATUSES = new Set(['done', 'error', 'stopped']); export class TriggerHost { private readonly queueCap: number; private readonly agents = new Map(); + /** `meta.enabled` per agent, refreshed on every reload (R09); absent means enabled. */ + private readonly enabled = new Map(); private readonly endListeners = new Set<(end: RunEnd) => void>(); private unsubscribers: Array<() => void> = []; private stopped = false; @@ -137,6 +139,7 @@ export class TriggerHost { const triggers = (spec as AgentSpecV2 | undefined)?.triggers ?? []; const state = this.stateFor(id); state.triggers = triggers; + await this.refreshEnabled(id, state); const known = new Set(triggers.map((t) => t.id)); const dropped = state.queue.filter((run) => !known.has(run.triggerId)); if (dropped.length > 0) { @@ -158,10 +161,39 @@ export class TriggerHost { return this.agents.get(agentId)?.triggers ?? []; } + /** + * `meta.enabled` for the agent (R09): false rejects every trigger + * dispatch while manual runs stay untouched. Unknown agents read as + * enabled, matching `defaultMeta`. + */ + isEnabled(agentId: string): boolean { + return this.enabled.get(agentId) ?? true; + } + + /** + * Re-reads `meta.enabled` into the cache; on a transition to disabled + * every queued host run is dropped (a running one finishes - R09). + */ + private async refreshEnabled(agentId: string, state: AgentTriggers): Promise { + // Test fakes may omit readMeta; a missing meta reads as enabled. + const meta = + this.deps.repo.readMeta === undefined + ? undefined + : await this.deps.repo.readMeta(agentId).catch(() => undefined); + const nowEnabled = meta?.enabled ?? true; + const wasEnabled = this.enabled.get(agentId) ?? true; + this.enabled.set(agentId, nowEnabled); + if (wasEnabled && !nowEnabled && state.queue.length > 0) { + const dropped = state.queue.splice(0); + this.deps.log(agentId, `Dropped ${dropped.length} queued runs: the agent was turned off`, 'warn'); + } + } + /** Accepts a trigger request: enqueues FIFO, starts immediately when the agent is free. */ dispatch(req: DispatchRequest): DispatchResult { - // R09 seam: the enabled-only gate lands here - reject with 'disabled' - // when the agent's meta.enabled is false before touching the queue. + // R09: a turned-off agent starts no triggered run (webhook/channel map + // this to their own skip responses; cron/chain log the skip). + if (!this.isEnabled(req.agentId)) return { ok: false, reason: 'disabled' }; const trigger = this.triggersOf(req.agentId).find((t) => t.id === req.triggerId); if (!trigger) return { ok: false, reason: 'unknown_trigger' }; diff --git a/apps/agent-forge/server/triggers/webhookRoute.ts b/apps/agent-forge/server/triggers/webhookRoute.ts index 7525b2aa2..53b729312 100644 --- a/apps/agent-forge/server/triggers/webhookRoute.ts +++ b/apps/agent-forge/server/triggers/webhookRoute.ts @@ -3,10 +3,12 @@ * * POST /hooks/:agentId/ -> 202 { queued, runKey, started, cancelled } * | 202 { skipped: 'filter' } - * | 400/401/403/404/422/429 + * | 400/401/403/404/409/422/429 * * `cancelled` lists the runKeys abort listeners (R03b) stopped for this * event, evaluated after signature verification and before the filter. + * A turned-off agent (`meta.enabled === false`, R09) answers 409 + * `LOUSHO_AGENT_DISABLED` right after auth - no event processing beyond it. * * The request path after `/hooks/:agentId/` must equal a webhook * trigger's `path` (the canvas default is `/hooks/`, so the @@ -117,6 +119,13 @@ export function registerWebhookRoutes(app: Express, deps: WebhookRouteDeps): voi const rawBody = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0); if (!(await checkSignature(req, res, trigger, rawBody, deps.resolveSecret))) return; + // R09: a turned-off agent does no event processing beyond auth - + // abort listeners, filter and dispatch all wait for it to be on. + if (!ws.triggers.isEnabled(agentId)) { + sendError(res, 409, 'Agent is turned off', 'LOUSHO_AGENT_DISABLED'); + return; + } + const event = eventOf(rawBody); // Abort listeners run after signature verification but before the @@ -158,7 +167,7 @@ export function registerWebhookRoutes(app: Express, deps: WebhookRouteDeps): voi return; } if (result.reason === 'disabled') { - sendError(res, 403, `Trigger '${trigger.id}' is disabled.`, 'LOUSHO_TRIGGER_DISABLED'); + sendError(res, 409, 'Agent is turned off', 'LOUSHO_AGENT_DISABLED'); return; } sendError(res, 404, `No webhook trigger '${trigger.id}' for agent '${agentId}'`); diff --git a/apps/agent-forge/src/canvas/NodeCard.tsx b/apps/agent-forge/src/canvas/NodeCard.tsx index 6e113adeb..05b404f72 100644 --- a/apps/agent-forge/src/canvas/NodeCard.tsx +++ b/apps/agent-forge/src/canvas/NodeCard.tsx @@ -32,6 +32,8 @@ export interface CanvasNodeData extends Record { graph?: Graph; /** Group nodes only: how many descendants the group hides while collapsed. */ insideCount?: number; + /** `meta.enabled` of the open agent (R09); false dims non-chat trigger nodes. Absent reads enabled. */ + agentEnabled?: boolean; } /** The flag-driven class list of a card or a group container ('rf-node' / 'rf-group' + state flags). */ @@ -190,9 +192,16 @@ export function NodeCard({ const chips = [...(chip !== undefined ? [chip] : []), ...(extraChips ?? [])]; // A proposal-edited node carries an "edited" marker chip (C21). const edited = graphNode.proposed === 'edit'; + // R09: a turned-off agent's non-chat trigger nodes dim with the contract tooltip. + const triggerOff = data.agentEnabled === false && graphNode.kind === 'trigger' && graphNode.data.item.type !== 'chat'; return ( -
+
{view.ports.in && } {view.ports.out && } void; /** Node ids with a pinned tool result (T04); their cards show the Pin badge. */ pinnedNodeIds?: ReadonlySet; + /** `meta.enabled` of the open agent (R09); false dims non-chat trigger nodes. Absent reads enabled. */ + agentEnabled?: boolean; } /** True when an `owner`-chain ancestor of `node` is a collapsed agent group. */ @@ -188,7 +190,7 @@ function groupSizes(graph: Graph, measured: MeasuredSizes): Map[] { - const { selectedNodeId, highlightedNodeId, breakpoints, statuses, problems, openNodeSpans, pinnedNodeIds } = decor; + const { selectedNodeId, highlightedNodeId, breakpoints, statuses, problems, openNodeSpans, pinnedNodeIds, agentEnabled } = decor; const sizes = groupSizes(graph, measured); return canvasNodeOrder(graph).map((n) => { const group = isGroupNode(n); @@ -212,6 +214,7 @@ export function toRfNodes(graph: Graph, decor: NodeDecor, measured: MeasuredSize problems: problems?.get(n.id) ?? [], onViewSpans: openNodeSpans, pinned: pinnedNodeIds?.has(n.id) === true, + agentEnabled, insideCount: group ? descendantCount(graph, n.id) : undefined, }, }; @@ -1066,6 +1069,7 @@ function useCanvasElements() { agentId, graph, displayGraph, + agents, proposal, hideProposal, setGraph, @@ -1096,6 +1100,9 @@ function useCanvasElements() { // The canvas node ids (`tool:`) carrying at least one pinned tool // result - their cards show the Pin badge (T04). const pinnedNodeIds = useMemo(() => new Set(pins.map((pin) => pin.nodeId)), [pins]); + // R09: the list item's `enabled` dims non-chat trigger nodes; an agent + // missing from the list (just created, mid-delete) reads enabled. + const agentEnabled = useMemo(() => agents.find((a) => a.id === agentId)?.enabled ?? true, [agents, agentId]); const problemsByNode = useMemo(() => { const map = new Map(); for (const problem of problems) { @@ -1109,10 +1116,10 @@ function useCanvasElements() { () => toRfNodes( displayGraph, - { selectedNodeId, highlightedNodeId, breakpoints, statuses, problems: problemsByNode, openNodeSpans, pinnedNodeIds }, + { selectedNodeId, highlightedNodeId, breakpoints, statuses, problems: problemsByNode, openNodeSpans, pinnedNodeIds, agentEnabled }, measured ), - [displayGraph, selectedNodeId, highlightedNodeId, breakpointsKey, statuses, problemsByNode, openNodeSpans, pinnedNodeIds, measured] + [displayGraph, selectedNodeId, highlightedNodeId, breakpointsKey, statuses, problemsByNode, openNodeSpans, pinnedNodeIds, agentEnabled, measured] ); const [selectedEdgeId, setSelectedEdgeId] = useAgentEdgeSelection(agentId); // Edges derive from the DISPLAY graph so proposal nodes are wired in; diff --git a/apps/agent-forge/src/screens/agentDetail/AgentOverviewTab.tsx b/apps/agent-forge/src/screens/agentDetail/AgentOverviewTab.tsx index b44038fc6..5f8156d14 100644 --- a/apps/agent-forge/src/screens/agentDetail/AgentOverviewTab.tsx +++ b/apps/agent-forge/src/screens/agentDetail/AgentOverviewTab.tsx @@ -2,7 +2,7 @@ import { ChevronRight, TriangleAlert } from 'lucide-react'; import { useEffect, useState, type ReactNode } from 'react'; import type { AgentSpecV2 } from 'lousho/spec'; -import type { ChannelTriggerStatus, TraceSummaryPayload } from '../../../shared/wireTypes'; +import type { AgentMeta, ChannelTriggerStatus, TraceSummaryPayload } from '../../../shared/wireTypes'; import { agentTabPath, BUILT_AGENT_TABS } from '../agents/agentTabs'; import { fmtAge, fmtDateTime, fmtDuration } from '../../observe/format'; import { Card } from '../../ui/Card'; @@ -158,11 +158,34 @@ function CapabilitiesCard({ rows }: { rows: CapabilityRow[] }): JSX.Element { ); } +/** + * The off-state note (R09): shown when `meta.enabled` is false and the + * agent has triggers that would fire - auto-disabled agents get the + * failure-streak copy instead. Em dashes are part of the contract copy. + */ +function triggersOffNote(meta: AgentMeta): string { + return meta.disabledReason === 'auto_failure_streak' + ? 'Turned off after 5 failed triggered runs in a row. Fix the cause, then turn it back on.' + : 'Off — triggers won\'t fire. Turn the agent on to start them.'; +} + /** The Triggers card: one icon + line per trigger (webhooks and channels also show their URL), or the empty copy. */ -function TriggersCard({ agentId, triggers }: { agentId: string; triggers: TriggerSpec[] }): JSX.Element { +function TriggersCard({ + agentId, + triggers, + meta, +}: { + agentId: string; + triggers: TriggerSpec[]; + meta: AgentMeta; +}): JSX.Element { const channelStatuses = useChannelStatuses(agentId); + const hasFirableTriggers = triggers.some((trigger) => trigger.type !== 'chat'); return ( + {meta.enabled === false && hasFirableTriggers && ( +

{triggersOffNote(meta)}

+ )} {triggers.length > 0 ? (
    {triggers.map((trigger) => { @@ -320,7 +343,7 @@ export function AgentOverviewTab() { )} - +
diff --git a/apps/agent-forge/src/screens/agentDetail/__tests__/AgentOverviewTab.test.tsx b/apps/agent-forge/src/screens/agentDetail/__tests__/AgentOverviewTab.test.tsx index e144fb097..64e929f3b 100644 --- a/apps/agent-forge/src/screens/agentDetail/__tests__/AgentOverviewTab.test.tsx +++ b/apps/agent-forge/src/screens/agentDetail/__tests__/AgentOverviewTab.test.tsx @@ -28,7 +28,7 @@ function meta(): AgentRecord['meta'] { }; } -function record(over: { spec?: Partial; requirements?: Requirement[] } = {}): AgentRecord { +function record(over: { spec?: Partial; requirements?: Requirement[]; meta?: Partial } = {}): AgentRecord { return { id: 'support-bot', spec: { @@ -38,7 +38,7 @@ function record(over: { spec?: Partial; requirements?: Requirement[ instructions: 'Be helpful.', ...over.spec, } as AgentSpecV2, - meta: meta(), + meta: { ...meta(), ...over.meta }, layout: { schemaVersion: 1, nodes: {}, notes: [] }, updatedAt: '2026-10-29T10:00:00Z', latestVersion: null, @@ -207,6 +207,43 @@ describe('AgentOverviewTab', () => { expect(triggersCard.textContent).toContain('Channel runs show up in Runs and Traces.'); }); + it('shows the off note on the Triggers card for a disabled agent with firable triggers (R09)', async () => { + const triggers: AgentSpecV2['triggers'] = [ + { id: 't1', type: 'chat' }, + { id: 't2', type: 'cron', cron: '0 9 * * *', prompt: 'x' }, + ]; + const host = renderTab(record({ spec: { triggers }, meta: { enabled: false } })); + await flush(); + + const note = card(host, 'Triggers').querySelector('.triggers-off-note'); + expect(note?.textContent).toBe("Off — triggers won't fire. Turn the agent on to start them."); + }); + + it('shows the auto-disabled copy when the streak turned the agent off (R09)', async () => { + const triggers: AgentSpecV2['triggers'] = [{ id: 't2', type: 'cron', cron: '0 9 * * *', prompt: 'x' }]; + const host = renderTab( + record({ spec: { triggers }, meta: { enabled: false, disabledReason: 'auto_failure_streak' } }), + ); + await flush(); + + expect(card(host, 'Triggers').querySelector('.triggers-off-note')?.textContent).toBe( + 'Turned off after 5 failed triggered runs in a row. Fix the cause, then turn it back on.', + ); + }); + + it('hides the off note for an enabled agent or one with only a chat trigger (R09)', async () => { + const cron: AgentSpecV2['triggers'] = [{ id: 't2', type: 'cron', cron: '0 9 * * *', prompt: 'x' }]; + const enabled = renderTab(record({ spec: { triggers: cron } })); + await flush(); + expect(card(enabled, 'Triggers').querySelector('.triggers-off-note')).toBeNull(); + + const chatOnly = renderTab( + record({ spec: { triggers: [{ id: 't1', type: 'chat' }] }, meta: { enabled: false } }), + ); + await flush(); + expect(card(chatOnly, 'Triggers').querySelector('.triggers-off-note')).toBeNull(); + }); + it('shows permission-rule counts by action in the Safety card', async () => { const host = renderTab( record({ diff --git a/apps/agent-forge/src/screens/agentDetail/agentDetail.css b/apps/agent-forge/src/screens/agentDetail/agentDetail.css index c909d7b72..96a555a55 100644 --- a/apps/agent-forge/src/screens/agentDetail/agentDetail.css +++ b/apps/agent-forge/src/screens/agentDetail/agentDetail.css @@ -171,10 +171,13 @@ a.readiness-chip:focus-visible { font-size: 12px; font-weight: 500; color: var(--text); transition: background-color var(--dur-fast) var(--ease); } .enabled-control:hover { background: var(--surface-2); } .enabled-control[data-pending] { cursor: progress; } -.agent-detail-notice { display: flex; align-items: flex-start; gap: var(--space-2); padding: var(--space-2) var(--space-3); +.agent-detail-notice, .triggers-off-note { padding: var(--space-2) var(--space-3); background: var(--warning-soft); border: 1px solid var(--warning-border); border-radius: var(--radius); font-size: 12px; color: var(--text); } +.agent-detail-notice { display: flex; align-items: flex-start; gap: var(--space-2); } .agent-detail-notice svg { flex: none; margin-top: 2px; color: var(--warning); } .agent-detail-notice p { margin: 0; } +/* R09: the Triggers card's off note (agentOverview.css places it in the card flow). */ +.triggers-off-note { margin: 0 0 var(--space-2); } .agent-detail-labels { display: flex; gap: 4px; list-style: none; margin: 0; padding: 0; } .agent-detail-label, .agent-detail-label-more { height: 20px; padding: 0 8px; border: 1px solid var(--border); border-radius: var(--radius-full); background: var(--surface-2); font-size: 11px; line-height: 18px; color: var(--text-muted); } diff --git a/apps/agent-forge/src/screens/agentDetail/agentOverview.css b/apps/agent-forge/src/screens/agentDetail/agentOverview.css index bc7b16937..0d74dca96 100644 --- a/apps/agent-forge/src/screens/agentDetail/agentOverview.css +++ b/apps/agent-forge/src/screens/agentDetail/agentOverview.css @@ -165,3 +165,4 @@ color: var(--danger); font-size: 12px; } + diff --git a/apps/agent-forge/src/screens/agents/agents.css b/apps/agent-forge/src/screens/agents/agents.css index c4963871f..052e91cf4 100644 --- a/apps/agent-forge/src/screens/agents/agents.css +++ b/apps/agent-forge/src/screens/agents/agents.css @@ -802,6 +802,10 @@ button.breakpoint-toggle { .canvas-wrap .rf-node.running { border-color: var(--accent); } +/* R09: a turned-off agent's non-chat trigger nodes dim (the title tooltip says why). */ +.rf-node.node-trigger-off { + opacity: 0.6; +} .canvas-wrap .rf-node.running::after { content: ''; position: absolute;