diff --git a/console/web/src/components/chat/ChatView.tsx b/console/web/src/components/chat/ChatView.tsx index d0d0a641a6..6184606d32 100644 --- a/console/web/src/components/chat/ChatView.tsx +++ b/console/web/src/components/chat/ChatView.tsx @@ -92,8 +92,6 @@ export function ChatView({ let thoughtId: string | null = null let thoughtBuffer = '' let fcallId: string | null = null - // PR #150: same iii function_call_id can fire fcall-start twice - // (once for function_execution_start, once for approval_requested). // Map UI message id → iii function_call_id so dedupe is reliable. const fcallMap = new Map() let assistantId: string | null = null diff --git a/console/web/src/lib/backend/pending-approvals-store.test.ts b/console/web/src/lib/backend/pending-approvals-store.test.ts new file mode 100644 index 0000000000..11ca0f1fa0 --- /dev/null +++ b/console/web/src/lib/backend/pending-approvals-store.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from 'vitest' +import { + diffPending, + type PendingApproval, +} from './pending-approvals-store' + +describe('diffPending', () => { + it('returns all entries as added when previous list is empty', () => { + const next: PendingApproval[] = [ + { function_call_id: 'fc-1', function_id: 'shell::fs::write', args: { path: '/tmp/x' } }, + ] + expect(diffPending([], next)).toEqual({ + added: next, + removed: [], + }) + }) + + it('returns all entries as removed when next list is empty', () => { + const prev: PendingApproval[] = [ + { function_call_id: 'fc-1', function_id: 'shell::fs::write', args: { path: '/tmp/x' } }, + ] + expect(diffPending(prev, [])).toEqual({ + added: [], + removed: prev, + }) + }) + + it('detects a single addition', () => { + const prev: PendingApproval[] = [ + { function_call_id: 'fc-1', function_id: 'shell::fs::write', args: {} }, + ] + const next: PendingApproval[] = [ + ...prev, + { function_call_id: 'fc-2', function_id: 'shell::shell', args: {} }, + ] + expect(diffPending(prev, next)).toEqual({ + added: [next[1]], + removed: [], + }) + }) + + it('detects a single removal', () => { + const prev: PendingApproval[] = [ + { function_call_id: 'fc-1', function_id: 'shell::fs::write', args: {} }, + { function_call_id: 'fc-2', function_id: 'shell::shell', args: {} }, + ] + const next: PendingApproval[] = [prev[0]!] + expect(diffPending(prev, next)).toEqual({ + added: [], + removed: [prev[1]], + }) + }) + + it('is keyed by function_call_id (treats objects with same id as same entry)', () => { + const prev: PendingApproval[] = [ + { function_call_id: 'fc-1', function_id: 'shell::shell', args: { command: 'ls' } }, + ] + const next: PendingApproval[] = [ + { function_call_id: 'fc-1', function_id: 'shell::shell', args: { command: 'rm' } }, + ] + expect(diffPending(prev, next)).toEqual({ added: [], removed: [] }) + }) +}) diff --git a/console/web/src/lib/backend/pending-approvals-store.ts b/console/web/src/lib/backend/pending-approvals-store.ts new file mode 100644 index 0000000000..e60bbdb144 --- /dev/null +++ b/console/web/src/lib/backend/pending-approvals-store.ts @@ -0,0 +1,33 @@ +/** + * Pure helpers for diffing two pending-approval lists by function_call_id. + * Used by the translate layer to derive synthetic `fcall-start` / + * `fcall-end` events from successive `turn_state_changed` updates. + * + * No state lives here — the per-session list lives in the closure of + * `realStream` (one chat = one mirror) and is passed in/out as plain + * arrays. Keeping this file framework-free makes it unit-testable in + * isolation. + */ + +export interface PendingApproval { + function_call_id: string + function_id: string + args: unknown +} + +export interface PendingDiff { + added: PendingApproval[] + removed: PendingApproval[] +} + +export function diffPending( + prev: PendingApproval[], + next: PendingApproval[], +): PendingDiff { + const prevIds = new Set(prev.map((e) => e.function_call_id)) + const nextIds = new Set(next.map((e) => e.function_call_id)) + return { + added: next.filter((e) => !prevIds.has(e.function_call_id)), + removed: prev.filter((e) => !nextIds.has(e.function_call_id)), + } +} diff --git a/console/web/src/lib/backend/real.ts b/console/web/src/lib/backend/real.ts index 5ccbf8e48b..ba44b63a6d 100644 --- a/console/web/src/lib/backend/real.ts +++ b/console/web/src/lib/backend/real.ts @@ -43,7 +43,7 @@ import { getIiiClient } from '@/lib/iii-client' import { newMessageId } from '@/lib/session-id' import type { Mode, ModelId } from '@/types/chat' import type { AgentEvent, SessionEventEnvelope } from '@/types/iii-agent-event' -import { translateAgentEvent } from './translate' +import { createTurnStateTranslator, translateAgentEvent } from './translate' import type { ChatBackend, ChatStreamOptions, StreamEvent } from './types' interface RunParams { @@ -112,23 +112,29 @@ async function* realStream( }) subscribed = true + const turnStateTranslator = createTurnStateTranslator() + + client + .call | null>('turn::get_state', { + session_id: sessionId, + }) + .then((record) => { + if (!record) return + queue.push({ + type: 'turn_state_changed', + event_type: 'state:created', + new_value: record, + }) + wake() + }) + .catch((err) => { + if (import.meta.env.DEV) { + console.warn('[real-backend] turn::get_state recovery failed', err) + } + }) + const { provider, model: modelId } = resolveRunParams(model) - // Route the turn-orchestrator kick-off through `harness::call` so the - // harness wraps the inner trigger with an instrumented span that - // seeds `iii.session.id` / `iii.message.id` as baggage. Without this - // wrapper, the engine's `engine::traces::group_by` returns empty - // results for "Group by session" / "Group by message" because the - // engine's own `handle_invocation` / `call run::start` spans aren't - // inside any baggage context. Mirrors `workers/harness/web/src/App.tsx` - // `send()` (it uses the HTTP bridge — we use the WS bus). - // - // The kickoff is fire-and-forget at the request level (the stream - // arrives via the subscribed channel, not the call's return value) - // but a synchronous rejection (policy denies `run::start`, bridge - // throws, transport fails) would otherwise leave the loop waiting - // for an `agent_end` that never arrives. Capture the error and - // surface it through the assistant stream so the UI shows it. let kickoffError: Error | null = null client .call('harness::call', { @@ -178,7 +184,11 @@ async function* realStream( if (kickoffError) continue const event = queue.shift() if (!event) continue - for (const streamEvent of translateAgentEvent(event, sessionId)) { + const streamEvents = + event.type === 'turn_state_changed' + ? turnStateTranslator(event, sessionId) + : translateAgentEvent(event, sessionId) + for (const streamEvent of streamEvents) { yield streamEvent } if (event.type === 'agent_end') return diff --git a/console/web/src/lib/backend/translate.test.ts b/console/web/src/lib/backend/translate.test.ts new file mode 100644 index 0000000000..1f6438090b --- /dev/null +++ b/console/web/src/lib/backend/translate.test.ts @@ -0,0 +1,119 @@ +import { describe, expect, it } from 'vitest' +import type { AgentEvent } from '@/types/iii-agent-event' +import { createTurnStateTranslator } from './translate' + +describe('createTurnStateTranslator', () => { + it('emits fcall-start { pendingApproval: true } when a new entry appears', () => { + const translate = createTurnStateTranslator() + const event: AgentEvent = { + type: 'turn_state_changed', + event_type: 'state:updated', + new_value: { + state: 'function_awaiting_approval', + awaiting_approval: [ + { + function_call_id: 'fc-1', + function_id: 'shell::fs::write', + args: { path: '/tmp/x' }, + }, + ], + }, + old_value: { + state: 'function_execute', + awaiting_approval: [], + }, + } + expect(translate(event, 'sess-a')).toEqual([ + { + kind: 'fcall-start', + functionId: 'shell::fs::write', + input: { path: '/tmp/x' }, + pendingApproval: true, + functionCallId: 'fc-1', + sessionId: 'sess-a', + }, + ]) + }) + + it('emits nothing when the awaiting_approval list is unchanged', () => { + const translate = createTurnStateTranslator() + const same = { + state: 'function_awaiting_approval', + awaiting_approval: [ + { function_call_id: 'fc-1', function_id: 'shell::shell', args: {} }, + ], + } + translate( + { + type: 'turn_state_changed', + event_type: 'state:created', + new_value: same, + }, + 'sess-a', + ) + expect( + translate( + { + type: 'turn_state_changed', + event_type: 'state:updated', + new_value: same, + old_value: same, + }, + 'sess-a', + ), + ).toEqual([]) + }) + + it('emits nothing when state leaves function_awaiting_approval (the orchestrator emits the matching function_execution_end)', () => { + const translate = createTurnStateTranslator() + translate( + { + type: 'turn_state_changed', + event_type: 'state:created', + new_value: { + state: 'function_awaiting_approval', + awaiting_approval: [ + { function_call_id: 'fc-1', function_id: 'shell::shell', args: {} }, + ], + }, + }, + 'sess-a', + ) + expect( + translate( + { + type: 'turn_state_changed', + event_type: 'state:updated', + new_value: { state: 'function_execute', awaiting_approval: [] }, + old_value: { + state: 'function_awaiting_approval', + awaiting_approval: [ + { function_call_id: 'fc-1', function_id: 'shell::shell', args: {} }, + ], + }, + }, + 'sess-a', + ), + ).toEqual([]) + }) + + it('partitions mirrors by sessionId so two chats do not interfere', () => { + const translate = createTurnStateTranslator() + const pending = { + state: 'function_awaiting_approval', + awaiting_approval: [ + { function_call_id: 'fc-1', function_id: 'shell::shell', args: {} }, + ], + } + translate( + { type: 'turn_state_changed', event_type: 'state:created', new_value: pending }, + 'sess-a', + ) + expect( + translate( + { type: 'turn_state_changed', event_type: 'state:created', new_value: pending }, + 'sess-b', + ), + ).toHaveLength(1) + }) +}) diff --git a/console/web/src/lib/backend/translate.ts b/console/web/src/lib/backend/translate.ts index 8711e95893..80537e6ec6 100644 --- a/console/web/src/lib/backend/translate.ts +++ b/console/web/src/lib/backend/translate.ts @@ -22,12 +22,10 @@ * Other events: * - `function_execution_start` → `fcall-start` (with args). * - `function_execution_end` → `fcall-end` (with result). - * - `approval_requested` → `fcall-start` with `pendingApproval: true`. - * - `approval_resolved` deny → synthetic `fcall-end`. - * - `approval_resolved` allow → noop (the matching exec_start follows). * - `agent_end` → `assistant-end`. * - `agent_start` / `turn_start` / `turn_end` / `message_end` / * `function_execution_update` → noop. + * - `turn_state_changed` → noop (routed through `createTurnStateTranslator`). */ import type { @@ -35,7 +33,10 @@ import type { AgentMessage, AssistantMessageEvent, ContentBlock, + TurnStateChangedEvent, } from '@/types/iii-agent-event' +import { diffPending, type PendingApproval } from './pending-approvals-store' +import { pendingApprovalsFromTurnState } from './turn-state-mirror' import type { StreamEvent } from './types' export function translateAgentEvent( @@ -49,11 +50,10 @@ export function translateAgentEvent( case 'function_execution_update': return [] + case 'turn_state_changed': + return [] + case 'message_end': - // Per-turn boundary: signal `assistant-end` so consumers can finalize - // the streaming assistant message and reset their per-turn pointer. - // Without this, multi-turn agents (text → fcall → text → ...) merge - // every turn's text into the first assistant message. if (event.message.role === 'assistant') { return [{ kind: 'assistant-end' }] } @@ -85,35 +85,6 @@ export function translateAgentEvent( }, ] - case 'approval_requested': - return [ - { - kind: 'fcall-start', - functionId: event.function_id, - input: event.args, - pendingApproval: true, - functionCallId: event.function_call_id, - sessionId, - }, - ] - - case 'approval_resolved': - if (event.decision === 'allow') { - return [] - } - return [ - { - kind: 'fcall-end', - output: { - error: { - kind: 'denied', - message: event.reason ?? 'denied by user', - }, - }, - durationMs: 0, - }, - ] - case 'agent_end': return [{ kind: 'assistant-end' }] } @@ -145,20 +116,9 @@ function translateMessageUpdate(llm: AssistantMessageEvent): StreamEvent[] { } function translateMessageStart(message: AgentMessage): StreamEvent[] { - // User prompts and tool/function results aren't surfaced as backend-driven - // StreamEvents — the consumer renders the user message itself and the - // tool result rides on `fcall-end`'s output. if (message.role !== 'assistant') { return [] } - // Phase 2.A: assistant text and thinking already streamed via - // `message_update`; don't double-emit them here. We only forward - // function-call / image blocks, which don't ride on the streamed - // delta path. (Function calls actually arrive on - // `function_execution_start`, but the assistant message technically - // owns them too — keeping the early return below preserves the - // historical contract for non-streaming providers that emit a single - // `MessageStart` with the full body.) const hasStreamableContent = message.content.some( (b) => b.type === 'text' || b.type === 'thinking', ) @@ -190,9 +150,40 @@ function appendBlock(block: ContentBlock, out: StreamEvent[]): void { case 'functionCall': case 'functionResult': case 'image': - // FunctionCall/Result blocks ride on dedicated AgentEvents - // (function_execution_start/end); images aren't part of the - // StreamEvent contract today. return } } + +/** + * Stateful translator for `turn_state_changed` events. Holds a per-session + * mirror of the previous `awaiting_approval` list so it can emit a + * `fcall-start { pendingApproval: true }` exactly once per new pending + * call, and suppress duplicates when the same record is re-broadcast + * (the backend emits on every turn_state write, not just transitions + * into the parking state). + * + * No `fcall-end` is emitted when the list shrinks — the orchestrator + * already fires `function_execution_end` for resolved/denied calls + * through its existing path. Removing the entry from our mirror is + * bookkeeping only. + */ +export function createTurnStateTranslator(): ( + event: TurnStateChangedEvent, + sessionId: string, +) => StreamEvent[] { + const mirrors = new Map() + return (event, sessionId) => { + const prev = mirrors.get(sessionId) ?? [] + const next = pendingApprovalsFromTurnState(event.new_value) + mirrors.set(sessionId, next) + const { added } = diffPending(prev, next) + return added.map((entry) => ({ + kind: 'fcall-start' as const, + functionId: entry.function_id, + input: entry.args, + pendingApproval: true, + functionCallId: entry.function_call_id, + sessionId, + })) + } +} diff --git a/console/web/src/lib/backend/turn-state-mirror.test.ts b/console/web/src/lib/backend/turn-state-mirror.test.ts new file mode 100644 index 0000000000..f354e0f493 --- /dev/null +++ b/console/web/src/lib/backend/turn-state-mirror.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, it } from 'vitest' +import { pendingApprovalsFromTurnState } from './turn-state-mirror' + +describe('pendingApprovalsFromTurnState', () => { + it('returns [] when the record is null', () => { + expect(pendingApprovalsFromTurnState(null)).toEqual([]) + }) + + it('returns [] when state is not function_awaiting_approval', () => { + expect( + pendingApprovalsFromTurnState({ + state: 'function_execute', + awaiting_approval: [ + { function_call_id: 'fc-1', function_id: 'shell::shell', args: {} }, + ], + }), + ).toEqual([]) + }) + + it('returns [] when awaiting_approval is missing or empty', () => { + expect( + pendingApprovalsFromTurnState({ + state: 'function_awaiting_approval', + awaiting_approval: [], + }), + ).toEqual([]) + expect( + pendingApprovalsFromTurnState({ + state: 'function_awaiting_approval', + }), + ).toEqual([]) + }) + + it('extracts pending entries when state is function_awaiting_approval', () => { + const entries = [ + { + function_call_id: 'fc-1', + function_id: 'shell::fs::write', + args: { path: '/tmp/x' }, + }, + ] + expect( + pendingApprovalsFromTurnState({ + state: 'function_awaiting_approval', + awaiting_approval: entries, + }), + ).toEqual(entries) + }) + + it('filters malformed entries (missing function_call_id or function_id)', () => { + expect( + pendingApprovalsFromTurnState({ + state: 'function_awaiting_approval', + awaiting_approval: [ + { function_call_id: 'fc-1', function_id: 'shell::shell', args: {} }, + { function_call_id: 'fc-2' }, // missing function_id + { function_id: 'shell::shell', args: {} }, // missing function_call_id + 'not-an-object', + ], + }), + ).toEqual([ + { function_call_id: 'fc-1', function_id: 'shell::shell', args: {} }, + ]) + }) +}) diff --git a/console/web/src/lib/backend/turn-state-mirror.ts b/console/web/src/lib/backend/turn-state-mirror.ts new file mode 100644 index 0000000000..92b2b86c26 --- /dev/null +++ b/console/web/src/lib/backend/turn-state-mirror.ts @@ -0,0 +1,34 @@ +/** + * Derive the canonical pending-approvals list from a turn_state record. + * The orchestrator parks in `function_awaiting_approval` with a list of + * pending function calls on the record itself; this function is the + * single place that knows the schema, so the rest of the FE can stay + * agnostic of the orchestrator's internal shape. + */ + +import type { PendingApproval } from './pending-approvals-store' + +const PARKING_STATE = 'function_awaiting_approval' + +export function pendingApprovalsFromTurnState( + record: unknown, +): PendingApproval[] { + if (!record || typeof record !== 'object') return [] + const r = record as Record + if (r.state !== PARKING_STATE) return [] + const raw = r.awaiting_approval + if (!Array.isArray(raw)) return [] + const out: PendingApproval[] = [] + for (const entry of raw) { + if (!entry || typeof entry !== 'object') continue + const e = entry as Record + if (typeof e.function_call_id !== 'string') continue + if (typeof e.function_id !== 'string') continue + out.push({ + function_call_id: e.function_call_id, + function_id: e.function_id, + args: e.args ?? {}, + }) + } + return out +} diff --git a/console/web/src/types/iii-agent-event.ts b/console/web/src/types/iii-agent-event.ts index 3933e2222e..8f35069953 100644 --- a/console/web/src/types/iii-agent-event.ts +++ b/console/web/src/types/iii-agent-event.ts @@ -112,9 +112,6 @@ export interface FunctionResult { terminate?: boolean } -/** Outcome of an approval gate. */ -export type ApprovalDecision = 'allow' | 'deny' - /** * Provider-side streaming event. Mirrors the Rust enum * `harness/crates/harness-types/src/stream_event.rs::AssistantMessageEvent`. @@ -200,18 +197,17 @@ export type AgentEvent = is_error: boolean } | { - type: 'approval_requested' - function_call_id: string - function_id: string - args: unknown - } - | { - type: 'approval_resolved' - function_call_id: string - decision: ApprovalDecision - reason?: string + type: 'turn_state_changed' + event_type: 'state:created' | 'state:updated' + new_value: Record + old_value?: Record } +export type TurnStateChangedEvent = Extract< + AgentEvent, + { type: 'turn_state_changed' } +> + /** * Envelope the harness fanout pushes to `ui::session::event::`. * See `harness/src/fanout.rs` `subscribers_for(session_id)`. diff --git a/harness-node/config.yaml b/harness-node/config.yaml index 0f5f1cad93..22ea0d11f9 100644 --- a/harness-node/config.yaml +++ b/harness-node/config.yaml @@ -11,12 +11,11 @@ permissions_path: "./iii-permissions.yaml" sync_default_timeout_ms: 120000 system_default_skills: - "iii://iii-directory/index" +policy_function_id: "policy::check_permissions" # approval-gate approval_gate: - topic: "agent::before_function_call" approval_state_scope: "approvals" - policy_function_id: "policy::check_permissions" # session worker session: diff --git a/harness-node/docs/architecture.md b/harness-node/docs/architecture.md index bce5757ebc..c831b4b429 100644 --- a/harness-node/docs/architecture.md +++ b/harness-node/docs/architecture.md @@ -20,7 +20,7 @@ workers. |---|---|---|---| | harness | [src/harness/](harness-node/src/harness/) | Meta-worker; loads `iii-permissions.yaml`, exposes `harness::call` (WS ingestion bridge — see [Telemetry & trace correlation](#telemetry--trace-correlation)) / `harness::status` / `policy::check_permissions` / `ui::*`, spins up `agent::events` fan-out. | [workers/harness.md](harness-node/docs/workers/harness.md) | | turn-orchestrator | [src/turn-orchestrator/](harness-node/src/turn-orchestrator/) | Durable FSM driving each agent turn; chokepoint dispatcher for `agent::call`. | [workers/turn-orchestrator.md](harness-node/docs/workers/turn-orchestrator.md) | -| approval-gate | [src/approval-gate/](harness-node/src/approval-gate/) | Hook subscriber on `agent::before_function_call`; consults policy and pauses for user resolution. | [workers/approval-gate.md](harness-node/docs/workers/approval-gate.md) | +| approval-gate | [src/approval-gate/](harness-node/src/approval-gate/) | Owns approval state. Registers `approval::resolve` and a state-trigger adapter on the `approvals` scope that wakes `turn::step` on every decision write. | [workers/approval-gate.md](harness-node/docs/workers/approval-gate.md) | | session | [src/session/](harness-node/src/session/) | Branching session storage (`session-tree::*`) plus per-session inbox queues (`session-inbox::*`). | [workers/session.md](harness-node/docs/workers/session.md) | | llm-budget | [src/llm-budget/](harness-node/src/llm-budget/) | Workspace + agent LLM spend caps with alerts, forecast, period rollover. | [workers/llm-budget.md](harness-node/docs/workers/llm-budget.md) | | hook-fanout | [src/hook-fanout/](harness-node/src/hook-fanout/) | Generic publish-and-collect primitive over a stream topic. | [workers/hook-fanout.md](harness-node/docs/workers/hook-fanout.md) | @@ -62,20 +62,15 @@ flowchart LR turnOrch -- "provider::*::stream" --> provAnth turnOrch -- "provider::*::stream" --> provOAI - turnOrch -- "agent::call → hook-fanout::publish_collect" --> hook + turnOrch -- "consultBefore: policy::check_permissions" --> harness + turnOrch -- "agent::call → hook-fanout::publish_collect (after-hook)" --> hook turnOrch -- "session-tree::* mirror" --> session turnOrch -- "state::* persistence" --> state - hook -- "publish agent::before_function_call" --> state - state -- "durable:subscriber" --> approval - approval -- "policy::check_permissions" --> harness - approval -- "stream::set agent::hook_reply" --> hook - hook -- "stream trigger on agent::hook_reply" --> hook - client -- "approval::resolve" --> approval approval -- "state::set approvals//" --> state state -- "state trigger (scope=approvals)" --> approval - approval -- "publish turn::step_requested" --> turnOrch + approval -- "iii.trigger turn::step" --> turnOrch provAnth -- "auth::get_token" --> auth provOAI -- "auth::get_token" --> auth @@ -120,51 +115,37 @@ stateDiagram-v2 ## Approval flow -The gate replies synchronously to every `agent::before_function_call` -event — `allow`, `deny`, or `pending`. There is no in-gate polling; -resume is driven by a `state` trigger on the `approvals` scope. The -orchestrator parks the turn in `function_awaiting_approval` until that -trigger fires and re-publishes `turn::step_requested`. +The orchestrator consults `policy::check_permissions` directly inside +`consultBefore` — `allow`, `deny`, or `pending`. There is no hook fanout on +the before path; resume is driven by a `state` trigger on the `approvals` +scope. The orchestrator parks the turn in `function_awaiting_approval` until +that trigger fires and re-triggers `turn::step`. ```mermaid sequenceDiagram participant Turn as turn-orchestrator (FSM) - participant Hook as hook-fanout::publish_collect participant Bus as iii bus (state::* + stream::*) - participant Gate as approval-gate (policy::approval_gate) participant Harness as harness (policy::check_permissions) + participant Gate as approval-gate participant User - Turn->>Hook: publish_collect("agent::before_function_call", call) - Hook->>Bus: iii::durable::publish - Bus-->>Gate: durable:subscriber wakes - Gate->>Harness: policy::check_permissions(function_id, args) + Turn->>Harness: policy::check_permissions(function_id, args) [5s timeout] alt rule.action == allow - Harness-->>Gate: allow - Gate-->>Hook: {block:false, subscriber, approval_gate} - Hook-->>Turn: merged.allow → dispatch the call + Harness-->>Turn: allow → dispatch the call else rule.action == deny - Harness-->>Gate: deny + DenialEnvelope - Gate->>Bus: stream::set agent::events (function_call_denied) - Gate-->>Hook: {block:true, denial, subscriber, approval_gate} - Hook-->>Turn: merged.deny → DenialResult + Harness-->>Turn: deny + DenialEnvelope → DenialResult else no rule (needs_approval) - Harness-->>Gate: needs_approval - Gate->>Bus: stream::set agent::events (approval_requested) - Gate-->>Hook: {block:true, status:"pending", subscriber, approval_gate} - Hook-->>Turn: merged.pending → park in function_awaiting_approval + Harness-->>Turn: needs_approval → park in function_awaiting_approval Note over Turn,Bus: Orchestrator stops re-publishing turn::step_requested.
The TurnStateRecord.awaiting_approval list pins the open calls. User->>Gate: approval::resolve(decision, reason) Gate->>Bus: state::set approvals// = {decision, reason} - Gate->>Bus: stream::set agent::events (approval_resolved) - Note over Gate,Turn: state trigger on scope=approvals
(approval::is_decision_write → approval::on_decision_written)
publishes turn::step_requested - Bus-->>Turn: durable wake → function_awaiting_approval reads
approvals// for each pending entry + Note over Gate,Turn: state trigger on scope=approvals
(approval::is_decision_write → approval::on_decision_written)
directly triggers turn::step + Bus-->>Turn: turn::step wakes → function_awaiting_approval reads
approvals// for each pending entry Turn->>Turn: fold decisions into prepared snapshot,
transition back to function_execute end ``` -Fail-closed: if `iii::durable::publish` errors or no subscriber replies, -`hook-fanout::publish_collect` returns `publish_failed: true` and +Fail-closed: policy unreachable (transport error or 5 s timeout) → `consultBefore` denies the call with a `gate_unavailable` envelope. Abort: `router::abort` writes `session//abort_signal = true` (waking @@ -185,7 +166,6 @@ and protect the gate / state surface. |---|---|---|---| | `kernel/no-self-approve` | `approval::resolve` | deny | An agent must not flip its own pending approval. | | `kernel/no-self-policy` | `policy::check_permissions` | deny | Don't let the model trace or precompute its own gate decision. | -| `kernel/no-self-policy-gate` | `policy::approval_gate` | deny | Same as above, for the subscriber. | | `kernel/no-self-hook` | `hook-fanout::publish_collect` | deny | The hook primitive is operator-only. | | `kernel/no-state-set` | `state::set` | deny | Use a worker function; never raw state writes. | | `kernel/no-state-update` | `state::update` | deny | Same. | diff --git a/harness-node/docs/workers/approval-gate.md b/harness-node/docs/workers/approval-gate.md index 6c5cf72980..56ff2ca8c2 100644 --- a/harness-node/docs/workers/approval-gate.md +++ b/harness-node/docs/workers/approval-gate.md @@ -1,49 +1,51 @@ # approval-gate -Hook subscriber on `agent::before_function_call` that consults -`policy::check_permissions` and parks calls for `approval::resolve` when a -call matches no rule. Resume is fully reactive — writing the decision to -state is itself the wake-up event. +Owns approval state. Registers `approval::resolve` and a state-trigger adapter +on the `approvals` scope that wakes `turn::step` on every decision write. ## Purpose -The approval gate is the runtime enforcement of -[iii-permissions.yaml](iii-permissions.yaml). Every dispatch from -`agent::call` publishes the call envelope to `agent::before_function_call`; -the gate is the durable subscriber on that topic. It asks the harness -worker's `policy::check_permissions` for a decision and returns -**synchronously** — there is no polling await-loop. The three outcomes: +The approval gate is the runtime resolution surface for +[iii-permissions.yaml](iii-permissions.yaml). It does **not** intercept +function calls on the bus — the turn-orchestrator consults +`policy::check_permissions` directly inside `consultBefore`. The gate's job +is to accept human decisions and release parked turns. -| Policy outcome | Gate reply | Orchestrator effect | -|---|---|---| -| `allow` | `{ block: false, subscriber, approval_gate }` | dispatch proceeds | -| `deny` | `{ block: true, reason, denial, subscriber, approval_gate }` + emits `function_call_denied` | dispatch short-circuits with the denial | -| `needs_approval` | `{ block: true, status: 'pending', subscriber, approval_gate }` + emits `approval_requested` | orchestrator parks the call on its turn record (`function_awaiting_approval`) and stops stepping until a decision lands | +| Policy outcome (in orchestrator) | Orchestrator effect | +|---|---| +| `allow` | dispatch proceeds immediately | +| `deny` | dispatch short-circuits with a `DenialEnvelope` | +| `needs_approval` | orchestrator parks the call in `function_awaiting_approval` and stops re-publishing `turn::step_requested` until a decision lands | Resume happens when an operator calls `approval::resolve`. The resolve -handler writes the decision to `approvals//`; a `state` -trigger on the approvals scope (`approval::on_decision_written`, gated by -the condition function `approval::is_decision_write`) then publishes -`turn::step_requested` so the orchestrator's FSM picks up where it paused -and reads the decision back out of state. The gate itself owns no -in-process pending map. +handler writes the decision to `approvals//`; the state +trigger (`approval::on_decision_written`, gated by +`approval::is_decision_write`) then directly triggers `turn::step` for +the affected session — the same primitive the orchestrator uses in its own +`on-record-written` pattern. The gate itself owns no in-process pending map. + +## Topology + +approval-gate owns two pieces of state surface: -Fail-closed: the orchestrator's `consultBefore` denies the call with a -`gate_unavailable` envelope if the hook fanout fails or no subscriber -replies, so a missing/erroring gate never lets a call through. +1. The `approval::resolve` function — bus entry point the UI calls to write a + decision into `approvals//`. +2. A state trigger on `scope: approvals` (condition: `new_value.decision` is a + string) whose handler directly triggers `turn::step` for the affected + session — the same primitive the orchestrator uses for `on-record-written`. + +The orchestrator consults policy directly (`policy::check_permissions`) inside +`consultBefore`; approval-gate is no longer a hook subscriber and there is no +`agent::before_function_call` topic. ## Registered functions - `approval::resolve` — Write the final approval decision (`allow` or `deny`) for a pending call. The state write is itself the wake-up event. -- `policy::approval_gate` — Consult `policy::check_permissions` and reply allow, deny, or pending. - `approval::is_decision_write` — Condition function bound to the approvals state trigger; returns `true` only for `state:created`/`state:updated` events whose `new_value.decision` is a string. -- `approval::on_decision_written` — State trigger adapter: extracts `session_id` from the `/` key and publishes `turn::step_requested`. +- `approval::on_decision_written` — State trigger adapter: extracts `session_id` from the `/` key and directly triggers `turn::step` for that session. ## Triggers -- **Durable subscriber** on `agent::before_function_call` (configurable via - `approval_gate.topic`) → `policy::approval_gate`. Registered in - [src/approval-gate/register.ts](harness-node/src/approval-gate/register.ts). - **State trigger** on `scope: approvals` gated by `condition_function_id: approval::is_decision_write` → `approval::on_decision_written`. This is what wakes the orchestrator @@ -64,52 +66,53 @@ All keys live under the iii state scope configured by [src/turn-orchestrator/abort.ts](harness-node/src/turn-orchestrator/abort.ts) when `router::abort` fires while the turn is paused on approvals. -## Approval / agent events - -Bound to `agent::events` so the UI can render the gate's lifecycle: +## Pending-approval signalling -| Event type | When | Written by | -|---|---|---| -| `approval_requested` | needs_approval outcome | `gate-subscriber.ts` | -| `function_call_denied` | deny outcome | `gate-subscriber.ts` | -| `approval_resolved` | `approval::resolve` succeeds | `handleResolveWithEvents` in `pending.ts` | +The frontend does not consume signal events for approvals. Instead, the +orchestrator emits a `turn_state_changed` event on every `turn_state` +write (see `harness-node/docs/workers/turn-orchestrator.md`). The +console reads `awaiting_approval` from the new record to render +approve/deny buttons, and uses `function_execution_end` (which already +carries the blocked result for denied calls via the orchestrator's +`handleExecute` blocked-result branch) to close the card. On page +reload the console fires a one-shot `turn::get_state { session_id }` +(orchestrator function — clients do not read iii state directly) to +recover any modals that were pending when the page loaded. ## Configuration From the `approval_gate` section of [config.yaml](harness-node/config.yaml): -- `topic` (default `agent::before_function_call`) — durable topic the gate - subscribes to. - `approval_state_scope` (default `approvals`) — iii state scope for decision records. The state trigger that wakes the orchestrator is registered against this same scope. -- `policy_function_id` (default `policy::check_permissions`) — function id - consulted for the rule decision. -There is no `default_timeout_ms`: the gate returns synchronously and the -turn record sits in `function_awaiting_approval` until a decision is -written, so the only relevant timeout is the durable bus's own backstop. +The policy function id is owned by the orchestrator's config slice, not +approval-gate's — set it at top level as `policy_function_id` (default +`policy::check_permissions`). See `harness-node/config.yaml` and +[workers/turn-orchestrator.md](workers/turn-orchestrator.md). + +There is no `default_timeout_ms`: the turn record sits in +`function_awaiting_approval` until a decision is written, so the only relevant +timeout is the durable bus's own backstop. ## Dependencies From [src/approval-gate/iii.worker.yaml](harness-node/src/approval-gate/iii.worker.yaml): -no explicit dependency block (the gate calls `policy::check_permissions` -on the harness worker over the bus and reads/writes iii state). +no explicit dependency block (the gate reads/writes iii state). ## Source layout | File | Purpose | |---|---| | [src/approval-gate/main.ts](harness-node/src/approval-gate/main.ts) | Binary entry point (`iii-approval-gate`). | -| [src/approval-gate/register.ts](harness-node/src/approval-gate/register.ts) | Registers `approval::resolve`, `policy::approval_gate`, the decision-written trigger pair, and the durable subscriber. | +| [src/approval-gate/register.ts](harness-node/src/approval-gate/register.ts) | Registers `approval::resolve` and the decision-written trigger pair. | | [src/approval-gate/config.ts](harness-node/src/approval-gate/config.ts) | Loads the `approval_gate` config section. | -| [src/approval-gate/types.ts](harness-node/src/approval-gate/types.ts) | Wire types: `IncomingCall`, `DenialEnvelope`, `MatchedConstraint`, `GateBlockReply`, `pendingKey`, `blockReplyFor`, `extractCall`. | -| [src/approval-gate/state-bus.ts](harness-node/src/approval-gate/state-bus.ts) | `StateBus` interface + `IiiStateBus` implementation. | -| [src/approval-gate/policy-consult.ts](harness-node/src/approval-gate/policy-consult.ts) | Calls `policy::check_permissions` and decodes the decision. | -| [src/approval-gate/gate-subscriber.ts](harness-node/src/approval-gate/gate-subscriber.ts) | `policy::approval_gate` handler: extracts the call, consults policy, emits the matching `agent::events` frame, returns the hook reply synchronously. | -| [src/approval-gate/pending.ts](harness-node/src/approval-gate/pending.ts) | `handleResolve` writes `{decision, reason}` to `approvals//`; `handleResolveWithEvents` adds the `approval_resolved` agent event. | -| [src/approval-gate/on-decision-written.ts](harness-node/src/approval-gate/on-decision-written.ts) | State trigger adapter — `approval::is_decision_write` (condition) + `approval::on_decision_written` (handler) — publishes `turn::step_requested` when a decision lands. | +| [src/approval-gate/types.ts](harness-node/src/approval-gate/types.ts) | Wire types and constants: `DenialEnvelope`, `MatchedConstraint`, `WireDecision`, `DeniedBy`, `FN_RESOLVE`, `STATE_SCOPE`, `pendingKey`. | +| [src/approval-gate/policy-consult.ts](harness-node/src/approval-gate/policy-consult.ts) | Calls `policy::check_permissions` and decodes the decision (imported by the orchestrator's `hook.ts`). | +| [src/approval-gate/pending.ts](harness-node/src/approval-gate/pending.ts) | `handleResolve` writes `{decision, reason}` to `approvals//`. | +| [src/approval-gate/on-decision-written.ts](harness-node/src/approval-gate/on-decision-written.ts) | State trigger adapter — `approval::is_decision_write` (condition) + `approval::on_decision_written` (handler) — directly triggers `turn::step` when a decision lands. | | [src/approval-gate/denial.ts](harness-node/src/approval-gate/denial.ts) | Builds `DenialEnvelope` instances (permissions, user, gate_unavailable). | | [src/approval-gate/iii.worker.yaml](harness-node/src/approval-gate/iii.worker.yaml) | Worker manifest. | diff --git a/harness-node/docs/workers/turn-orchestrator.md b/harness-node/docs/workers/turn-orchestrator.md index 49912711f1..efe7539c80 100644 --- a/harness-node/docs/workers/turn-orchestrator.md +++ b/harness-node/docs/workers/turn-orchestrator.md @@ -10,32 +10,39 @@ immediately; the rest of the work happens inside the durable `turn::step` state machine, woken once per state transition by a publish to the `turn::step_requested` topic. The FSM provisions the sandbox, streams the assistant turn from a provider, executes any returned function calls -through the `agent::call` chokepoint (which the approval gate intercepts), -emits `agent::events` for the harness fanout, and persists everything to -iii state so the run survives restarts. +through the `agent::call` chokepoint, emits `agent::events` for the +harness fanout, and persists everything to iii state so the run survives +restarts. `agent::call` is the single dispatcher every agent-issued tool call passes -through. It runs `consultBefore` (publishes `agent::before_function_call` -and waits for the approval gate's reply) before forwarding to the target -function id. Fail-closed: a missing/erroring gate denies the call with a -`gate_unavailable` `DenialEnvelope`. +through. It runs `consultBefore` before forwarding to the target function +id. `consultBefore` triggers `policy::check_permissions` directly (5 s +timeout) and maps the reply to allow / deny / pending. Fail-closed: policy +unreachable → deny with a `gate_unavailable` `DenialEnvelope`. ## Registered functions - `run::start` — Start a durable agent session and return immediately. - `run::start_and_wait` — Start a durable agent session and block until terminal (test/dev convenience). - `turn::step` — Run one durable state machine transition for a session. +- `turn::get_state` — Read the current `TurnStateRecord` for a session (or null for unknown sessions). UI clients use this on reload to recover any in-progress modals (e.g. `function_awaiting_approval`) without reading iii state directly. - `agent::call` — LLM-facing dispatcher: dispatches an iii function and returns a FunctionResult. - `turn::is_abort_signal_set` — Condition function bound to the agent-scope state trigger; matches `state:created`/`state:updated` writes that set `session//abort_signal` to `true`. - `turn::on_abort_signal` — State trigger adapter: publishes `turn::step_requested` when the abort signal is set so the FSM advances on the next safe boundary. - `turn::is_terminal_state_write` — Condition function bound to the terminal state trigger; matches writes to `session//turn_state` whose `new_value.state === 'stopped'`. - `turn::on_terminal_state` — State trigger adapter: resolves the in-process waiter installed by `run::start_and_wait` for that session id. +- `turn::is_stepable_record_write` — Condition function bound to the record-written state trigger; matches `turn_state` writes whose `new_value.state` is non-terminal and non-parking (i.e. excludes `stopped` and `function_awaiting_approval`). +- `turn::on_record_written` — State trigger adapter: directly triggers `turn::step` for the affected session, so saving the record is itself the wake-up event. +- `turn::is_turn_state_write` — Condition function bound to the turn-state-changed trigger; matches every `state:created` / `state:updated` write to `session//turn_state` regardless of FSM state. +- `turn::on_turn_state_changed` — State trigger adapter: emits a `turn_state_changed` agent event carrying the full new (and prior) `TurnStateRecord` so the UI can derive pending approvals from state. ## Triggers - **Durable subscriber** on `turn::step_requested` → `turn::step`. Registered in [src/turn-orchestrator/subscriber.ts](harness-node/src/turn-orchestrator/subscriber.ts). Each `step` loads the `TurnStateRecord`, runs one transition, saves it back, and re-publishes `turn::step_requested` unless the run is terminal **or** paused on approvals (`function_awaiting_approval`). The approval-gate worker wakes those paused turns through its own state trigger on the `approvals` scope. - **State trigger** on `scope: agent` gated by `condition_function_id: turn::is_abort_signal_set` → `turn::on_abort_signal`. Registered in [src/turn-orchestrator/on-abort-signal.ts](harness-node/src/turn-orchestrator/on-abort-signal.ts). Publishes `turn::step_requested` the moment `session//abort_signal` is set to `true`, so the FSM advances to `steering_check` (and observes the abort) on the next safe boundary without waiting for the current step to time out. - **State trigger** on `scope: agent` gated by `condition_function_id: turn::is_terminal_state_write` → `turn::on_terminal_state`. Registered in [src/turn-orchestrator/on-terminal.ts](harness-node/src/turn-orchestrator/on-terminal.ts). Fires on the `session//turn_state` write that lands `stopped`; the handler resolves the per-session waiter installed by `run::start_and_wait` so the sync wrapper returns without polling. +- **State trigger** on `scope: agent` gated by `condition_function_id: turn::is_stepable_record_write` → `turn::on_record_written`. Registered in [src/turn-orchestrator/on-record-written.ts](harness-node/src/turn-orchestrator/on-record-written.ts). Directly triggers `turn::step` for the affected session on every non-terminal, non-parking `session//turn_state` write. Replaces the imperative `publishStep` self-publish — saving the record is now the wake. +- **State trigger** on `scope: agent` gated by `condition_function_id: turn::is_turn_state_write` → `turn::on_turn_state_changed`. Registered in [src/turn-orchestrator/on-turn-state-changed.ts](harness-node/src/turn-orchestrator/on-turn-state-changed.ts). Fires on every `session//turn_state` write (created or updated) and emits a `turn_state_changed` event to `agent::events` carrying the full new (and prior) record so the UI can derive pending approvals from state rather than from a signal event. ## Turn FSM @@ -118,9 +125,10 @@ From | [src/turn-orchestrator/main.ts](harness-node/src/turn-orchestrator/main.ts) | Binary entry point. | | [src/turn-orchestrator/register.ts](harness-node/src/turn-orchestrator/register.ts) | Composes `run::start*`, `agent::call`, `turn::step`, the abort-signal and terminal-state state triggers, and kicks off the bootstrap. | | [src/turn-orchestrator/run-start.ts](harness-node/src/turn-orchestrator/run-start.ts) | `run::start` + `run::start_and_wait` handlers and the `publishStep` helper. `executeSync` installs a terminal-state waiter, kicks the run, then races the waiter against `sync_default_timeout_ms` — no polling. | +| [src/turn-orchestrator/get-state.ts](harness-node/src/turn-orchestrator/get-state.ts) | `turn::get_state` — one-shot reader that returns the current `TurnStateRecord` for a session. UI clients call this on reload to recover in-progress modals; the orchestrator owns the state schema/key layout so clients never read iii state directly. | | [src/turn-orchestrator/on-terminal.ts](harness-node/src/turn-orchestrator/on-terminal.ts) | State trigger adapter — `turn::is_terminal_state_write` (condition) + `turn::on_terminal_state` (handler) — plus the in-process `installTerminalWaiter` / `clearTerminalWaiter` API used by `executeSync` to await a terminal `turn_state` write reactively. | | [src/turn-orchestrator/agent-call.ts](harness-node/src/turn-orchestrator/agent-call.ts) | The dispatcher chokepoint; `dispatchWithHook` runs `consultBefore` before triggering the function and returns `result` / `deny` / `pending`. | -| [src/turn-orchestrator/hook.ts](harness-node/src/turn-orchestrator/hook.ts) | `consultBefore` — publishes `agent::before_function_call` and decodes the gate reply (`allow` / `pending` / `deny`); fails closed with a `gate_unavailable` envelope. | +| [src/turn-orchestrator/hook.ts](harness-node/src/turn-orchestrator/hook.ts) | `consultBefore` — calls `policy::check_permissions` directly (5 s timeout) and maps the reply to `allow` / `pending` / `deny`; fails closed with a `gate_unavailable` envelope. `publishAfter` still routes through `hook-fanout::publish_collect` for the after-hook fanout path. | | [src/turn-orchestrator/abort.ts](harness-node/src/turn-orchestrator/abort.ts) | `performAbortSideEffects` — writes `session//abort_signal = true` and, for turns paused on approvals, one `{decision: 'aborted'}` record per pending call so the approvals state trigger wakes the turn. | | [src/turn-orchestrator/on-abort-signal.ts](harness-node/src/turn-orchestrator/on-abort-signal.ts) | State trigger adapter — `turn::is_abort_signal_set` (condition) + `turn::on_abort_signal` (handler) — publishes `turn::step_requested` whenever `session//abort_signal` is set to `true`. | | [src/turn-orchestrator/subscriber.ts](harness-node/src/turn-orchestrator/subscriber.ts) | `turn::step` durable subscriber. Skips the auto re-publish of `turn::step_requested` while the record is in `function_awaiting_approval` (the approvals state trigger owns that wake). | @@ -129,6 +137,8 @@ From | [src/turn-orchestrator/state.ts](harness-node/src/turn-orchestrator/state.ts) | `TurnState`, `TurnStateRecord` (now with `awaiting_approval?: AwaitingApprovalEntry[]`), state-key helpers. | | [src/turn-orchestrator/persistence.ts](harness-node/src/turn-orchestrator/persistence.ts) | Load/save helpers + the `session-tree::*` messages mirror. `PreparedEntry` now carries `pre_approved` so resumed turns can dispatch the call without re-asking the gate. | | [src/turn-orchestrator/events.ts](harness-node/src/turn-orchestrator/events.ts) | `emit(iii, sid, event)` — appends a sequenced `AgentEvent` to the `agent::events` stream. | +| [src/turn-orchestrator/on-record-written.ts](harness-node/src/turn-orchestrator/on-record-written.ts) | State-trigger adapter — `turn::is_stepable_record_write` (condition) + `turn::on_record_written` (handler) — directly triggers `turn::step` on every non-terminal, non-parking `turn_state` write. Replaces the imperative `publishStep` self-publish so saving the record is itself the wake. | +| [src/turn-orchestrator/on-turn-state-changed.ts](harness-node/src/turn-orchestrator/on-turn-state-changed.ts) | State-trigger adapter — `turn::is_turn_state_write` (condition) + `turn::on_turn_state_changed` (handler) — emits `turn_state_changed` to `agent::events` on every `turn_state` write (created or updated). Carries the full new (and prior) `TurnStateRecord` so the console can derive pending approvals from state rather than from a signal event. | | [src/turn-orchestrator/provider-router.ts](harness-node/src/turn-orchestrator/provider-router.ts) | Picks `provider::::stream` for the run's `provider` field. | | [src/turn-orchestrator/system-prompt.ts](harness-node/src/turn-orchestrator/system-prompt.ts) | Builds the system prompt from `run_request.system_prompt` + bootstrap skills. | | [src/turn-orchestrator/bootstrap.ts](harness-node/src/turn-orchestrator/bootstrap.ts) | Best-effort skill download via `directory::skills::download`. | diff --git a/harness-node/src/approval-gate/config.ts b/harness-node/src/approval-gate/config.ts index 9114fbc49d..996b85342e 100644 --- a/harness-node/src/approval-gate/config.ts +++ b/harness-node/src/approval-gate/config.ts @@ -1,16 +1,12 @@ import { getSection, getString } from '../runtime/config.js'; export type ApprovalGateConfig = { - topic: string; approval_state_scope: string; - policy_function_id: string; }; export function loadApprovalGateConfig(cfg: Record): ApprovalGateConfig { const section = getSection(cfg, 'approval_gate'); return { - topic: getString(section, 'topic', 'agent::before_function_call'), approval_state_scope: getString(section, 'approval_state_scope', 'approvals'), - policy_function_id: getString(section, 'policy_function_id', 'policy::check_permissions'), }; } diff --git a/harness-node/src/approval-gate/gate-subscriber.ts b/harness-node/src/approval-gate/gate-subscriber.ts deleted file mode 100644 index d529c922aa..0000000000 --- a/harness-node/src/approval-gate/gate-subscriber.ts +++ /dev/null @@ -1,131 +0,0 @@ -/** - * `policy::approval_gate` durable subscriber. Mirrors the body of - * `approval-gate/src/lib.rs::register::subscriber_fn` post-PR #150. - * - * Flow: - * 1. extract the incoming call from the durable envelope - * 2. consult policy (allow / deny / needs_approval). Allow and Deny both - * return immediately with a marked block reply (subscriber + - * approval_gate flags so the orchestrator's `publishFailureFromResponse` - * can detect that the gate actually replied — fail-closed). - * 3. needs_approval: emit `approval_requested` and return a pending block - * envelope. The orchestrator owns pending-call state on its turn record. - * - * Note: `approval_resolved` events are written by `handleResolveWithEvents` - * (in `pending.ts`) when the human resolves, not from this subscriber. This - * subscriber returns synchronously without waiting for the human decision. - */ - -import { uuidLike } from '../runtime/ids.js'; -import type { ISdk } from '../runtime/iii.js'; -import { streamSet } from '../runtime/stream.js'; -import type { ApprovalGateConfig } from './config.js'; -import { permissionsDenyEnvelope } from './denial.js'; -import { type PolicyOutcome, consultPolicy } from './policy-consult.js'; -import type { StateBus } from './state-bus.js'; -import { type IncomingCall, SUBSCRIBER_NAME, blockReplyFor, extractCall } from './types.js'; - -async function writeHookReply( - iii: ISdk, - stream_name: string, - event_id: string, - reply: unknown, -): Promise { - if (!stream_name || !event_id) return; - await streamSet(iii, { - stream_name, - group_id: event_id, - item_id: uuidLike(), - data: reply, - }); -} - -export type GateHandlerContext = { - iii: ISdk; - bus: StateBus; - cfg: ApprovalGateConfig; -}; - -export async function handleGateEvent( - ctx: GateHandlerContext, - envelope: unknown, -): Promise { - const call = extractCall(envelope); - if (!call) return { block: false }; - - // Step 1: consult policy. - const outcome = await consultPolicyOrFallback(ctx.iii, ctx.cfg, call); - - if (outcome.kind === 'allow') { - const reply = blockReplyFor({ kind: 'allow' }); - await writeHookReply(ctx.iii, call.reply_stream, call.event_id, reply); - return reply; - } - - if (outcome.kind === 'deny') { - const env = permissionsDenyEnvelope( - call.function_id, - outcome.rule_id, - outcome.matched_constraint, - call.args, - ); - const reply = { - block: true, - reason: env.reason, - denial: env, - subscriber: SUBSCRIBER_NAME, - approval_gate: true, - }; - await streamSet(ctx.iii, { - stream_name: 'agent::events', - group_id: call.session_id, - item_id: `approval-${uuidLike()}`, - data: { - type: 'function_call_denied', - function_call_id: call.function_call_id, - tool_call_id: call.function_call_id, - function_id: call.function_id, - tool_name: call.function_id, - denial: env, - }, - }); - await writeHookReply(ctx.iii, call.reply_stream, call.event_id, reply); - return reply; - } - - await streamSet(ctx.iii, { - stream_name: 'agent::events', - group_id: call.session_id, - item_id: `approval-${uuidLike()}`, - data: { - type: 'approval_requested', - function_call_id: call.function_call_id, - tool_call_id: call.function_call_id, - function_id: call.function_id, - tool_name: call.function_id, - args: call.args, - }, - }); - const reply = { - block: true, - status: 'pending' as const, - subscriber: SUBSCRIBER_NAME, - approval_gate: true, - }; - await writeHookReply(ctx.iii, call.reply_stream, call.event_id, reply); - return reply; -} - -async function consultPolicyOrFallback( - iii: ISdk, - cfg: ApprovalGateConfig, - call: IncomingCall, -): Promise { - const o = await consultPolicy(iii, cfg.policy_function_id, call.function_id, call.args); - if (o !== null) return o; - // Legacy fallback: synthesize from approval_required. - if (call.approval_required.includes(call.function_id)) { - return { kind: 'needs_approval' }; - } - return { kind: 'allow', rule_id: 'legacy/approval_required' }; -} diff --git a/harness-node/src/approval-gate/iii.worker.yaml b/harness-node/src/approval-gate/iii.worker.yaml index b37e2cdd6a..1f560ba988 100644 --- a/harness-node/src/approval-gate/iii.worker.yaml +++ b/harness-node/src/approval-gate/iii.worker.yaml @@ -4,7 +4,7 @@ language: node deploy: binary manifest: package.json bin: iii-approval-gate -description: Hook subscriber on agent::before_function_call that consults policy::check_permissions and pauses for approval::resolve when a call matches no rule. +description: Owns approval state. Registers approval::resolve and a state-trigger adapter on the approvals scope that wakes turn::step on every decision write. runtime: kind: node diff --git a/harness-node/src/approval-gate/main.ts b/harness-node/src/approval-gate/main.ts index 0612552ab5..3b2dad6724 100644 --- a/harness-node/src/approval-gate/main.ts +++ b/harness-node/src/approval-gate/main.ts @@ -5,6 +5,6 @@ import { register } from './register.js'; await bootstrapWorker({ name: 'approval-gate', description: - 'Hook subscriber on agent::before_function_call that consults policy::check_permissions and pauses calls for user approval.', + 'Owns approval state: registers approval::resolve and a state-trigger adapter on the approvals scope that wakes turn::step on every decision write.', register: (iii, ctx) => register(iii, ctx), }); diff --git a/harness-node/src/approval-gate/on-decision-written.ts b/harness-node/src/approval-gate/on-decision-written.ts index 76463d1286..d5d4c60e7d 100644 --- a/harness-node/src/approval-gate/on-decision-written.ts +++ b/harness-node/src/approval-gate/on-decision-written.ts @@ -1,13 +1,14 @@ /** * `state` trigger adapter. Registered against `scope: 'approvals'` so it fires * on every decision write. Extracts the session id from the `/` key - * and wakes the orchestrator through the normal turn step topic. + * and wakes the orchestrator by invoking turn::step directly. */ import type { ISdk } from '../runtime/iii.js'; import { logger } from '../runtime/otel.js'; +import { STEP_FN_ID } from '../turn-orchestrator/subscriber.js'; -export const STEP_TOPIC = 'turn::step_requested'; +export { STEP_FN_ID }; export const TRIGGER_FN_ID = 'approval::on_decision_written'; export const CONDITION_FN_ID = 'approval::is_decision_write'; @@ -42,11 +43,11 @@ export async function handleDecisionWritten(iii: ISdk, event: unknown): Promise< try { await iii.trigger({ - function_id: 'iii::durable::publish', - payload: { topic: STEP_TOPIC, data: { session_id } }, + function_id: STEP_FN_ID, + payload: { session_id }, }); } catch (err) { - logger.warn('approval::on_decision_written: publish failed', { + logger.warn('approval::on_decision_written: turn::step invoke failed', { session_id, err: String(err), }); diff --git a/harness-node/src/approval-gate/pending.ts b/harness-node/src/approval-gate/pending.ts index 75c725eab1..26cf705c3c 100644 --- a/harness-node/src/approval-gate/pending.ts +++ b/harness-node/src/approval-gate/pending.ts @@ -3,16 +3,12 @@ * decision record directly to the same key the paused turn reads. */ -import { requireString } from '../runtime/handler.js'; -import { uuidLike } from '../runtime/ids.js'; import type { ISdk } from '../runtime/iii.js'; import { logger } from '../runtime/otel.js'; -import { streamSet } from '../runtime/stream.js'; -import type { StateBus } from './state-bus.js'; import { pendingKey } from './types.js'; export async function handleResolve( - bus: StateBus, + iii: ISdk, state_scope: string, payload: unknown, ): Promise { @@ -35,50 +31,13 @@ export async function handleResolve( const key = pendingKey(session_id, function_call_id); try { - await bus.set(state_scope, key, { decision, reason }); + await iii.trigger({ + function_id: 'state::set', + payload: { scope: state_scope, key, value: { decision, reason } }, + }); } catch (err) { logger.error('approval-gate: failed to write resolved state', { err: String(err) }); return { ok: false, error: 'state_write_failed' }; } return { ok: true }; } - -export async function handleResolveWithEvents( - iii: ISdk, - bus: StateBus, - state_scope: string, - payload: unknown, -): Promise { - const out = await handleResolve(bus, state_scope, payload); - const result = out as Record; - if (result.ok !== true) return out; - - const obj = (payload ?? {}) as Record; - const session_id = typeof obj.session_id === 'string' ? obj.session_id : ''; - const function_call_id = - (typeof obj.function_call_id === 'string' && obj.function_call_id) || - (typeof obj.tool_call_id === 'string' && obj.tool_call_id) || - ''; - const decision = - obj.decision === 'allow' || obj.decision === 'deny' - ? (obj.decision as 'allow' | 'deny') - : 'deny'; - const reason = typeof obj.reason === 'string' ? obj.reason : null; - - await streamSet(iii, { - stream_name: 'agent::events', - group_id: session_id, - item_id: `approval-${uuidLike()}`, - data: { - type: 'approval_resolved', - function_call_id, - tool_call_id: function_call_id, - decision, - reason, - }, - }); - - return out; -} - -export { requireString }; diff --git a/harness-node/src/approval-gate/policy-consult.ts b/harness-node/src/approval-gate/policy-consult.ts index b1cbae4e05..df67d998f6 100644 --- a/harness-node/src/approval-gate/policy-consult.ts +++ b/harness-node/src/approval-gate/policy-consult.ts @@ -1,11 +1,10 @@ /** - * Consult the harness's `policy::check_permissions`. Returns `null` when - * the policy worker is unreachable (callers fall back to the legacy - * `approval_required` substring check). + * Decode the wire reply from `policy::check_permissions` into a typed + * `PolicyOutcome`. The actual `iii.trigger` to the policy function lives + * in the orchestrator's `consultBefore` (`harness-node/src/turn-orchestrator/hook.ts`) + * — that's the only caller of this module. */ -import type { ISdk } from '../runtime/iii.js'; -import { logger } from '../runtime/otel.js'; import type { MatchedConstraint } from './types.js'; export type PolicyOutcome = @@ -39,25 +38,3 @@ export function parsePolicyReply(value: unknown): PolicyOutcome { } return { kind: 'needs_approval' }; } - -export async function consultPolicy( - iii: ISdk, - policy_function_id: string, - function_id: string, - args: unknown, -): Promise { - try { - const resp = await iii.trigger({ - function_id: policy_function_id, - payload: { function_id, args }, - timeoutMs: 5_000, - }); - return parsePolicyReply(resp); - } catch (err) { - logger.debug('policy worker unreachable; falling back to approval_required list', { - policy_function_id, - err: String(err), - }); - return null; - } -} diff --git a/harness-node/src/approval-gate/register.ts b/harness-node/src/approval-gate/register.ts index ca9dd891a8..565c169c82 100644 --- a/harness-node/src/approval-gate/register.ts +++ b/harness-node/src/approval-gate/register.ts @@ -1,45 +1,27 @@ import { loadConfig } from '../runtime/config.js'; import type { ISdk } from '../runtime/iii.js'; import { loadApprovalGateConfig } from './config.js'; -import { handleGateEvent } from './gate-subscriber.js'; import { CONDITION_FN_ID as ON_DECISION_CONDITION, TRIGGER_FN_ID as ON_DECISION_FN, handleDecisionWritten, isDecisionWrite, } from './on-decision-written.js'; -import { handleResolveWithEvents } from './pending.js'; -import { IiiStateBus } from './state-bus.js'; +import { handleResolve } from './pending.js'; import { FN_RESOLVE } from './types.js'; export async function register(iii: ISdk, ctx: { configPath: string }): Promise { const cfg = loadApprovalGateConfig(await loadConfig(ctx.configPath)); - const bus = new IiiStateBus(iii); iii.registerFunction( FN_RESOLVE, - async (payload: unknown) => - handleResolveWithEvents(iii, bus, cfg.approval_state_scope, payload), + async (payload: unknown) => handleResolve(iii, cfg.approval_state_scope, payload), { description: 'Flip an approval to allow or deny. Writing the decision is itself the wake-up event.', }, ); - iii.registerFunction( - 'policy::approval_gate', - async (envelope: unknown) => handleGateEvent({ iii, bus, cfg }, envelope), - { - description: 'Consult policy::check_permissions and reply allow, deny, or pending.', - }, - ); - - iii.registerTrigger({ - type: 'durable:subscriber', - function_id: 'policy::approval_gate', - config: { topic: cfg.topic }, - }); - iii.registerFunction(ON_DECISION_CONDITION, async (event: unknown) => isDecisionWrite(event), { description: 'Condition: state event is a real approval decision write (state:created or state:updated, new_value.decision present).', @@ -50,7 +32,7 @@ export async function register(iii: ISdk, ctx: { configPath: string }): Promise< async (event: unknown) => handleDecisionWritten(iii, event), { description: - 'State trigger adapter on scope=approvals; extracts session_id from key and publishes turn::step_requested.', + 'State trigger adapter on scope=approvals; extracts session_id from key and invokes turn::step directly.', }, ); diff --git a/harness-node/src/approval-gate/state-bus.ts b/harness-node/src/approval-gate/state-bus.ts deleted file mode 100644 index c56b212536..0000000000 --- a/harness-node/src/approval-gate/state-bus.ts +++ /dev/null @@ -1,79 +0,0 @@ -/** - * Iii-backed state bus + an in-memory implementation for tests. - */ - -import type { ISdk } from '../runtime/iii.js'; -import { logger } from '../runtime/otel.js'; - -export interface StateBus { - set(scope: string, key: string, value: unknown): Promise; - get(scope: string, key: string): Promise; - listPrefix(scope: string, prefix: string): Promise; -} - -export class IiiStateBus implements StateBus { - constructor(private readonly iii: ISdk) {} - - async set(scope: string, key: string, value: unknown): Promise { - await this.iii.trigger({ - function_id: 'state::set', - payload: { scope, key, value }, - }); - } - - async get(scope: string, key: string): Promise { - try { - const v = await this.iii.trigger({ - function_id: 'state::get', - payload: { scope, key }, - }); - return v === null || v === undefined ? null : v; - } catch (err) { - logger.debug('state::get failed', { scope, key, err: String(err) }); - return null; - } - } - - async listPrefix(scope: string, prefix: string): Promise { - try { - const resp = await this.iii.trigger({ - function_id: 'state::list', - payload: { scope, prefix }, - }); - // The engine's state::list returns the values directly as an array. - // Earlier harness versions returned `{items: [...]}` so accept both. - const items = Array.isArray(resp) - ? resp - : Array.isArray((resp as { items?: unknown[] } | null)?.items) - ? ((resp as { items: unknown[] }).items as unknown[]) - : []; - return items.map((entry) => { - if (entry && typeof entry === 'object' && 'value' in (entry as Record)) { - return (entry as Record).value; - } - return entry; - }); - } catch (err) { - logger.debug('state::list failed', { scope, prefix, err: String(err) }); - return []; - } - } -} - -export class InMemoryStateBus implements StateBus { - private readonly store = new Map(); - async set(scope: string, key: string, value: unknown): Promise { - this.store.set(`${scope}/${key}`, value); - } - async get(scope: string, key: string): Promise { - return this.store.get(`${scope}/${key}`) ?? null; - } - async listPrefix(scope: string, prefix: string): Promise { - const out: unknown[] = []; - const fullPrefix = `${scope}/${prefix}`; - for (const [k, v] of this.store) { - if (k.startsWith(fullPrefix)) out.push(v); - } - return out; - } -} diff --git a/harness-node/src/approval-gate/types.ts b/harness-node/src/approval-gate/types.ts index 31e2171a9a..054173fd09 100644 --- a/harness-node/src/approval-gate/types.ts +++ b/harness-node/src/approval-gate/types.ts @@ -6,7 +6,6 @@ export const FN_RESOLVE = 'approval::resolve'; export const STATE_SCOPE = 'approvals'; export const DENIAL_SCHEMA_VERSION = 1; -export const SUBSCRIBER_NAME = 'approval-gate'; export type DeniedBy = 'permissions' | 'user' | 'gate_unavailable'; @@ -30,76 +29,8 @@ export type DenialEnvelope = { export type WireDecision = 'allow' | 'deny'; -export type IncomingCall = { - session_id: string; - function_call_id: string; - function_id: string; - args: unknown; - approval_required: string[]; - event_id: string; - reply_stream: string; -}; - export function pendingKey(session_id: string, function_call_id: string): string { if (session_id.includes('/')) throw new Error('session_id must not contain "/"'); if (function_call_id.includes('/')) throw new Error('function_call_id must not contain "/"'); return `${session_id}/${function_call_id}`; } - -export type GateDecision = { kind: 'allow' } | { kind: 'deny'; reason: string }; - -export type GateBlockReply = - | { block: false; subscriber: typeof SUBSCRIBER_NAME; approval_gate: true } - | { block: true; reason: string; subscriber: typeof SUBSCRIBER_NAME; approval_gate: true }; - -/** - * Mirrors `approval-gate/src/lib.rs::block_reply_for`. Allow and Deny replies - * both carry the `subscriber` + `approval_gate` flags so the orchestrator's - * `publish_failure_from_response` can detect that approval-gate actually - * replied (fail-closed if no such reply is present). - */ -export function blockReplyFor(decision: GateDecision): GateBlockReply { - if (decision.kind === 'allow') { - return { block: false, subscriber: SUBSCRIBER_NAME, approval_gate: true }; - } - return { - block: true, - reason: `approval-gate: ${decision.reason}`, - subscriber: SUBSCRIBER_NAME, - approval_gate: true, - }; -} - -export function extractCall(envelope: unknown): IncomingCall | null { - if (!envelope || typeof envelope !== 'object') return null; - const obj = envelope as Record; - const event_id = typeof obj.event_id === 'string' ? obj.event_id : null; - const reply_stream = typeof obj.reply_stream === 'string' ? obj.reply_stream : null; - if (!event_id || !reply_stream) return null; - const inner = - obj.payload && typeof obj.payload === 'object' ? (obj.payload as Record) : obj; - const session_id = typeof inner.session_id === 'string' ? inner.session_id : null; - const fc = - (inner.function_call as Record | undefined) ?? - (inner.tool_call as Record | undefined); - if (!session_id || !fc) return null; - const function_id = - (typeof fc.function_id === 'string' && fc.function_id) || - (typeof fc.name === 'string' && fc.name) || - null; - const id = typeof fc.id === 'string' ? fc.id : null; - if (!function_id || !id) return null; - const args = fc.arguments ?? {}; - const approval_required = Array.isArray(inner.approval_required) - ? (inner.approval_required as unknown[]).filter((s): s is string => typeof s === 'string') - : []; - return { - session_id, - function_call_id: id, - function_id, - args, - approval_required, - event_id, - reply_stream, - }; -} diff --git a/harness-node/src/index.ts b/harness-node/src/index.ts index 7e95d5ab27..f4dc1a0ede 100644 --- a/harness-node/src/index.ts +++ b/harness-node/src/index.ts @@ -45,7 +45,7 @@ const WORKERS: readonly WorkerDefinition[] = [ { name: 'approval-gate', description: - 'Hook subscriber on agent::before_function_call that consults policy::check_permissions and pauses calls for user approval.', + 'Owns approval state: registers approval::resolve and a state-trigger adapter on the approvals scope that wakes turn::step on every decision write.', register: (iii, ctx) => registerApprovalGate(iii, ctx), }, { diff --git a/harness-node/src/runtime/state.ts b/harness-node/src/runtime/state.ts index 73160a1059..62934ffc39 100644 --- a/harness-node/src/runtime/state.ts +++ b/harness-node/src/runtime/state.ts @@ -1,7 +1,6 @@ /** * Tiny `state::*` wrappers. Mirrors - * `turn-orchestrator/src/persistence.rs::state_get` / `state_set` and the - * approval-gate `IiiStateBus`. + * `turn-orchestrator/src/persistence.rs::state_get` / `state_set`. * * All helpers are tolerant: trigger errors degrade to `null` / `[]` and * are logged at warn level so a single failed read never aborts a turn. diff --git a/harness-node/src/turn-orchestrator/agent-call.ts b/harness-node/src/turn-orchestrator/agent-call.ts index 6a1df5bc97..c00602da50 100644 --- a/harness-node/src/turn-orchestrator/agent-call.ts +++ b/harness-node/src/turn-orchestrator/agent-call.ts @@ -104,8 +104,8 @@ function isTimeout(err: unknown): boolean { export async function dispatchWithHook( iii: ISdk, function_call: FunctionCall, - approval_required: string[], - session_id?: string, + session_id: string | undefined, + policy_function_id: string, ): Promise { if (!function_call.function_id || function_call.function_id.length === 0) { return { @@ -116,7 +116,7 @@ export async function dispatchWithHook( }), }; } - const outcome = await consultBefore(iii, function_call, approval_required, session_id); + const outcome = await consultBefore(iii, function_call, session_id, policy_function_id); if (outcome.kind === 'deny') return { kind: 'deny', result: denialResult(outcome.denial) }; if (outcome.kind === 'pending') { return { kind: 'pending' }; @@ -163,6 +163,7 @@ export async function dispatch( session_id: string, fn: unknown, payload: unknown, + policy_function_id: string, ): Promise { if (typeof fn !== 'string' || fn.length === 0) { return errorResult({ @@ -175,7 +176,7 @@ export async function dispatch( function_id: fn, arguments: payload ?? {}, }; - const out = await dispatchWithHook(iii, fc, [], session_id); + const out = await dispatchWithHook(iii, fc, session_id, policy_function_id); if (out.kind === 'pending') { return errorResult({ error: 'awaiting_approval', @@ -186,7 +187,7 @@ export async function dispatch( return out.result; } -export function register(iii: ISdk): void { +export function register(iii: ISdk, policy_function_id: string): void { iii.registerFunction( FUNCTION_ID, async (payload: unknown) => { @@ -194,7 +195,7 @@ export function register(iii: ISdk): void { const session_id = typeof obj.session_id === 'string' ? obj.session_id : ''; const fn = obj.function; const inner = obj.payload ?? {}; - return await dispatch(iii, session_id, fn, inner); + return await dispatch(iii, session_id, fn, inner, policy_function_id); }, { description: diff --git a/harness-node/src/turn-orchestrator/config.ts b/harness-node/src/turn-orchestrator/config.ts index f994fba2cb..a3d16df0da 100644 --- a/harness-node/src/turn-orchestrator/config.ts +++ b/harness-node/src/turn-orchestrator/config.ts @@ -1,8 +1,9 @@ -import { getNumber, getStringArray } from '../runtime/config.js'; +import { getNumber, getString, getStringArray } from '../runtime/config.js'; export type TurnOrchestratorConfig = { sync_default_timeout_ms: number; system_default_skills: string[]; + policy_function_id: string; }; export function loadOrchestratorConfig(cfg: Record): TurnOrchestratorConfig { @@ -11,5 +12,6 @@ export function loadOrchestratorConfig(cfg: Record): TurnOrches system_default_skills: getStringArray(cfg, 'system_default_skills', [ 'iii://iii-directory/index', ]), + policy_function_id: getString(cfg, 'policy_function_id', 'policy::check_permissions'), }; } diff --git a/harness-node/src/turn-orchestrator/get-state.ts b/harness-node/src/turn-orchestrator/get-state.ts new file mode 100644 index 0000000000..d30f88a4c4 --- /dev/null +++ b/harness-node/src/turn-orchestrator/get-state.ts @@ -0,0 +1,28 @@ +/** + * `turn::get_state` — one-shot reader for a session's current + * turn_state record. Exists so UI clients can recover pending modals + * after a page reload without reaching into iii state from the + * browser. The orchestrator owns the turn_state schema and key + * layout; clients call this and get back the record (or null for + * an unknown session). + */ + +import { requireString } from '../runtime/handler.js'; +import type { ISdk } from '../runtime/iii.js'; +import * as persistence from './persistence.js'; +import type { TurnStateRecord } from './state.js'; + +export const FUNCTION_ID = 'turn::get_state'; + +export async function execute(iii: ISdk, payload: unknown): Promise { + const obj = (payload ?? {}) as Record; + const session_id = requireString(obj, 'session_id'); + return persistence.loadRecord(iii, session_id); +} + +export function register(iii: ISdk): void { + iii.registerFunction(FUNCTION_ID, async (payload: unknown) => execute(iii, payload), { + description: + "Read the current turn_state record for a session. Returns null if the session is unknown. UI clients use this on page reload to recover any in-progress modals (e.g. function_awaiting_approval) without reading iii state directly.", + }); +} diff --git a/harness-node/src/turn-orchestrator/hook.ts b/harness-node/src/turn-orchestrator/hook.ts index 33653a23ad..e35f47819b 100644 --- a/harness-node/src/turn-orchestrator/hook.ts +++ b/harness-node/src/turn-orchestrator/hook.ts @@ -1,36 +1,28 @@ /** - * `agent::before_function_call` hook. Mirrors - * `turn-orchestrator/src/hook.rs`. + * Approval consultation. Calls `policy::check_permissions` directly and maps + * the reply to allow / deny / pending. Fail-closed on transport errors: + * unreachable policy → deny with `gate_unavailable`. * - * Fail-closed: any hook failure (timeout, error, no subscriber) returns - * `Deny` with `denied_by: gate_unavailable` so dispatch never silently - * allows a call. + * `publishAfter` still goes through hook-fanout because the after-hook is a + * pluggable merge point with multiple potential consumers. */ +import { permissionsDenyEnvelope } from '../approval-gate/denial.js'; +import { parsePolicyReply } from '../approval-gate/policy-consult.js'; +import { DENIAL_SCHEMA_VERSION } from '../approval-gate/types.js'; +import type { DenialEnvelope } from '../approval-gate/types.js'; import type { ISdk } from '../runtime/iii.js'; +export type { DeniedBy, DenialEnvelope } from '../approval-gate/types.js'; import { logger } from '../runtime/otel.js'; import type { FunctionCall } from '../types/function.js'; -export const TOPIC_BEFORE = 'agent::before_function_call'; export const TOPIC_AFTER = 'agent::after_function_call'; -export const HOOK_TIMEOUT_MS = 10_000; -export const DENIAL_SCHEMA_VERSION = 1; -export type DenialEnvelope = { - schema_version: number; - status: 'denied'; - denied_by: 'permissions' | 'user' | 'gate_unavailable'; - function_id: string; - rule_id?: string; - rule_action?: 'deny'; - matched_constraint?: { field: string; operator: string; value: unknown }; - args_excerpt?: unknown; - reason: string; -}; +export const HOOK_TIMEOUT_MS = 500; export type HookOutcome = | { kind: 'allow' } - | { kind: 'pending'; merged: Record } + | { kind: 'pending' } | { kind: 'deny'; denial: DenialEnvelope | Record }; export function gateUnavailableEnvelope(function_id: string, reason: string): DenialEnvelope { @@ -46,33 +38,18 @@ export function gateUnavailableEnvelope(function_id: string, reason: string): De export async function consultBefore( iii: ISdk, function_call: FunctionCall, - approval_required: string[], - session_id?: string, + _session_id: string | undefined, + policy_function_id: string, ): Promise { - // PR #150: include session_id at the inner-payload root so approval-gate's - // extractCall can route it. Without this, the gate sees a null session_id, - // returns {block:false}, and the call passes through unapproved. - const inner = session_id - ? { session_id, function_call, approval_required } - : { function_call, approval_required }; - const payload = { - topic: TOPIC_BEFORE, - payload: inner, - merge_rule: 'first_block_wins', - timeout_ms: HOOK_TIMEOUT_MS, - }; - let merged: Record; + let raw: unknown; try { - const resp = await iii.trigger({ - function_id: 'hook-fanout::publish_collect', - payload, + raw = await iii.trigger({ + function_id: policy_function_id, + payload: { function_id: function_call.function_id, args: function_call.arguments }, + timeoutMs: 5_000, }); - merged = - resp?.merged && typeof resp.merged === 'object' - ? (resp.merged as Record) - : {}; } catch (err) { - logger.warn('before_function_call hook failed; failing closed', { + logger.warn('policy consult failed; failing closed', { function_id: function_call.function_id, err: String(err), }); @@ -80,22 +57,25 @@ export async function consultBefore( kind: 'deny', denial: gateUnavailableEnvelope( function_call.function_id, - `permission gate did not respond: ${String(err)}`, + `policy unreachable: ${String(err)}`, ), }; } - const blocked = merged.block === true; - if (!blocked) return { kind: 'allow' }; - if (merged.status === 'pending') { - return { kind: 'pending', merged }; + + const decision = parsePolicyReply(raw); + if (decision.kind === 'allow') return { kind: 'allow' }; + if (decision.kind === 'deny') { + return { + kind: 'deny', + denial: permissionsDenyEnvelope( + function_call.function_id, + decision.rule_id, + decision.matched_constraint, + function_call.arguments, + ), + }; } - const denial = - (merged.denial as DenialEnvelope | undefined) ?? - gateUnavailableEnvelope( - function_call.function_id, - typeof merged.reason === 'string' ? merged.reason : 'Permission denied.', - ); - return { kind: 'deny', denial }; + return { kind: 'pending' }; } export async function publishAfter( @@ -119,4 +99,3 @@ export async function publishAfter( return null; } } -// reload marker 1779108905 diff --git a/harness-node/src/turn-orchestrator/on-record-written.ts b/harness-node/src/turn-orchestrator/on-record-written.ts new file mode 100644 index 0000000000..28c48f081e --- /dev/null +++ b/harness-node/src/turn-orchestrator/on-record-written.ts @@ -0,0 +1,106 @@ +/** + * Self-loop wake: a state trigger on `scope: 'agent'` filtered by the + * turn_state key shape and a stepable state TRANSITION (new state differs + * from old, non-terminal, non-awaiting) invokes `turn::step`. Saving the + * record on a real transition is the wake — replaces the durable + * `turn::step_requested` self-publish that used to live in `subscriber.ts`. + * + * Same-state writes (e.g. `handlePrepare` calling `saveRecord` while still + * in `function_prepare` to persist normalized calls) MUST NOT wake step, + * otherwise the orchestrator races itself: a duplicate `turn::step` runs + * the same handler again, re-emitting events and re-persisting prepared + * calls. We filter those out by requiring `new_value.state !== old_value.state`. + */ + +import type { ISdk } from '../runtime/iii.js'; +import { logger } from '../runtime/otel.js'; +import type { TurnState } from './state.js'; +import { STEP_FN_ID, STEP_TOPIC } from './subscriber.js'; + +export { STEP_FN_ID }; +export const HANDLER_FN_ID = 'turn::on_record_written'; +export const CONDITION_FN_ID = 'turn::is_stepable_record_write'; + +const TURN_STATE_KEY_RE = /^session\/(?[^/]+)\/turn_state$/; + +const NON_STEPABLE_STATES: ReadonlySet = new Set([ + 'stopped', + 'function_awaiting_approval', +]); + +type StepableWrite = { + session_id: string; + state: TurnState; +}; + +/** + * One source of truth for "is this a stepable turn_state write?". Returns the + * extracted session_id + state on match, null otherwise. Both the condition + * (boolean check) and the handler (needs the session_id) route through this, + * so they can't drift — and the handler can't fire on a parking-state write + * even if the condition was bypassed. + */ +function parseStepableWrite(event: unknown): StepableWrite | null { + if (!event || typeof event !== 'object') return null; + const obj = event as Record; + + if (obj.event_type !== 'state:created' && obj.event_type !== 'state:updated') return null; + + const key = obj.key; + if (typeof key !== 'string') return null; + const session_id = TURN_STATE_KEY_RE.exec(key)?.groups?.session_id; + if (!session_id) return null; + + const nv = obj.new_value; + if (!nv || typeof nv !== 'object') return null; + const state = (nv as Record).state; + if (typeof state !== 'string') return null; + if (NON_STEPABLE_STATES.has(state as TurnState)) return null; + + if (obj.event_type === 'state:updated') { + const ov = obj.old_value; + const old_state = + ov && typeof ov === 'object' ? (ov as Record).state : undefined; + if (typeof old_state === 'string' && old_state === state) return null; + } + + return { session_id, state: state as TurnState }; +} + +export function isStepableRecordWrite(event: unknown): boolean { + return parseStepableWrite(event) !== null; +} + +export async function handleStepableRecordWrite(iii: ISdk, event: unknown): Promise { + const parsed = parseStepableWrite(event); + if (!parsed) return; + + try { + await iii.trigger({ + function_id: STEP_FN_ID, + payload: { session_id: parsed.session_id }, + }); + return; + } catch (err) { + // Direct invoke failed (timeout, transient throw, etc). The triggering + // state write already landed, so without a durable retry the session + // would sit stuck in this state forever. Fall back to publishing + // `turn::step_requested` so the durable subscriber on `subscriber.ts` + // buffers + retries. + logger.warn( + 'turn::on_record_written: direct turn::step failed; falling back to durable publish', + { session_id: parsed.session_id, err: String(err) }, + ); + try { + await iii.trigger({ + function_id: 'iii::durable::publish', + payload: { topic: STEP_TOPIC, data: { session_id: parsed.session_id } }, + }); + } catch (publishErr) { + logger.error( + 'turn::on_record_written: durable publish fallback also failed; session may be stuck', + { session_id: parsed.session_id, err: String(publishErr) }, + ); + } + } +} diff --git a/harness-node/src/turn-orchestrator/on-turn-state-changed.ts b/harness-node/src/turn-orchestrator/on-turn-state-changed.ts new file mode 100644 index 0000000000..a1e1894250 --- /dev/null +++ b/harness-node/src/turn-orchestrator/on-turn-state-changed.ts @@ -0,0 +1,64 @@ +/** + * State-trigger adapter that mirrors `on-record-written` but emits a + * `turn_state_changed` agent event instead of triggering `turn::step`. + * Gives the frontend a live signal carrying the new turn_state record + * so it can derive pending approvals from state directly. + */ + +import type { ISdk } from '../runtime/iii.js'; +import { logger } from '../runtime/otel.js'; +import { emit } from './events.js'; + +export const HANDLER_FN_ID = 'turn::on_turn_state_changed'; +export const CONDITION_FN_ID = 'turn::is_turn_state_write'; + +const TURN_STATE_KEY_RE = /^session\/(?[^/]+)\/turn_state$/; + +type ParsedWrite = { + session_id: string; + event_type: 'state:created' | 'state:updated'; + new_value: Record; + old_value?: Record; +}; + +function parseWrite(event: unknown): ParsedWrite | null { + if (!event || typeof event !== 'object') return null; + const obj = event as Record; + if (obj.event_type !== 'state:created' && obj.event_type !== 'state:updated') return null; + const key = obj.key; + if (typeof key !== 'string') return null; + const session_id = TURN_STATE_KEY_RE.exec(key)?.groups?.session_id; + if (!session_id) return null; + const nv = obj.new_value; + if (!nv || typeof nv !== 'object') return null; + const ov = obj.old_value; + return { + session_id, + event_type: obj.event_type, + new_value: nv as Record, + old_value: ov && typeof ov === 'object' ? (ov as Record) : undefined, + }; +} + +export function isTurnStateWrite(event: unknown): boolean { + return parseWrite(event) !== null; +} + +export async function handleTurnStateWrite(iii: ISdk, event: unknown): Promise { + const parsed = parseWrite(event); + if (!parsed) return; + + try { + await emit(iii, parsed.session_id, { + type: 'turn_state_changed', + event_type: parsed.event_type, + new_value: parsed.new_value, + ...(parsed.old_value !== undefined && { old_value: parsed.old_value }), + }); + } catch (err) { + logger.warn('turn::on_turn_state_changed: emit failed', { + session_id: parsed.session_id, + err: String(err), + }); + } +} diff --git a/harness-node/src/turn-orchestrator/register.ts b/harness-node/src/turn-orchestrator/register.ts index 76f307890f..a1ff339160 100644 --- a/harness-node/src/turn-orchestrator/register.ts +++ b/harness-node/src/turn-orchestrator/register.ts @@ -3,12 +3,25 @@ import type { ISdk } from '../runtime/iii.js'; import { register as registerAgentCall } from './agent-call.js'; import * as bootstrap from './bootstrap.js'; import { loadOrchestratorConfig } from './config.js'; +import { register as registerGetState } from './get-state.js'; import { CONDITION_FN_ID as ABORT_CONDITION_FN, HANDLER_FN_ID as ABORT_HANDLER_FN, handleAbortSignalWrite, isAbortSignalWrite, } from './on-abort-signal.js'; +import { + CONDITION_FN_ID as RECORD_CONDITION_FN, + HANDLER_FN_ID as RECORD_HANDLER_FN, + handleStepableRecordWrite, + isStepableRecordWrite, +} from './on-record-written.js'; +import { + CONDITION_FN_ID as TURN_STATE_CHANGED_CONDITION_FN, + HANDLER_FN_ID as TURN_STATE_CHANGED_HANDLER_FN, + handleTurnStateWrite, + isTurnStateWrite, +} from './on-turn-state-changed.js'; import { CONDITION_FN_ID as TERMINAL_CONDITION_FN, HANDLER_FN_ID as TERMINAL_HANDLER_FN, @@ -22,11 +35,10 @@ export async function register(iii: ISdk, ctx: { configPath: string }): Promise< const cfg = await loadConfig(ctx.configPath); const orchestratorCfg = loadOrchestratorConfig(cfg); registerRunStart(iii, orchestratorCfg); - registerAgentCall(iii); + registerAgentCall(iii, orchestratorCfg.policy_function_id); registerSubscriber(iii, orchestratorCfg); + registerGetState(iii); - // Reactive abort wake. Mirrors the pattern in - // harness/fanout/sessions-poll.ts (post c59210788c). iii.registerFunction(ABORT_CONDITION_FN, async (event: unknown) => isAbortSignalWrite(event), { description: 'Condition: state event sets session//abort_signal = true (state:created or state:updated).', @@ -50,10 +62,6 @@ export async function register(iii: ISdk, ctx: { configPath: string }): Promise< }, }); - // Reactive `run::start_and_wait` wake. The condition matches terminal - // turn_state writes; the handler resolves the in-process waiter - // installed by executeSync. Replaces the old sync_poll_interval_ms - // sleep loop. iii.registerFunction( TERMINAL_CONDITION_FN, async (event: unknown) => isTerminalStateWrite(event), @@ -82,6 +90,59 @@ export async function register(iii: ISdk, ctx: { configPath: string }): Promise< }, }); + iii.registerFunction( + RECORD_CONDITION_FN, + async (event: unknown) => isStepableRecordWrite(event), + { + description: + 'Condition: state event sets session//turn_state to a stepable state (excludes stopped + function_awaiting_approval).', + }, + ); + + iii.registerFunction( + RECORD_HANDLER_FN, + async (event: unknown) => handleStepableRecordWrite(iii, event), + { + description: + 'State trigger adapter on scope=agent for stepable turn_state writes; invokes turn::step. Replaces the imperative publishStep self-publish.', + }, + ); + + iii.registerTrigger({ + type: 'state', + function_id: RECORD_HANDLER_FN, + config: { + scope: 'agent', + condition_function_id: RECORD_CONDITION_FN, + }, + }); + + iii.registerFunction( + TURN_STATE_CHANGED_CONDITION_FN, + async (event: unknown) => isTurnStateWrite(event), + { + description: 'Condition: state event is a write to session//turn_state.', + }, + ); + + iii.registerFunction( + TURN_STATE_CHANGED_HANDLER_FN, + async (event: unknown) => handleTurnStateWrite(iii, event), + { + description: + 'State trigger adapter on scope=agent for turn_state writes; emits turn_state_changed on agent::events for the subscribed UI.', + }, + ); + + iii.registerTrigger({ + type: 'state', + function_id: TURN_STATE_CHANGED_HANDLER_FN, + config: { + scope: 'agent', + condition_function_id: TURN_STATE_CHANGED_CONDITION_FN, + }, + }); + // Bootstrap best-effort skill download in the background. void bootstrap.run(iii, orchestratorCfg); } diff --git a/harness-node/src/turn-orchestrator/run-start.ts b/harness-node/src/turn-orchestrator/run-start.ts index 70edc5139f..d8524a6416 100644 --- a/harness-node/src/turn-orchestrator/run-start.ts +++ b/harness-node/src/turn-orchestrator/run-start.ts @@ -5,7 +5,6 @@ import { requireString } from '../runtime/handler.js'; import type { ISdk } from '../runtime/iii.js'; -import { logger } from '../runtime/otel.js'; import type { AgentEvent } from '../types/agent-event.js'; import type { AgentMessage } from '../types/agent-message.js'; import type { TurnOrchestratorConfig } from './config.js'; @@ -16,21 +15,6 @@ import { newRecord } from './state.js'; export const FUNCTION_ID = 'run::start'; export const SYNC_FUNCTION_ID = 'run::start_and_wait'; -export const STEP_TOPIC = 'turn::step_requested'; - -export async function publishStep(iii: ISdk, session_id: string): Promise { - try { - await iii.trigger({ - function_id: 'iii::durable::publish', - payload: { topic: STEP_TOPIC, data: { session_id } }, - }); - } catch (err) { - logger.warn('turn::step_requested publish failed', { - session_id, - err: String(err), - }); - } -} function buildRunRequest(payload: Record): Record { return { @@ -38,7 +22,6 @@ function buildRunRequest(payload: Record): Record { - const runRequest = await persistence.loadRunRequest(iii, rec.session_id); - const approval_required = Array.isArray(runRequest.approval_required) - ? (runRequest.approval_required as string[]).filter((x) => typeof x === 'string') - : []; - +export async function handleExecute( + iii: ISdk, + cfg: TurnOrchestratorConfig, + rec: TurnStateRecord, +): Promise { const prepared = await persistence.loadPreparedCalls(iii, rec.session_id); const results = await persistence.loadExecutedCalls(iii, rec.session_id); @@ -80,12 +80,19 @@ export async function handleExecute(iii: ISdk, rec: TurnStateRecord): Promise({ - function_id: fc.function_id, - payload: fc.arguments ?? {}, - }); - const result = decodeOrPassthroughResult(value); - const is_error = isErrorResult(result); + let result: FunctionResult; + let is_error: boolean; + try { + const value = await iii.trigger({ + function_id: fc.function_id, + payload: fc.arguments ?? {}, + }); + result = decodeOrPassthroughResult(value); + is_error = isErrorResult(result); + } catch (err) { + result = triggerErrorResult(fc.function_id, err); + is_error = true; + } persistence.upsertExecutedCall(results, { function_call: fc, result, is_error }); await persistence.saveExecutedCalls(iii, rec.session_id, results); await emit(iii, rec.session_id, buildFunctionExecutionEnd(fc, result, is_error)); @@ -124,7 +131,7 @@ export async function handleExecute(iii: ISdk, rec: TurnStateRecord): Promise).message === 'string' + ? ((err as Record).message as string) + : String(err); + const details = { + error: 'trigger_failed', + function: function_id, + message, + }; + return { + content: [text(JSON.stringify(details))], + details, + terminate: false, + }; +} + function decodeOrPassthroughResult(value: unknown): FunctionResult { if ( value && diff --git a/harness-node/src/turn-orchestrator/subscriber.ts b/harness-node/src/turn-orchestrator/subscriber.ts index 45c4ef494d..99d3662357 100644 --- a/harness-node/src/turn-orchestrator/subscriber.ts +++ b/harness-node/src/turn-orchestrator/subscriber.ts @@ -1,18 +1,16 @@ /** - * `turn::step` durable subscriber. Mirrors - * `turn-orchestrator/src/subscriber.rs`. + * `turn::step` durable subscriber. */ import type { ISdk } from '../runtime/iii.js'; import { logger } from '../runtime/otel.js'; import type { TurnOrchestratorConfig } from './config.js'; import * as persistence from './persistence.js'; -import { publishStep } from './run-start.js'; import { isTerminal } from './state.js'; import { step } from './transitions.js'; export const STEP_FN_ID = 'turn::step'; -const STEP_TOPIC = 'turn::step_requested'; +export const STEP_TOPIC = 'turn::step_requested'; function extractSessionId(payload: unknown): string | null { if (!payload || typeof payload !== 'object') return null; @@ -50,9 +48,6 @@ export async function execute( throw new Error(`transition from ${from_state} failed: ${String(err)}`); } await persistence.saveRecord(iii, rec); - if (!isTerminal(rec) && rec.state !== 'function_awaiting_approval') { - await publishStep(iii, session_id); - } return { ok: true, from_state, to_state: rec.state }; } diff --git a/harness-node/src/turn-orchestrator/transitions.ts b/harness-node/src/turn-orchestrator/transitions.ts index 0ca196d1a9..6ccc07d05d 100644 --- a/harness-node/src/turn-orchestrator/transitions.ts +++ b/harness-node/src/turn-orchestrator/transitions.ts @@ -1,41 +1,44 @@ -import type { ISdk } from '../runtime/iii.js' -import type { TurnOrchestratorConfig } from './config.js' -import type { TurnStateRecord } from './state.js' -import { handleAwaiting, handleFinished, handleStreaming } from './states/assistant.js' +import type { ISdk } from '../runtime/iii.js'; +import type { TurnOrchestratorConfig } from './config.js'; +import type { TurnStateRecord } from './state.js'; +import { handleAwaiting, handleFinished, handleStreaming } from './states/assistant.js'; import { handleAwaitingApproval, handleExecute, handleFinalize, handlePrepare, -} from './states/functions.js' -import { handleProvisioning } from './states/provisioning.js' -import { handleSteering } from './states/steering.js' -import { handleTearingDown } from './states/tearing-down.js' +} from './states/functions.js'; +import { handleProvisioning } from './states/provisioning.js'; +import { handleSteering } from './states/steering.js'; +import { handleTearingDown } from './states/tearing-down.js'; -export async function step(iii: ISdk, cfg: TurnOrchestratorConfig, rec: TurnStateRecord): Promise { - console.log('step transition from', rec.state) +export async function step( + iii: ISdk, + cfg: TurnOrchestratorConfig, + rec: TurnStateRecord, +): Promise { switch (rec.state) { case 'provisioning': - return handleProvisioning(iii, cfg, rec) + return handleProvisioning(iii, cfg, rec); case 'awaiting_assistant': - return handleAwaiting(iii, rec) + return handleAwaiting(iii, rec); case 'assistant_streaming': - return handleStreaming(iii, rec) + return handleStreaming(iii, rec); case 'assistant_finished': - return handleFinished(iii, rec) + return handleFinished(iii, rec); case 'function_prepare': - return handlePrepare(iii, rec) + return handlePrepare(iii, rec); case 'function_execute': - return handleExecute(iii, rec) + return handleExecute(iii, cfg, rec); case 'function_awaiting_approval': - return handleAwaitingApproval(iii, rec) + return handleAwaitingApproval(iii, rec); case 'function_finalize': - return handleFinalize(iii, rec) + return handleFinalize(iii, rec); case 'steering_check': - return handleSteering(iii, rec) + return handleSteering(iii, rec); case 'tearing_down': - return handleTearingDown(iii, rec) + return handleTearingDown(iii, rec); case 'stopped': - return // idempotent terminal + return; // idempotent terminal } } diff --git a/harness-node/src/types/agent-event.ts b/harness-node/src/types/agent-event.ts index 93118ab91d..f2bd95353e 100644 --- a/harness-node/src/types/agent-event.ts +++ b/harness-node/src/types/agent-event.ts @@ -10,8 +10,6 @@ import type { AgentMessage, FunctionResultMessage } from './agent-message.js'; import type { FunctionResult } from './function.js'; import type { AssistantMessageEvent } from './stream-event.js'; -export type ApprovalDecision = 'allow' | 'deny'; - export type AgentEvent = | { type: 'agent_start' } | { type: 'agent_end'; messages: AgentMessage[] } @@ -49,14 +47,10 @@ export type AgentEvent = is_error: boolean; } | { - type: 'approval_requested'; - function_call_id: string; - function_id: string; - args: unknown; - } - | { - type: 'approval_resolved'; - function_call_id: string; - decision: ApprovalDecision; - reason?: string | null; + type: 'turn_state_changed'; + /** Full new turn_state record (the orchestrator's persisted state). */ + new_value: Record; + /** Previous record, present on state:updated; absent on state:created. */ + old_value?: Record; + event_type: 'state:created' | 'state:updated'; }; diff --git a/harness-node/tests/approval-gate/gate-subscriber.test.ts b/harness-node/tests/approval-gate/gate-subscriber.test.ts deleted file mode 100644 index 322115e3c9..0000000000 --- a/harness-node/tests/approval-gate/gate-subscriber.test.ts +++ /dev/null @@ -1,178 +0,0 @@ -import { describe, expect, it } from 'vitest'; -import type { ApprovalGateConfig } from '../../src/approval-gate/config.js'; -import { handleGateEvent } from '../../src/approval-gate/gate-subscriber.js'; -import { InMemoryStateBus } from '../../src/approval-gate/state-bus.js'; -import { STATE_SCOPE, pendingKey } from '../../src/approval-gate/types.js'; -import type { ISdk } from '../../src/runtime/iii.js'; - -type TriggerCall = { function_id: string; payload: unknown }; - -function fakeIii(handler: (call: TriggerCall) => unknown = () => undefined): { - iii: ISdk; - calls: TriggerCall[]; -} { - const calls: TriggerCall[] = []; - const iii = { - trigger: async (req: { function_id: string; payload: T }): Promise => { - const call = { function_id: req.function_id, payload: req.payload }; - calls.push(call); - const result = handler(call); - if (result instanceof Error) throw result; - return result as R; - }, - } as unknown as ISdk; - return { iii, calls }; -} - -const cfg: ApprovalGateConfig = { - approval_state_scope: STATE_SCOPE, - policy_function_id: 'policy::check_permissions', - topic: 'agent::before_function_call', -}; - -function approvalRequiredEnvelope() { - return { - event_id: 'evt-1', - reply_stream: 'rs-1', - payload: { - session_id: 's1', - function_call: { id: 'tc-1', function_id: 'shell::exec', arguments: { command: 'date' } }, - approval_required: ['shell::exec'], - }, - }; -} - -describe('handleGateEvent (post PR #150)', () => { - it('returns {block:false} for an envelope it cannot parse', async () => { - const { iii } = fakeIii(); - const reply = (await handleGateEvent( - { iii, bus: new InMemoryStateBus(), cfg }, - { not: 'an envelope' }, - )) as Record; - expect(reply.block).toBe(false); - }); - - it('does not write to state-bus on needs_approval; only emits event and returns pending', async () => { - const bus = new InMemoryStateBus(); - const { iii, calls } = fakeIii((call) => { - if (call.function_id === 'policy::check_permissions') return null; // legacy fallback - return undefined; - }); - const reply = (await handleGateEvent({ iii, bus, cfg }, approvalRequiredEnvelope())) as Record< - string, - unknown - >; - - expect(reply.block).toBe(true); - expect(reply.status).toBe('pending'); - expect(reply.subscriber).toBe('approval-gate'); - expect(reply.approval_gate).toBe(true); - - expect(await bus.get(STATE_SCOPE, pendingKey('s1', 'tc-1'))).toBeNull(); - - const requestedEvent = calls - .filter((c) => c.function_id === 'stream::set') - .map((c) => (c.payload as Record).data as Record) - .find((d) => d.type === 'approval_requested'); - expect(requestedEvent).toBeDefined(); - expect(requestedEvent?.function_call_id).toBe('tc-1'); - expect(requestedEvent).not.toHaveProperty('expires_at'); - }); - - it('stays stateless when policy needs approval even if approval_required is empty', async () => { - const bus = new InMemoryStateBus(); - const { iii } = fakeIii((call) => { - if (call.function_id === 'policy::check_permissions') { - return { decision: 'needs_approval' }; - } - return undefined; - }); - const envelope = { - event_id: 'evt-1', - reply_stream: 'rs-1', - payload: { - session_id: 's1', - function_call: { id: 'tc-1', function_id: 'shell::exec', arguments: { command: 'date' } }, - approval_required: [], - }, - }; - const reply = (await handleGateEvent({ iii, bus, cfg }, envelope)) as Record; - - expect(reply.block).toBe(true); - expect(reply.status).toBe('pending'); - expect(reply.approval_gate).toBe(true); - expect(await bus.get(STATE_SCOPE, pendingKey('s1', 'tc-1'))).toBeNull(); - }); - - it('returns a marked allow reply when policy explicitly returns allow', async () => { - const bus = new InMemoryStateBus(); - const { iii } = fakeIii((call) => { - if (call.function_id === 'policy::check_permissions') { - return { decision: 'allow', rule_id: 'test/allow' }; - } - return undefined; - }); - const envelope = { - event_id: 'evt-1', - reply_stream: 'rs-1', - payload: { - session_id: 's1', - function_call: { id: 'tc-1', function_id: 'shell::fs::ls', arguments: {} }, - approval_required: ['shell::exec'], - }, - }; - const reply = (await handleGateEvent({ iii, bus, cfg }, envelope)) as Record; - - expect(reply.block).toBe(false); - expect(reply.subscriber).toBe('approval-gate'); - expect(reply.approval_gate).toBe(true); - expect(await bus.get(STATE_SCOPE, pendingKey('s1', 'tc-1'))).toBeNull(); - }); - - it('returns pending on needs_approval even when approval_required is empty (policy is source of truth)', async () => { - const bus = new InMemoryStateBus(); - const { iii } = fakeIii((call) => { - if (call.function_id === 'policy::check_permissions') { - return { decision: 'needs_approval' }; - } - return undefined; - }); - const envelope = { - event_id: 'evt-1', - reply_stream: 'rs-1', - payload: { - session_id: 's1', - function_call: { id: 'tc-1', function_id: 'shell::exec', arguments: { command: 'date' } }, - approval_required: [], - }, - }; - const reply = (await handleGateEvent({ iii, bus, cfg }, envelope)) as Record; - - expect(reply.block).toBe(true); - expect(reply.status).toBe('pending'); - expect(await bus.get(STATE_SCOPE, pendingKey('s1', 'tc-1'))).toBeNull(); - }); - - it('returns a deny envelope when policy denies', async () => { - const bus = new InMemoryStateBus(); - const { iii } = fakeIii((call) => { - if (call.function_id === 'policy::check_permissions') { - return { - decision: 'deny', - rule_id: 'deny-rule', - matched_constraint: { field: 'command', operator: 'eq', value: 'rm -rf /' }, - }; - } - return undefined; - }); - const reply = (await handleGateEvent({ iii, bus, cfg }, approvalRequiredEnvelope())) as Record< - string, - unknown - >; - - expect(reply.block).toBe(true); - expect(reply.subscriber).toBe('approval-gate'); - expect(reply.approval_gate).toBe(true); - expect(reply.denial).toBeDefined(); - }); -}); diff --git a/harness-node/tests/approval-gate/on-decision-written.test.ts b/harness-node/tests/approval-gate/on-decision-written.test.ts index 41ecfaebd6..537c871648 100644 --- a/harness-node/tests/approval-gate/on-decision-written.test.ts +++ b/harness-node/tests/approval-gate/on-decision-written.test.ts @@ -1,13 +1,13 @@ import { describe, expect, it, vi } from 'vitest'; import { - STEP_TOPIC, + STEP_FN_ID, handleDecisionWritten, isDecisionWrite, } from '../../src/approval-gate/on-decision-written.js'; import type { ISdk } from '../../src/runtime/iii.js'; describe('handleDecisionWritten', () => { - it('extracts session_id from key and publishes turn::step_requested', async () => { + it('extracts session_id from key and triggers turn::step directly', async () => { const triggers: Array<{ function_id: string; payload: unknown }> = []; const iii = { trigger: vi.fn(async (req: { function_id: string; payload: unknown }) => { @@ -28,11 +28,8 @@ describe('handleDecisionWritten', () => { await handleDecisionWritten(iii, event); expect(triggers).toHaveLength(1); - expect(triggers[0]?.function_id).toBe('iii::durable::publish'); - expect(triggers[0]?.payload).toMatchObject({ - topic: STEP_TOPIC, - data: { session_id: 'sess-abc' }, - }); + expect(triggers[0]?.function_id).toBe(STEP_FN_ID); + expect(triggers[0]?.payload).toMatchObject({ session_id: 'sess-abc' }); }); it('handles keys with multiple slashes by taking only the prefix before the first slash', async () => { @@ -53,7 +50,7 @@ describe('handleDecisionWritten', () => { message_type: 'state', }); - expect(triggers[0]?.payload).toMatchObject({ data: { session_id: 'sess-1' } }); + expect(triggers[0]?.payload).toMatchObject({ session_id: 'sess-1' }); }); it('no-ops when the key has no slash separator', async () => { @@ -100,10 +97,10 @@ describe('handleDecisionWritten', () => { }); expect(triggers).toHaveLength(1); - expect(triggers[0]?.payload).toMatchObject({ data: { session_id: 'sess-xyz' } }); + expect(triggers[0]?.payload).toMatchObject({ session_id: 'sess-xyz' }); }); - it('still publishes when handed a state:deleted event directly (condition would have blocked at engine)', async () => { + it('still triggers when handed a state:deleted event directly (condition would have blocked at engine)', async () => { // Documents that the handler is no longer the filter — the condition is. // In production, the engine ensures this never happens because the // condition function returns false for state:deleted. This test asserts @@ -126,7 +123,7 @@ describe('handleDecisionWritten', () => { message_type: 'state', }); - // No filter in the handler → publishes anyway. Production safety comes from + // No filter in the handler → triggers anyway. Production safety comes from // the condition function registered alongside this handler. expect(triggers).toHaveLength(1); }); diff --git a/harness-node/tests/approval-gate/pending.test.ts b/harness-node/tests/approval-gate/pending.test.ts index 7153ed02c4..60c5c61f6e 100644 --- a/harness-node/tests/approval-gate/pending.test.ts +++ b/harness-node/tests/approval-gate/pending.test.ts @@ -1,44 +1,72 @@ import { describe, expect, it, vi } from 'vitest'; -import { handleResolve, handleResolveWithEvents } from '../../src/approval-gate/pending.js'; -import { InMemoryStateBus } from '../../src/approval-gate/state-bus.js'; +import { handleResolve } from '../../src/approval-gate/pending.js'; import type { ISdk } from '../../src/runtime/iii.js'; +function fakeIii(): { + iii: ISdk; + setCalls: Array<{ scope: string; key: string; value: unknown }>; + streamSets: unknown[]; +} { + const setCalls: Array<{ scope: string; key: string; value: unknown }> = []; + const streamSets: unknown[] = []; + const iii = { + trigger: vi.fn(async ({ function_id, payload }: { function_id: string; payload: unknown }) => { + if (function_id === 'state::set') { + setCalls.push(payload as { scope: string; key: string; value: unknown }); + return null; + } + if (function_id === 'stream::set') { + streamSets.push(payload); + return null; + } + return null; + }), + } as unknown as ISdk; + return { iii, setCalls, streamSets }; +} + describe('handleResolve (simplified)', () => { - it('writes { decision, reason } to the state-bus on allow', async () => { - const bus = new InMemoryStateBus(); - const out = await handleResolve(bus, 'approvals', { + it('writes { decision, reason } to state via state::set on allow', async () => { + const { iii, setCalls } = fakeIii(); + const out = await handleResolve(iii, 'approvals', { session_id: 's1', function_call_id: 'fc-1', decision: 'allow', }); expect(out).toEqual({ ok: true }); - expect(await bus.get('approvals', 's1/fc-1')).toEqual({ decision: 'allow', reason: null }); + expect(setCalls).toHaveLength(1); + expect(setCalls[0]).toEqual({ + scope: 'approvals', + key: 's1/fc-1', + value: { decision: 'allow', reason: null }, + }); }); it('preserves a reason when provided on deny', async () => { - const bus = new InMemoryStateBus(); - const out = await handleResolve(bus, 'approvals', { + const { iii, setCalls } = fakeIii(); + const out = await handleResolve(iii, 'approvals', { session_id: 's1', function_call_id: 'fc-1', decision: 'deny', reason: 'user cancelled', }); expect(out).toEqual({ ok: true }); - expect(await bus.get('approvals', 's1/fc-1')).toEqual({ - decision: 'deny', - reason: 'user cancelled', + expect(setCalls[0]).toEqual({ + scope: 'approvals', + key: 's1/fc-1', + value: { decision: 'deny', reason: 'user cancelled' }, }); }); it('returns missing_id when ids are absent', async () => { - const bus = new InMemoryStateBus(); - const out = await handleResolve(bus, 'approvals', { decision: 'allow' }); + const { iii } = fakeIii(); + const out = await handleResolve(iii, 'approvals', { decision: 'allow' }); expect(out).toEqual({ ok: false, error: 'missing_id' }); }); it('returns bad_decision when decision is invalid', async () => { - const bus = new InMemoryStateBus(); - const out = await handleResolve(bus, 'approvals', { + const { iii } = fakeIii(); + const out = await handleResolve(iii, 'approvals', { session_id: 's1', function_call_id: 'fc-1', decision: 'maybe', @@ -46,66 +74,25 @@ describe('handleResolve (simplified)', () => { expect(out).toEqual({ ok: false, error: 'bad_decision' }); }); - it('returns state_write_failed when bus.set throws', async () => { - const bus = new InMemoryStateBus(); - vi.spyOn(bus, 'set').mockRejectedValue(new Error('boom')); - const out = await handleResolve(bus, 'approvals', { + it('returns state_write_failed when state::set throws', async () => { + const { iii } = fakeIii(); + (iii.trigger as ReturnType).mockRejectedValue(new Error('boom')); + const out = await handleResolve(iii, 'approvals', { session_id: 's1', function_call_id: 'fc-1', decision: 'allow', }); expect(out).toEqual({ ok: false, error: 'state_write_failed' }); }); -}); -describe('handleResolveWithEvents', () => { - it('emits approval_resolved on success but does not publish turn::step_requested', async () => { - const bus = new InMemoryStateBus(); - const triggers: Array<{ function_id: string; payload: unknown }> = []; - const iii = { - trigger: vi.fn( - async ({ function_id, payload }: { function_id: string; payload: unknown }) => { - triggers.push({ function_id, payload }); - return null; - }, - ), - } as unknown as ISdk; - - await handleResolveWithEvents(iii, bus, 'approvals', { + it('never emits to agent::events stream (denial flows through function_execution_end)', async () => { + const { iii, streamSets } = fakeIii(); + await handleResolve(iii, 'approvals', { session_id: 's1', function_call_id: 'fc-1', - decision: 'allow', - reason: 'looks good', - }); - - const fns = triggers.map((t) => t.function_id); - expect(fns).toContain('stream::set'); - expect(fns).not.toContain('iii::durable::publish'); - - const resolved = triggers - .filter((t) => t.function_id === 'stream::set') - .map((t) => (t.payload as Record).data as Record) - .find((d) => d.type === 'approval_resolved'); - expect(resolved).toBeDefined(); - expect(resolved?.function_call_id).toBe('fc-1'); - expect(resolved?.tool_call_id).toBe('fc-1'); - expect(resolved?.decision).toBe('allow'); - expect(resolved?.reason).toBe('looks good'); - }); - - it('does not publish when state write fails', async () => { - const bus = new InMemoryStateBus(); - vi.spyOn(bus, 'set').mockRejectedValue(new Error('boom')); - const iii = { trigger: vi.fn() } as unknown as ISdk; - const out = await handleResolveWithEvents(iii, bus, 'approvals', { - session_id: 's1', - function_call_id: 'fc-1', - decision: 'allow', + decision: 'deny', + reason: 'user cancelled', }); - expect((out as Record).ok).toBe(false); - const fns = (iii.trigger as ReturnType).mock.calls.map( - (c) => (c[0] as { function_id: string }).function_id, - ); - expect(fns).not.toContain('iii::durable::publish'); + expect(streamSets).toHaveLength(0); }); }); diff --git a/harness-node/tests/approval-gate/types.test.ts b/harness-node/tests/approval-gate/types.test.ts index dc976d7412..2d0aeb6ed1 100644 --- a/harness-node/tests/approval-gate/types.test.ts +++ b/harness-node/tests/approval-gate/types.test.ts @@ -1,10 +1,5 @@ import { describe, expect, it } from 'vitest'; -import { - FN_RESOLVE, - blockReplyFor, - extractCall, - pendingKey, -} from '../../src/approval-gate/types.js'; +import { FN_RESOLVE, pendingKey } from '../../src/approval-gate/types.js'; describe('pendingKey', () => { it('joins session and call ids', () => { @@ -21,64 +16,3 @@ describe('approval-gate function constants', () => { expect(FN_RESOLVE).toBe('approval::resolve'); }); }); - -describe('blockReplyFor', () => { - it('marks allow replies with subscriber and approval_gate flags', () => { - expect(blockReplyFor({ kind: 'allow' })).toEqual({ - block: false, - subscriber: 'approval-gate', - approval_gate: true, - }); - }); - - it('marks deny replies with reason, subscriber, and approval_gate flags', () => { - expect(blockReplyFor({ kind: 'deny', reason: 'denied by policy' })).toEqual({ - block: true, - reason: 'approval-gate: denied by policy', - subscriber: 'approval-gate', - approval_gate: true, - }); - }); -}); - -describe('extractCall', () => { - it('reads modern envelope shape', () => { - const envelope = { - event_id: 'evt-1', - reply_stream: 'rs-1', - payload: { - function_call: { id: 'tc-1', function_id: 'write', arguments: { path: '/tmp/x' } }, - approval_required: ['write'], - session_id: 's1', - }, - }; - const call = extractCall(envelope); - expect(call?.session_id).toBe('s1'); - expect(call?.function_id).toBe('write'); - expect(call?.function_call_id).toBe('tc-1'); - expect(call?.approval_required).toContain('write'); - }); - - it('accepts legacy tool_call shape with name field', () => { - const envelope = { - event_id: 'evt-1', - reply_stream: 'rs-1', - payload: { - tool_call: { id: 'tc-1', name: 'write', arguments: {} }, - approval_required: ['write'], - session_id: 's1', - }, - }; - expect(extractCall(envelope)?.function_id).toBe('write'); - }); - - it('returns null when function_call missing', () => { - expect( - extractCall({ - event_id: 'evt-1', - reply_stream: 'rs-1', - payload: { session_id: 's1' }, - }), - ).toBeNull(); - }); -}); diff --git a/harness-node/tests/integration/approval-resume.e2e.test.ts b/harness-node/tests/integration/approval-resume.e2e.test.ts index e47098fe4c..ac8267f500 100644 --- a/harness-node/tests/integration/approval-resume.e2e.test.ts +++ b/harness-node/tests/integration/approval-resume.e2e.test.ts @@ -6,9 +6,9 @@ import { isAbortSignalWrite, } from '../../src/turn-orchestrator/on-abort-signal.js'; -function fakeIii(): { iii: ISdk; publishes: Array<{ topic: string; data: unknown }> } { +function fakeIii(): { iii: ISdk; stepTriggers: Array<{ session_id: string }> } { const stateStore = new Map(); - const publishes: Array<{ topic: string; data: unknown }> = []; + const stepTriggers: Array<{ session_id: string }> = []; const iii = { trigger: vi.fn(async ({ function_id, payload }: { function_id: string; payload: unknown }) => { if (function_id === 'state::set') { @@ -51,8 +51,17 @@ function fakeIii(): { iii: ISdk; publishes: Array<{ topic: string; data: unknown return stateStore.get(`${p.scope}/${p.key}`) ?? null; } + if (function_id === 'turn::step') { + stepTriggers.push(payload as { session_id: string }); + return null; + } + + // abort-signal handler still uses durable publish — capture it here too if (function_id === 'iii::durable::publish') { - publishes.push(payload as { topic: string; data: unknown }); + const p = payload as { topic: string; data: { session_id: string } }; + if (p.topic === 'turn::step_requested') { + stepTriggers.push({ session_id: p.data.session_id }); + } return null; } @@ -60,12 +69,12 @@ function fakeIii(): { iii: ISdk; publishes: Array<{ topic: string; data: unknown }), }; - return { iii: iii as unknown as ISdk, publishes }; + return { iii: iii as unknown as ISdk, stepTriggers }; } describe('approval resume reactive trigger', () => { - it('writing a decision to approvals// automatically publishes turn::step_requested', async () => { - const { iii, publishes } = fakeIii(); + it('writing a decision to approvals// automatically triggers turn::step directly', async () => { + const { iii, stepTriggers } = fakeIii(); await iii.trigger({ function_id: 'state::set', @@ -78,15 +87,12 @@ describe('approval resume reactive trigger', () => { await Promise.resolve(); - expect(publishes).toHaveLength(1); - expect(publishes[0]).toMatchObject({ - topic: 'turn::step_requested', - data: { session_id: 'sess-x' }, - }); + expect(stepTriggers).toHaveLength(1); + expect(stepTriggers[0]).toMatchObject({ session_id: 'sess-x' }); }); - it('writing session//abort_signal=true publishes turn::step_requested', async () => { - const { iii, publishes } = fakeIii(); + it('writing session//abort_signal=true wakes turn::step (via durable publish)', async () => { + const { iii, stepTriggers } = fakeIii(); await iii.trigger({ function_id: 'state::set', @@ -99,15 +105,12 @@ describe('approval resume reactive trigger', () => { await Promise.resolve(); - expect(publishes).toHaveLength(1); - expect(publishes[0]).toMatchObject({ - topic: 'turn::step_requested', - data: { session_id: 'sess-abort' }, - }); + expect(stepTriggers).toHaveLength(1); + expect(stepTriggers[0]).toMatchObject({ session_id: 'sess-abort' }); }); - it('writing session//abort_signal=false does NOT publish (condition rejects clears)', async () => { - const { iii, publishes } = fakeIii(); + it('writing session//abort_signal=false does NOT trigger (condition rejects clears)', async () => { + const { iii, stepTriggers } = fakeIii(); // Seed a previous true so this update sets it to false. await iii.trigger({ @@ -115,7 +118,7 @@ describe('approval resume reactive trigger', () => { payload: { scope: 'agent', key: 'session/sess-clear/abort_signal', value: true }, }); await Promise.resolve(); - publishes.length = 0; // drop the first wake + stepTriggers.length = 0; // drop the first wake await iii.trigger({ function_id: 'state::set', @@ -123,11 +126,11 @@ describe('approval resume reactive trigger', () => { }); await Promise.resolve(); - expect(publishes).toHaveLength(0); + expect(stepTriggers).toHaveLength(0); }); - it('writing an unrelated agent-scope key does NOT publish', async () => { - const { iii, publishes } = fakeIii(); + it('writing an unrelated agent-scope key does NOT trigger', async () => { + const { iii, stepTriggers } = fakeIii(); await iii.trigger({ function_id: 'state::set', @@ -139,6 +142,6 @@ describe('approval resume reactive trigger', () => { }); await Promise.resolve(); - expect(publishes).toHaveLength(0); + expect(stepTriggers).toHaveLength(0); }); }); diff --git a/harness-node/tests/integration/on-record-written.e2e.test.ts b/harness-node/tests/integration/on-record-written.e2e.test.ts new file mode 100644 index 0000000000..17692ebf29 --- /dev/null +++ b/harness-node/tests/integration/on-record-written.e2e.test.ts @@ -0,0 +1,143 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { ISdk } from '../../src/runtime/iii.js'; +import { + STEP_FN_ID, + handleStepableRecordWrite, + isStepableRecordWrite, +} from '../../src/turn-orchestrator/on-record-written.js'; + +function fakeIii(): { iii: ISdk; stepInvocations: Array<{ session_id: string }> } { + const stateStore = new Map(); + const stepInvocations: Array<{ session_id: string }> = []; + const iii = { + trigger: vi.fn(async ({ function_id, payload }: { function_id: string; payload: unknown }) => { + if (function_id === 'state::set') { + const p = payload as { scope: string; key: string; value: unknown }; + const fullKey = `${p.scope}/${p.key}`; + const old_value = stateStore.get(fullKey) ?? null; + stateStore.set(fullKey, p.value); + if (p.scope === 'agent') { + const event = { + event_type: old_value == null ? 'state:created' : 'state:updated', + scope: p.scope, + key: p.key, + old_value, + new_value: p.value, + message_type: 'state', + }; + if (isStepableRecordWrite(event)) { + queueMicrotask(() => { + void handleStepableRecordWrite(iii as unknown as ISdk, event); + }); + } + } + return null; + } + + if (function_id === STEP_FN_ID) { + stepInvocations.push(payload as { session_id: string }); + return null; + } + + return null; + }), + }; + + return { iii: iii as unknown as ISdk, stepInvocations }; +} + +describe('turn-step reactive wake', () => { + it('writing session//turn_state with a stepable state invokes turn::step', async () => { + const { iii, stepInvocations } = fakeIii(); + + await iii.trigger({ + function_id: 'state::set', + payload: { + scope: 'agent', + key: 'session/sess-a/turn_state', + value: { state: 'provisioning' }, + }, + }); + + await Promise.resolve(); + + expect(stepInvocations).toEqual([{ session_id: 'sess-a' }]); + }); + + it('subsequent transitions also wake turn::step', async () => { + const { iii, stepInvocations } = fakeIii(); + + await iii.trigger({ + function_id: 'state::set', + payload: { + scope: 'agent', + key: 'session/sess-b/turn_state', + value: { state: 'provisioning' }, + }, + }); + await Promise.resolve(); + + await iii.trigger({ + function_id: 'state::set', + payload: { + scope: 'agent', + key: 'session/sess-b/turn_state', + value: { state: 'awaiting_assistant' }, + }, + }); + await Promise.resolve(); + + expect(stepInvocations).toEqual([ + { session_id: 'sess-b' }, + { session_id: 'sess-b' }, + ]); + }); + + it('parking in function_awaiting_approval does NOT wake', async () => { + const { iii, stepInvocations } = fakeIii(); + + await iii.trigger({ + function_id: 'state::set', + payload: { + scope: 'agent', + key: 'session/sess-c/turn_state', + value: { state: 'function_awaiting_approval' }, + }, + }); + await Promise.resolve(); + + expect(stepInvocations).toEqual([]); + }); + + it('terminal stopped state does NOT wake', async () => { + const { iii, stepInvocations } = fakeIii(); + + await iii.trigger({ + function_id: 'state::set', + payload: { + scope: 'agent', + key: 'session/sess-d/turn_state', + value: { state: 'stopped' }, + }, + }); + await Promise.resolve(); + + expect(stepInvocations).toEqual([]); + }); + + it('non-turn_state agent keys do NOT wake (no leakage from abort_signal etc.)', async () => { + const { iii, stepInvocations } = fakeIii(); + + await iii.trigger({ + function_id: 'state::set', + payload: { + scope: 'agent', + key: 'session/sess-e/abort_signal', + value: true, + }, + }); + await Promise.resolve(); + + expect(stepInvocations).toEqual([]); + }); +}); diff --git a/harness-node/tests/turn-orchestrator/agent-call.test.ts b/harness-node/tests/turn-orchestrator/agent-call.test.ts index 038f8ac549..05c7043308 100644 --- a/harness-node/tests/turn-orchestrator/agent-call.test.ts +++ b/harness-node/tests/turn-orchestrator/agent-call.test.ts @@ -80,13 +80,13 @@ describe('isErrorResult', () => { describe('dispatchWithHook returns DispatchResult', () => { it('returns kind:pending when consultBefore returns pending', async () => { - vi.spyOn(hookModule, 'consultBefore').mockResolvedValue({ kind: 'pending', merged: {} }); + vi.spyOn(hookModule, 'consultBefore').mockResolvedValue({ kind: 'pending' }); const iii = { trigger: vi.fn() } as unknown as ISdk; const out = await dispatchWithHook( iii, { id: 'fc-1', function_id: 'shell::run', arguments: { command: 'ls' } }, - [], 's1', + 'policy::check_permissions', ); expect(out.kind).toBe('pending'); }); @@ -106,8 +106,8 @@ describe('dispatchWithHook returns DispatchResult', () => { const out = await dispatchWithHook( iii, { id: 'fc-1', function_id: 'shell::run', arguments: {} }, - [], 's1', + 'policy::check_permissions', ); expect(out.kind).toBe('deny'); if (out.kind === 'deny') { @@ -123,9 +123,28 @@ describe('dispatchWithHook returns DispatchResult', () => { const out = await dispatchWithHook( iii, { id: 'fc-1', function_id: 'shell::run', arguments: {} }, - [], 's1', + 'policy::check_permissions', ); expect(out.kind).toBe('result'); }); + + it('passes policy_function_id through to consultBefore', async () => { + const spy = vi.spyOn(hookModule, 'consultBefore').mockResolvedValue({ kind: 'allow' }); + const iii = { + trigger: vi.fn().mockResolvedValue({ ok: true }), + } as unknown as ISdk; + await dispatchWithHook( + iii, + { id: 'fc-1', function_id: 'shell::run', arguments: {} }, + 's1', + 'custom::policy', + ); + expect(spy).toHaveBeenCalledWith( + iii, + expect.objectContaining({ function_id: 'shell::run' }), + 's1', + 'custom::policy', + ); + }); }); diff --git a/harness-node/tests/turn-orchestrator/config.test.ts b/harness-node/tests/turn-orchestrator/config.test.ts new file mode 100644 index 0000000000..3a2f8de90b --- /dev/null +++ b/harness-node/tests/turn-orchestrator/config.test.ts @@ -0,0 +1,14 @@ +import { describe, expect, it } from 'vitest'; +import { loadOrchestratorConfig } from '../../src/turn-orchestrator/config.js'; + +describe('loadOrchestratorConfig', () => { + it('exposes policy_function_id with a sane default', () => { + const cfg = loadOrchestratorConfig({}); + expect(cfg.policy_function_id).toBe('policy::check_permissions'); + }); + + it('lets policy_function_id override the default', () => { + const cfg = loadOrchestratorConfig({ policy_function_id: 'custom::policy' }); + expect(cfg.policy_function_id).toBe('custom::policy'); + }); +}); diff --git a/harness-node/tests/turn-orchestrator/functions.test.ts b/harness-node/tests/turn-orchestrator/functions.test.ts index e674bec9b3..e64817a54b 100644 --- a/harness-node/tests/turn-orchestrator/functions.test.ts +++ b/harness-node/tests/turn-orchestrator/functions.test.ts @@ -1,12 +1,19 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import type { ISdk } from '../../src/runtime/iii.js'; import * as agentCallModule from '../../src/turn-orchestrator/agent-call.js'; +import type { TurnOrchestratorConfig } from '../../src/turn-orchestrator/config.js'; import * as hookModule from '../../src/turn-orchestrator/hook.js'; import * as persistence from '../../src/turn-orchestrator/persistence.js'; import type { TurnStateRecord } from '../../src/turn-orchestrator/state.js'; import { newRecord } from '../../src/turn-orchestrator/state.js'; import { handleExecute } from '../../src/turn-orchestrator/states/functions.js'; +const cfg: TurnOrchestratorConfig = { + policy_function_id: 'policy::check_permissions', + sync_default_timeout_ms: 120_000, + system_default_skills: [], +}; + afterEach(() => { vi.restoreAllMocks(); }); @@ -32,9 +39,7 @@ describe('handleExecute new flow', () => { ]); vi.spyOn(persistence, 'loadExecutedCalls').mockResolvedValue([]); vi.spyOn(persistence, 'saveExecutedCalls').mockResolvedValue(undefined); - vi.spyOn(persistence, 'loadRunRequest').mockResolvedValue({ approval_required: [] }); - - await handleExecute(iii, rec); + await handleExecute(iii, cfg, rec); expect(rec.state).toBe('function_awaiting_approval'); expect(rec.awaiting_approval).toHaveLength(1); @@ -60,11 +65,9 @@ describe('handleExecute new flow', () => { ]); vi.spyOn(persistence, 'loadExecutedCalls').mockResolvedValue([]); vi.spyOn(persistence, 'saveExecutedCalls').mockResolvedValue(undefined); - vi.spyOn(persistence, 'loadRunRequest').mockResolvedValue({ approval_required: [] }); - const consultBeforeSpy = vi.spyOn(hookModule, 'consultBefore'); - await handleExecute(iii, rec); + await handleExecute(iii, cfg, rec); expect(consultBeforeSpy).not.toHaveBeenCalled(); const triggerCalls = triggerSpy.mock.calls.map( @@ -73,6 +76,46 @@ describe('handleExecute new flow', () => { expect(triggerCalls).toContain('shell::run'); }); + it('synthesizes an error result when a pre_approved trigger rejects (does not throw out of handleExecute)', async () => { + const triggerSpy = vi.fn(async (req: { function_id: string }) => { + if (req.function_id === 'shell::fs::write') { + throw new Error('handler error: {"code":"S210","message":"bad write payload"}'); + } + return null; + }); + const iii = { trigger: triggerSpy } as unknown as ISdk; + const rec: TurnStateRecord = newRecord('s1'); + rec.state = 'function_execute'; + + vi.spyOn(persistence, 'loadPreparedCalls').mockResolvedValue([ + { + function_call: { + id: 'fc-1', + function_id: 'shell::fs::write', + arguments: { content: 'Tue May 19 08:17:10 -03 2026\n' }, + }, + blocked: null, + pre_approved: true, + }, + ]); + vi.spyOn(persistence, 'loadExecutedCalls').mockResolvedValue([]); + const saveSpy = vi.spyOn(persistence, 'saveExecutedCalls').mockResolvedValue(undefined); + + await expect(handleExecute(iii, cfg, rec)).resolves.toBeUndefined(); + + expect(rec.state).toBe('function_finalize'); + expect(saveSpy).toHaveBeenCalled(); + const lastSave = saveSpy.mock.calls.at(-1)?.[2] as Array<{ + is_error: boolean; + result: { details: unknown }; + }>; + expect(lastSave?.[0]?.is_error).toBe(true); + const details = lastSave?.[0]?.result.details as Record; + expect(details?.error).toBe('trigger_failed'); + expect(details?.function).toBe('shell::fs::write'); + expect(String(details?.message)).toContain('S210'); + }); + it('emits denial result without dispatching when blocked is set', async () => { const triggerSpy = vi.fn().mockResolvedValue(null); const iii = { trigger: triggerSpy } as unknown as ISdk; @@ -93,9 +136,7 @@ describe('handleExecute new flow', () => { ]); vi.spyOn(persistence, 'loadExecutedCalls').mockResolvedValue([]); vi.spyOn(persistence, 'saveExecutedCalls').mockResolvedValue(undefined); - vi.spyOn(persistence, 'loadRunRequest').mockResolvedValue({ approval_required: [] }); - - await handleExecute(iii, rec); + await handleExecute(iii, cfg, rec); const shellCalls = triggerSpy.mock.calls.filter( (call) => (call[0] as { function_id: string }).function_id === 'shell::run', diff --git a/harness-node/tests/turn-orchestrator/get-state.test.ts b/harness-node/tests/turn-orchestrator/get-state.test.ts new file mode 100644 index 0000000000..7d6cacfb31 --- /dev/null +++ b/harness-node/tests/turn-orchestrator/get-state.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { ISdk } from '../../src/runtime/iii.js'; +import { FUNCTION_ID, execute } from '../../src/turn-orchestrator/get-state.js'; +import { newRecord } from '../../src/turn-orchestrator/state.js'; + +describe('turn::get_state', () => { + it('exposes the canonical function id', () => { + expect(FUNCTION_ID).toBe('turn::get_state'); + }); + + it('returns the turn_state record for a known session via persistence.loadRecord', async () => { + const rec = newRecord('sess-abc'); + rec.state = 'function_awaiting_approval'; + const iii = { + trigger: vi.fn(async (req: { function_id: string; payload: unknown }) => { + if ( + req.function_id === 'state::get' && + (req.payload as Record).key === 'session/sess-abc/turn_state' + ) { + return rec; + } + return null; + }), + } as unknown as ISdk; + + const out = await execute(iii, { session_id: 'sess-abc' }); + expect(out).toEqual(rec); + }); + + it('returns null when no record exists for the session', async () => { + const iii = { + trigger: vi.fn(async () => null), + } as unknown as ISdk; + const out = await execute(iii, { session_id: 'unknown' }); + expect(out).toBeNull(); + }); + + it('throws on missing session_id', async () => { + const iii = { trigger: vi.fn() } as unknown as ISdk; + await expect(execute(iii, {})).rejects.toThrow(/session_id/); + }); +}); diff --git a/harness-node/tests/turn-orchestrator/hook.test.ts b/harness-node/tests/turn-orchestrator/hook.test.ts new file mode 100644 index 0000000000..d1c667b9b2 --- /dev/null +++ b/harness-node/tests/turn-orchestrator/hook.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { ISdk } from '../../src/runtime/iii.js'; +import { consultBefore } from '../../src/turn-orchestrator/hook.js'; + +function fakeIii(triggerImpl: (req: { function_id: string; payload: unknown }) => unknown) { + return { + trigger: vi.fn(async (req: { function_id: string; payload: unknown }) => triggerImpl(req)), + } as unknown as ISdk; +} + +describe('consultBefore (direct policy call)', () => { + const fc = { id: 'fc-1', function_id: 'shell::fs::write', arguments: { path: '/tmp/x' } }; + + it('returns allow when policy decides allow', async () => { + const iii = fakeIii(({ function_id }) => { + expect(function_id).toBe('policy::check_permissions'); + return { decision: 'allow', rule_id: 'allow-tmp' }; + }); + const outcome = await consultBefore(iii, fc, 'sess-a', 'policy::check_permissions'); + expect(outcome.kind).toBe('allow'); + }); + + it('returns deny with a permissions envelope when policy decides deny', async () => { + const iii = fakeIii(() => ({ + decision: 'deny', + rule_id: 'deny-rm-rf', + matched_constraint: { field: 'path', operator: 'matches', value: '^/$' }, + })); + const outcome = await consultBefore(iii, fc, 'sess-a', 'policy::check_permissions'); + expect(outcome.kind).toBe('deny'); + if (outcome.kind !== 'deny') return; + expect(outcome.denial.denied_by).toBe('permissions'); + expect(outcome.denial.rule_id).toBe('deny-rm-rf'); + }); + + it('returns pending when policy says needs_approval', async () => { + const iii = fakeIii(() => ({ decision: 'needs_approval' })); + const outcome = await consultBefore(iii, fc, 'sess-a', 'policy::check_permissions'); + expect(outcome.kind).toBe('pending'); + }); + + it('fails closed (deny gate_unavailable) when policy is unreachable', async () => { + const iii = fakeIii(() => { + throw new Error('policy worker down'); + }); + const outcome = await consultBefore(iii, fc, 'sess-a', 'policy::check_permissions'); + expect(outcome.kind).toBe('deny'); + if (outcome.kind !== 'deny') return; + expect(outcome.denial.denied_by).toBe('gate_unavailable'); + }); + + it('does NOT call hook-fanout::publish_collect', async () => { + const trigger = vi.fn(async () => ({ decision: 'allow', rule_id: 'r' })); + const iii = { trigger } as unknown as ISdk; + await consultBefore(iii, fc, 'sess-a', 'policy::check_permissions'); + expect(trigger).toHaveBeenCalledTimes(1); + expect(trigger.mock.calls[0][0].function_id).toBe('policy::check_permissions'); + }); +}); diff --git a/harness-node/tests/turn-orchestrator/on-record-written.test.ts b/harness-node/tests/turn-orchestrator/on-record-written.test.ts new file mode 100644 index 0000000000..49c4f6bbf8 --- /dev/null +++ b/harness-node/tests/turn-orchestrator/on-record-written.test.ts @@ -0,0 +1,221 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { ISdk } from '../../src/runtime/iii.js'; +import { + STEP_FN_ID, + handleStepableRecordWrite, + isStepableRecordWrite, +} from '../../src/turn-orchestrator/on-record-written.js'; + +describe('isStepableRecordWrite condition', () => { + it('matches turn_state writes with a non-terminal, non-awaiting state', () => { + expect( + isStepableRecordWrite({ + event_type: 'state:created', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: null, + new_value: { state: 'provisioning' }, + message_type: 'state', + }), + ).toBe(true); + + expect( + isStepableRecordWrite({ + event_type: 'state:updated', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: { state: 'provisioning' }, + new_value: { state: 'awaiting_assistant' }, + message_type: 'state', + }), + ).toBe(true); + }); + + it('rejects terminal state (stopped)', () => { + expect( + isStepableRecordWrite({ + event_type: 'state:updated', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: { state: 'tearing_down' }, + new_value: { state: 'stopped' }, + message_type: 'state', + }), + ).toBe(false); + }); + + it('rejects function_awaiting_approval (orchestrator parks here)', () => { + expect( + isStepableRecordWrite({ + event_type: 'state:updated', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: { state: 'function_prepare' }, + new_value: { state: 'function_awaiting_approval' }, + message_type: 'state', + }), + ).toBe(false); + }); + + it('rejects state:deleted', () => { + expect( + isStepableRecordWrite({ + event_type: 'state:deleted', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: { state: 'provisioning' }, + new_value: null, + message_type: 'state', + }), + ).toBe(false); + }); + + it('rejects non-turn_state keys in the agent scope', () => { + expect( + isStepableRecordWrite({ + event_type: 'state:updated', + scope: 'agent', + key: 'session/sess-abc/abort_signal', + old_value: null, + new_value: true, + message_type: 'state', + }), + ).toBe(false); + }); + + it('rejects same-state writes (old_value.state === new_value.state)', () => { + expect( + isStepableRecordWrite({ + event_type: 'state:updated', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: { state: 'function_prepare' }, + new_value: { state: 'function_prepare' }, + message_type: 'state', + }), + ).toBe(false); + + expect( + isStepableRecordWrite({ + event_type: 'state:updated', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: { state: 'function_prepare' }, + new_value: { state: 'function_execute' }, + message_type: 'state', + }), + ).toBe(true); + }); + + it('rejects writes whose new_value lacks a string state', () => { + expect( + isStepableRecordWrite({ + event_type: 'state:updated', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: null, + new_value: { not_state: 'provisioning' }, + message_type: 'state', + }), + ).toBe(false); + + expect( + isStepableRecordWrite({ + event_type: 'state:updated', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: null, + new_value: null, + message_type: 'state', + }), + ).toBe(false); + }); +}); + +describe('handleStepableRecordWrite', () => { + it('extracts session_id and invokes turn::step directly', async () => { + const triggers: Array<{ function_id: string; payload: unknown }> = []; + const iii = { + trigger: vi.fn(async (req: { function_id: string; payload: unknown }) => { + triggers.push(req); + return null; + }), + } as unknown as ISdk; + + await handleStepableRecordWrite(iii, { + event_type: 'state:created', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: null, + new_value: { state: 'provisioning' }, + message_type: 'state', + }); + + expect(triggers).toHaveLength(1); + expect(triggers[0]?.function_id).toBe(STEP_FN_ID); + expect(triggers[0]?.payload).toEqual({ session_id: 'sess-abc' }); + }); + + it('no-ops when key does not match the turn_state pattern', async () => { + const iii = { trigger: vi.fn() } as unknown as ISdk; + await handleStepableRecordWrite(iii, { + event_type: 'state:updated', + scope: 'agent', + key: 'session/sess-abc/abort_signal', + old_value: null, + new_value: true, + message_type: 'state', + }); + expect(iii.trigger).not.toHaveBeenCalled(); + }); + + it('falls back to durable publish when the direct turn::step invoke fails', async () => { + const triggers: Array<{ function_id: string; payload: unknown }> = []; + const iii = { + trigger: vi.fn(async (req: { function_id: string; payload: unknown }) => { + triggers.push(req); + // Fail the direct turn::step invoke; let the durable publish succeed. + if (req.function_id === STEP_FN_ID) { + throw new Error('engine down'); + } + return null; + }), + } as unknown as ISdk; + + await handleStepableRecordWrite(iii, { + event_type: 'state:created', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: null, + new_value: { state: 'provisioning' }, + message_type: 'state', + }); + + expect(triggers).toHaveLength(2); + expect(triggers[0]?.function_id).toBe(STEP_FN_ID); + expect(triggers[1]?.function_id).toBe('iii::durable::publish'); + expect(triggers[1]?.payload).toEqual({ + topic: 'turn::step_requested', + data: { session_id: 'sess-abc' }, + }); + }); + + it('swallows when BOTH the direct invoke and durable publish fallback fail', async () => { + const iii = { + trigger: vi.fn(async () => { + throw new Error('engine down'); + }), + } as unknown as ISdk; + + await expect( + handleStepableRecordWrite(iii, { + event_type: 'state:created', + scope: 'agent', + key: 'session/sess-abc/turn_state', + old_value: null, + new_value: { state: 'provisioning' }, + message_type: 'state', + }), + ).resolves.toBeUndefined(); + }); +}); diff --git a/harness-node/tests/turn-orchestrator/on-turn-state-changed.test.ts b/harness-node/tests/turn-orchestrator/on-turn-state-changed.test.ts new file mode 100644 index 0000000000..881fd5162e --- /dev/null +++ b/harness-node/tests/turn-orchestrator/on-turn-state-changed.test.ts @@ -0,0 +1,129 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { ISdk } from '../../src/runtime/iii.js'; +import { + CONDITION_FN_ID, + handleTurnStateWrite, + isTurnStateWrite, +} from '../../src/turn-orchestrator/on-turn-state-changed.js'; + +function fakeIii(): { iii: ISdk; emits: Array<{ session_id: string; event: unknown }> } { + const emits: Array<{ session_id: string; event: unknown }> = []; + const iii = { + trigger: vi.fn(async ({ function_id, payload }: { function_id: string; payload: unknown }) => { + if (function_id === 'stream::set') { + const p = payload as { group_id: string; data: unknown }; + emits.push({ session_id: p.group_id, event: p.data }); + return null; + } + return null; + }), + } as unknown as ISdk; + return { iii, emits }; +} + +describe('CONDITION_FN_ID', () => { + it('is the stable string the trigger config will reference', () => { + expect(CONDITION_FN_ID).toBe('turn::is_turn_state_write'); + }); +}); + +describe('isTurnStateWrite', () => { + it('returns true for state:created on session//turn_state', () => { + expect( + isTurnStateWrite({ + event_type: 'state:created', + key: 'session/sess-a/turn_state', + new_value: { state: 'provisioning' }, + }), + ).toBe(true); + }); + + it('returns true for state:updated on session//turn_state', () => { + expect( + isTurnStateWrite({ + event_type: 'state:updated', + key: 'session/sess-a/turn_state', + new_value: { state: 'function_awaiting_approval' }, + old_value: { state: 'function_execute' }, + }), + ).toBe(true); + }); + + it('returns false for non-turn_state agent keys', () => { + expect( + isTurnStateWrite({ + event_type: 'state:created', + key: 'session/sess-a/abort_signal', + new_value: true, + }), + ).toBe(false); + }); + + it('returns false for state:deleted', () => { + expect( + isTurnStateWrite({ + event_type: 'state:deleted', + key: 'session/sess-a/turn_state', + }), + ).toBe(false); + }); +}); + +describe('handleTurnStateWrite', () => { + it('emits turn_state_changed on agent::events with group_id = session_id', async () => { + const { iii, emits } = fakeIii(); + await handleTurnStateWrite(iii, { + event_type: 'state:updated', + key: 'session/sess-a/turn_state', + new_value: { state: 'function_awaiting_approval', awaiting_approval: [] }, + old_value: { state: 'function_execute', awaiting_approval: null }, + }); + expect(emits).toHaveLength(1); + expect(emits[0]?.session_id).toBe('sess-a'); + expect(emits[0]?.event).toMatchObject({ + type: 'turn_state_changed', + event_type: 'state:updated', + new_value: { state: 'function_awaiting_approval' }, + old_value: { state: 'function_execute' }, + }); + }); + + it('is a no-op when the event does not match the condition', async () => { + const { iii, emits } = fakeIii(); + await handleTurnStateWrite(iii, { + event_type: 'state:created', + key: 'session/sess-a/abort_signal', + new_value: true, + }); + expect(emits).toEqual([]); + }); + + it('swallows emit failures (logs only, never rethrows)', async () => { + const iii = { + trigger: vi.fn(async () => { + throw new Error('stream::set down'); + }), + } as unknown as ISdk; + // Should NOT throw. + await expect( + handleTurnStateWrite(iii, { + event_type: 'state:created', + key: 'session/sess-a/turn_state', + new_value: { state: 'provisioning' }, + }), + ).resolves.toBeUndefined(); + }); + + it('omits old_value from the emitted event when state:created', async () => { + const { iii, emits } = fakeIii(); + await handleTurnStateWrite(iii, { + event_type: 'state:created', + key: 'session/sess-a/turn_state', + new_value: { state: 'provisioning' }, + }); + expect(emits).toHaveLength(1); + const event = emits[0]?.event as Record; + expect(event.type).toBe('turn_state_changed'); + expect('old_value' in event).toBe(false); + }); +}); diff --git a/harness-node/tests/turn-orchestrator/run-start.test.ts b/harness-node/tests/turn-orchestrator/run-start.test.ts index a6f6fdda0a..c429cbc0f6 100644 --- a/harness-node/tests/turn-orchestrator/run-start.test.ts +++ b/harness-node/tests/turn-orchestrator/run-start.test.ts @@ -23,7 +23,7 @@ describe('run-start constants', () => { }); describe('execute', () => { - it('saves initial session state and publishes the first step', async () => { + it('saves initial session state to wake the reactive step trigger', async () => { const { iii, calls } = fakeIii(); await execute(iii, { @@ -33,12 +33,21 @@ describe('execute', () => { messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }], timestamp: 1 }], }); - const stateSets = calls.filter((c) => c.function_id === 'state::set'); - expect(stateSets.length).toBeGreaterThan(0); + // The reactive wake is now state-driven: the turn_state write at + // state='provisioning' is what the on-record-written trigger picks up. + // run-start no longer self-publishes turn::step_requested. + const turnStateSet = calls.find( + (c) => + c.function_id === 'state::set' && + (c.payload as { scope?: string; key?: string }).scope === 'agent' && + (c.payload as { scope?: string; key?: string }).key === 'session/s1/turn_state', + ); + expect(turnStateSet).toBeDefined(); + expect((turnStateSet?.payload as { value: { state: string } }).value.state).toBe( + 'provisioning', + ); + const publish = calls.find((c) => c.function_id === 'iii::durable::publish'); - expect(publish?.payload).toEqual({ - topic: 'turn::step_requested', - data: { session_id: 's1' }, - }); + expect(publish).toBeUndefined(); }); });