feat(harness): agent trigger subscriptions — be notified instead of polling

console/web/src/components/chat/Message.tsx

@@ -33,7 +33,11 @@ export function Message({
 }: MessageProps) {
   switch (message.role) {
     case 'user':
-      return <UserMessage message={message} />
+      return message.notification ? (
+        <NotificationMessage message={message} />
+      ) : (
+        <UserMessage message={message} />
+      )
     case 'assistant':
       return <AssistantMessage message={message} />
     case 'thought':
@@ -124,6 +128,15 @@ function CompactionMarker({ message }: { message: SystemMessageType }) {
   )
 }
 
+function NotificationMessage({ message }: { message: UserMessageType }) {
+  return (
+    <article className="border-l-2 border-l-rule pl-3 py-1 font-mono text-[12px] text-ink-faint flex items-start gap-2">
+      <span aria-hidden="true">🔔</span>
+      <span className="break-words">{message.content}</span>
+    </article>
+  )
+}
+
 function UserMessage({ message }: { message: UserMessageType }) {
   return (
     <article className="flex flex-col items-end gap-2">

console/web/src/hooks/use-conversations.ts

@@ -381,6 +381,7 @@ export function useConversations(
                 entry_id: event.entry_id,
                 message: event.message,
                 custom: event.custom,
+                origin: event.origin,
               },
               { sessionId },
             ),
@@ -396,7 +397,11 @@ export function useConversations(
             ...c,
             messages: applyEntryUpsert(
               c.messages,
-              { entry_id: event.entry_id, message: event.message },
+              {
+                entry_id: event.entry_id,
+                message: event.message,
+                origin: event.origin,
+              },
               { sessionId, streaming: c.status === 'working' },
             ),
             updatedAt: event.timestamp,

console/web/src/lib/sessions/entry-mapper.test.ts

@@ -9,9 +9,14 @@ import {
 } from './entry-mapper'
 import type { AgentMessage, TranscriptItem } from './types'
 
-function userItem(entryId: string, text: string): TranscriptItem {
+function userItem(
+  entryId: string,
+  text: string,
+  origin?: TranscriptItem['origin'],
+): TranscriptItem {
   return {
     entry_id: entryId,
+    ...(origin ? { origin } : {}),
     message: { role: 'user', content: [{ type: 'text', text }], timestamp: 1 },
   }
 }
@@ -63,6 +68,18 @@ describe('entrySegments', () => {
     })
   })
 
+  it('marks only trusted notification user entries', () => {
+    expect(
+      entrySegments(userItem('e-1', 'normal', { notification: false }))[0],
+    ).not.toHaveProperty('notification')
+    expect(
+      entrySegments(userItem('e-2', 'wake', { notification: true }))[0],
+    ).toMatchObject({ notification: true })
+    expect(entrySegments(userItem('e_notify_sub_1', 'wake'))[0]).toMatchObject({
+      notification: true,
+    })
+  })
+
   it('splits an assistant entry into thought/text/function-call segments by block', () => {
     const segments = entrySegments(
       assistantItem('e-a', [
@@ -82,7 +99,6 @@ describe('entrySegments', () => {
       ['e-a:1', 'assistant'],
       ['e-a:2', 'function-call'],
     ])
-    // agent_trigger is unwrapped to the real target for display.
     expect(segments[2]).toMatchObject({
       functionId: 'shell::run',
       input: { command: 'ls' },
@@ -128,7 +144,6 @@ describe('applyEntryUpsert', () => {
     )
     expect(next).toHaveLength(1)
     expect(next[0]).toMatchObject({ id: 'msg-1-user-0', content: 'hello' })
-    // Attachments are client-only; preserved across the replacement.
     expect((next[0] as { attachments?: unknown[] }).attachments).toHaveLength(1)
   })
 
@@ -199,8 +214,6 @@ describe('applyEntryUpsert', () => {
   })
 
   it('absorbs a locally-created fcall row (pending approval) into the entry segment', () => {
-    // The live approval flow appended a local row before the assistant
-    // snapshot arrived.
     const local: Message = {
       id: 'local-1',
       role: 'function-call',

console/web/src/lib/sessions/entry-mapper.ts

@@ -106,11 +106,15 @@ export function entrySegments(
 
   switch (message.role) {
     case 'user': {
+      const notif = (item.origin as { notification?: unknown } | undefined)
+        ?.notification
+      const isNotif = notif === true || item.entry_id.startsWith('e_notify_')
       const msg: UserMessage = {
         id: item.entry_id,
         role: 'user',
         content: textOf(message.content),
         createdAt: message.timestamp,
+        ...(isNotif ? { notification: true } : {}),
       }
       return [msg]
     }

console/web/src/lib/sessions/types.ts

@@ -71,6 +71,7 @@ export type TranscriptItem = {
   entry_id: string
   message?: AgentMessage
   custom?: { custom_type: string; data: unknown }
+  origin?: Record<string, unknown>
 }
 
 export const SESSION_TRIGGER_TYPES = [

console/web/src/types/chat.ts

@@ -59,6 +59,7 @@ export interface UserMessage extends BaseMessage {
   role: 'user'
   content: string
   attachments?: Attachment[]
+  notification?: boolean
 }
 
 export interface AssistantMessage extends BaseMessage {

harness/prompts/anthropic.txt

@@ -21,7 +21,13 @@ Consequences worth internalising:
   registration. Restarting a worker is invisible to callers as long as it re-registers the
   same function ids; two workers registering the same id load-balance automatically.
 - Triggers are the engine's push channel. NEVER poll (a timer re-reading a queue, file, or
-  table) when a trigger type fits — bind a trigger instead.
+  table) when a trigger type fits — bind a trigger instead. To be notified yourself, call
+  `engine::register_trigger { trigger_type, config }` (cron, state, stream, or another worker's
+  custom trigger type; optional `once`, `label`). When it fires a notification message arrives in
+  this session — non-blocking, so keep working; the event will reach you. For an ad-hoc signal,
+  subscribe to `state` on a key and have the signaller call `state::set` on it (the engine fans
+  the trigger out to every subscriber). It returns a subscription_id; tear it down with
+  `engine::unregister_trigger { id: <subscription_id> }`.
 
 # Discovery
 

harness/prompts/gpt.txt

@@ -14,7 +14,12 @@ that invoke them). Every call routes worker → engine → worker — there is n
 worker-to-worker traffic, and the function id is the only contract between two workers. A
 function is callable the instant its worker connects; workers registering the same id
 load-balance; restarts are invisible to callers. Triggers are the engine's push channel —
-never poll when a trigger type fits.
+never poll when a trigger type fits. To be notified yourself instead of polling, call
+`engine::register_trigger { trigger_type, config }` (cron, state, stream, or another worker's
+trigger type; optional `once`, `label`); it delivers a notification message into this session
+when it fires (non-blocking — keep working) and returns a subscription_id. For an ad-hoc signal,
+subscribe to `state` on a key and have the signaller call `state::set` on it. Tear it down with
+`engine::unregister_trigger { id: <subscription_id> }`.
 
 ## Discovery
 

harness/prompts/kimi.txt

@@ -19,7 +19,12 @@ that invoke them). Every call routes worker → engine → worker. There is no d
 worker-to-worker traffic. The function id is the ONLY contract between two workers. Functions
 are callable the moment their worker connects; workers registering the same id load-balance;
 restarts are invisible. Triggers are the engine's push channel — you MUST NOT poll when a
-trigger type fits.
+trigger type fits. To be notified yourself instead of polling, call
+`engine::register_trigger { trigger_type, config }` (cron, state, stream, or another worker's
+trigger type; optional `once`, `label`); it delivers a notification message into this session
+when it fires (non-blocking — keep working) and returns a subscription_id. For an ad-hoc signal,
+subscribe to `state` on a key and have the signaller call `state::set` on it. Tear it down with
+`engine::unregister_trigger { id: <subscription_id> }`.
 
 # Discovery
 

harness/src/configuration.rs

@@ -28,6 +28,7 @@ use tokio::sync::RwLock;
 
 use crate::config::WorkerConfig;
 use crate::functions::sweep_pending::SWEEP_PENDING_ID;
+use crate::subscriptions::ON_SESSION_DELETED_ID;
 
 /// Hot-swappable config snapshot shared with every handler.
 pub type ConfigCell = Arc<RwLock<Arc<WorkerConfig>>>;
@@ -173,9 +174,19 @@ pub async fn apply_config(cell: &ConfigCell, cfg: WorkerConfig) {
     *cell.write().await = Arc::new(cfg);
 }
 
-/// Live handle for the one hot-reloadable trigger binding — the cron sweep.
+/// Live trigger handles retained for the worker lifetime.
 pub struct TriggerHandles {
     pub sweep: std::sync::Mutex<Option<Trigger>>,
+    _session_deleted: Option<Trigger>,
+}
+
+impl TriggerHandles {
+    pub fn new(sweep: Option<Trigger>, session_deleted: Option<Trigger>) -> Self {
+        Self {
+            sweep: std::sync::Mutex::new(sweep),
+            _session_deleted: session_deleted,
+        }
+    }
 }
 
 /// Best-effort binding: the cron trigger type always exists (engine built-in),
@@ -209,6 +220,10 @@ pub fn bind_sweep(iii: &IIIClient, cfg: &WorkerConfig) -> Option<Trigger> {
     )
 }
 
+pub fn bind_session_deleted(iii: &IIIClient) -> Option<Trigger> {
+    bind(iii, "session::deleted", ON_SESSION_DELETED_ID, json!({}))
+}
+
 /// Store the freshly-registered handle, then unregister the old one
 /// (register-new-then-unregister-old: a fail-safe overlap).
 fn rebind_slot(slot: &std::sync::Mutex<Option<Trigger>>, new: Option<Trigger>) {

harness/src/deferred.rs

@@ -90,16 +90,25 @@ pub async fn resolve(
             let arguments = find_call_arguments(deps, &record, &req.function_call_id)
                 .await
                 .unwrap_or(Value::Null);
-
             if let Some(cp) = record.calls.get_mut(&req.function_call_id) {
                 cp.state = CallState::Triggered;
             }
             crate::state::put_turn(&deps.iii, &record, cfg.session_timeout_ms).await?;
 
             let policy = crate::policy::CompiledPolicy::from(record.options.functions.as_ref());
             let engine = deps.engine().await;
-            let raw =
-                crate::trigger::invoke_target(&engine, &policy, &function_id, &arguments).await;
+            // Held calls resume outside the inline loop, so use the same
+            // invocation chokepoint to keep subscription session injection/owner
+            // checks.
+            let raw = crate::functions::subscribe::invoke(
+                deps,
+                &engine,
+                &policy,
+                &function_id,
+                &arguments,
+                &record.session_id,
+            )
+            .await;
             let (data, annotations) = deps
                 .hooks
                 .run_post_trigger(

harness/src/deps.rs

@@ -15,6 +15,7 @@ use crate::discovery::FunctionsCell;
 use crate::events::TurnEvents;
 use crate::hooks::HookRegistry;
 use crate::locks::SessionLocks;
+use crate::subscriptions::SubscriptionRegistry;
 
 #[derive(Clone)]
 pub struct Deps {
@@ -24,6 +25,7 @@ pub struct Deps {
     pub events: TurnEvents,
     pub hooks: HookRegistry,
     pub locks: SessionLocks,
+    pub subscriptions: Arc<SubscriptionRegistry>,
 }
 
 impl Deps {
@@ -41,6 +43,7 @@ impl Deps {
             events,
             hooks,
             locks: SessionLocks::new(),
+            subscriptions: Arc::new(SubscriptionRegistry::new()),
         }
     }
 

harness/src/functions/function_trigger.rs

@@ -130,7 +130,17 @@ pub async fn handle(
         }
     }
 
-    let raw = trigger::invoke_target(&engine, &policy, &req.call.function_id, &arguments).await;
+    // Single invocation chokepoint: subscription control calls are intercepted
+    // (trusted session injected); everything else invokes the target.
+    let raw = crate::functions::subscribe::invoke(
+        deps,
+        &engine,
+        &policy,
+        &req.call.function_id,
+        &arguments,
+        &req.session_id,
+    )
+    .await;
     let data = if let Some(rec) = &record {
         deps.hooks
             .run_post_trigger(rec, rec.step, &req.call.id, &req.call.function_id, raw)

harness/src/functions/mod.rs

@@ -5,10 +5,12 @@
 
 pub mod function_resolve;
 pub mod function_trigger;
+pub mod on_session_deleted;
 pub mod send;
 pub mod spawn;
 pub mod status;
 pub mod stop;
+pub mod subscribe;
 pub mod sweep_pending;
 pub mod turn;
 
@@ -119,5 +121,18 @@ pub fn register_all(iii: &Arc<IIIClient>, deps: &Arc<Deps>) {
         |d, r| async move { sweep_pending::handle(&d, r).await },
     );
 
+    // Internal session::deleted cleanup — registered, kept off the catalog.
+    register(
+        iii,
+        deps,
+        crate::subscriptions::ON_SESSION_DELETED_ID,
+        crate::subscriptions::ON_SESSION_DELETED_DESC,
+        |d, r| async move { on_session_deleted::handle(&d, r).await },
+    );
+
+    // The single shared subscription fire handler — registered once, kept off
+    // the catalog. Bound to by every subscription's trigger via the engine proxy.
+    crate::subscriptions::notify_agent::register(deps.clone());
+
     tracing::info!("all harness::* functions registered");
 }

harness/src/functions/on_session_deleted.rs

@@ -0,0 +1,41 @@
+//! `harness::on-session-deleted` — drop a deleted session's ephemeral
+//! subscriptions. Bound to session-manager's `session::deleted` trigger.
+
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::deps::Deps;
+use crate::error::HarnessError;
+
+/// `session::deleted` payload (only the field we read).
+#[derive(Debug, Clone, Deserialize, JsonSchema)]
+pub struct SessionDeletedEvent {
+    pub session_id: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
+pub struct SessionDeletedAck {
+    pub ok: bool,
+    pub removed: u64,
+}
+
+pub async fn handle(
+    deps: &Deps,
+    event: SessionDeletedEvent,
+) -> Result<SessionDeletedAck, HarnessError> {
+    let dropped = deps.subscriptions.take_session(&event.session_id);
+    let removed = dropped.len() as u64;
+    if removed > 0 {
+        for (_sub_id, trigger_id) in dropped {
+            if let Some(trigger_id) = trigger_id {
+                crate::functions::subscribe::unregister_engine_trigger(deps, &trigger_id).await;
+            }
+        }
+        tracing::info!(
+            session_id = %event.session_id,
+            removed,
+            "session deleted: ephemeral subscriptions dropped"
+        );
+    }
+    Ok(SessionDeletedAck { ok: true, removed })
+}