diff --git a/console/web/src/components/chat/Message.tsx b/console/web/src/components/chat/Message.tsx
index 3bd39dc8d..838506d16 100644
--- a/console/web/src/components/chat/Message.tsx
+++ b/console/web/src/components/chat/Message.tsx
@@ -33,7 +33,11 @@ export function Message({
}: MessageProps) {
switch (message.role) {
case 'user':
- return
+ return message.notification ? (
+
+ ) : (
+
+ )
case 'assistant':
return
case 'thought':
@@ -124,6 +128,15 @@ function CompactionMarker({ message }: { message: SystemMessageType }) {
)
}
+function NotificationMessage({ message }: { message: UserMessageType }) {
+ return (
+
+ ๐
+ {message.content}
+
+ )
+}
+
function UserMessage({ message }: { message: UserMessageType }) {
return (
diff --git a/console/web/src/hooks/use-conversations.ts b/console/web/src/hooks/use-conversations.ts
index feb0dfc08..860526301 100644
--- a/console/web/src/hooks/use-conversations.ts
+++ b/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,
diff --git a/console/web/src/lib/sessions/entry-mapper.test.ts b/console/web/src/lib/sessions/entry-mapper.test.ts
index 32738897a..f7b7943ac 100644
--- a/console/web/src/lib/sessions/entry-mapper.test.ts
+++ b/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',
diff --git a/console/web/src/lib/sessions/entry-mapper.ts b/console/web/src/lib/sessions/entry-mapper.ts
index 4f84f97a7..fee85525d 100644
--- a/console/web/src/lib/sessions/entry-mapper.ts
+++ b/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]
}
diff --git a/console/web/src/lib/sessions/types.ts b/console/web/src/lib/sessions/types.ts
index 9a9c39083..5958febde 100644
--- a/console/web/src/lib/sessions/types.ts
+++ b/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
}
export const SESSION_TRIGGER_TYPES = [
diff --git a/console/web/src/types/chat.ts b/console/web/src/types/chat.ts
index 4b7b1a009..03aec4b4c 100644
--- a/console/web/src/types/chat.ts
+++ b/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 {
diff --git a/harness/Cargo.lock b/harness/Cargo.lock
index 452332fc2..c029992e5 100644
--- a/harness/Cargo.lock
+++ b/harness/Cargo.lock
@@ -502,7 +502,7 @@ dependencies = [
[[package]]
name = "harness"
-version = "1.0.4"
+version = "1.0.6"
dependencies = [
"anyhow",
"async-trait",
diff --git a/harness/prompts/anthropic.txt b/harness/prompts/anthropic.txt
index 1f14666d8..ac57cae16 100644
--- a/harness/prompts/anthropic.txt
+++ b/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: }`.
# Discovery
diff --git a/harness/prompts/gpt.txt b/harness/prompts/gpt.txt
index 7b81967f7..c30acc33b 100644
--- a/harness/prompts/gpt.txt
+++ b/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: }`.
## Discovery
diff --git a/harness/prompts/kimi.txt b/harness/prompts/kimi.txt
index 8c2eb69ad..91e491482 100644
--- a/harness/prompts/kimi.txt
+++ b/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: }`.
# Discovery
diff --git a/harness/src/configuration.rs b/harness/src/configuration.rs
index eee742307..c8eb1f1e2 100644
--- a/harness/src/configuration.rs
+++ b/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>>;
@@ -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>,
+ _session_deleted: Option,
+}
+
+impl TriggerHandles {
+ pub fn new(sweep: Option, session_deleted: Option) -> 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 {
)
}
+pub fn bind_session_deleted(iii: &IIIClient) -> Option {
+ 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>, new: Option) {
diff --git a/harness/src/deferred.rs b/harness/src/deferred.rs
index 4afcb943e..3b1bceecf 100644
--- a/harness/src/deferred.rs
+++ b/harness/src/deferred.rs
@@ -90,7 +90,6 @@ 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;
}
@@ -98,8 +97,18 @@ pub async fn resolve(
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(
diff --git a/harness/src/deps.rs b/harness/src/deps.rs
index 91f8ff9db..df2543d31 100644
--- a/harness/src/deps.rs
+++ b/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,
}
impl Deps {
@@ -41,6 +43,7 @@ impl Deps {
events,
hooks,
locks: SessionLocks::new(),
+ subscriptions: Arc::new(SubscriptionRegistry::new()),
}
}
diff --git a/harness/src/functions/function_trigger.rs b/harness/src/functions/function_trigger.rs
index 36c9fe7a2..79f6eb803 100644
--- a/harness/src/functions/function_trigger.rs
+++ b/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)
diff --git a/harness/src/functions/mod.rs b/harness/src/functions/mod.rs
index 1f71171a5..5d0815ec4 100644
--- a/harness/src/functions/mod.rs
+++ b/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, deps: &Arc) {
|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");
}
diff --git a/harness/src/functions/on_session_deleted.rs b/harness/src/functions/on_session_deleted.rs
new file mode 100644
index 000000000..e6896024e
--- /dev/null
+++ b/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 {
+ 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 })
+}
diff --git a/harness/src/functions/send.rs b/harness/src/functions/send.rs
index a13680d39..0de48983f 100644
--- a/harness/src/functions/send.rs
+++ b/harness/src/functions/send.rs
@@ -182,6 +182,42 @@ pub async fn start(deps: &Deps, req: SendRequest) -> Result,
+ origin: Option<&Value>,
+) -> Result {
+ let cfg = deps.cfg().await;
+ let session = deps.session().await;
+
+ let options = crate::state::get_turn(&deps.iii, session_id, cfg.session_timeout_ms)
+ .await?
+ .map(|rec| rec.options)
+ .ok_or_else(|| {
+ HarnessError::InvalidRequest(format!(
+ "cannot deliver notification to session `{session_id}`: it has no prior turn to \
+ inherit model/options from"
+ ))
+ })?;
+
+ session
+ .append(session_id, &message, entry_id, None, origin)
+ .await?;
+ seed_or_merge(deps, &cfg, session_id, options).await
+}
+
pub(crate) fn normalize_message(input: MessageInput) -> Result {
match input {
MessageInput::Text(text) => Ok(AgentMessage::User(UserMessage {
diff --git a/harness/src/functions/subscribe.rs b/harness/src/functions/subscribe.rs
new file mode 100644
index 000000000..59713b9b2
--- /dev/null
+++ b/harness/src/functions/subscribe.rs
@@ -0,0 +1,321 @@
+//! Subscriptions โ register an ephemeral iii trigger and be notified when it
+//! fires, instead of polling (harness.md ยง Subscriptions). The agent calls
+//! `engine::register_trigger` / `engine::unregister_trigger`; the harness
+//! intercepts those calls (see [`invoke`]) so the trusted owning session,
+//! `harness::notify_agent` target, and engine-proxied subscription metadata are
+//! injected, and teardown stays owner-checked โ the agent can never supply those.
+
+use iii_sdk::protocol::TriggerRequest;
+use iii_sdk::TriggerAction;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+
+use crate::clients::EngineClient;
+use crate::deps::Deps;
+use crate::error::HarnessError;
+use crate::policy::CompiledPolicy;
+use crate::subscriptions::{self, NOTIFY_AGENT_ID};
+use crate::trigger::{self, ResultData};
+use crate::types::content::ContentBlock;
+
+/// The engine function the agent calls to subscribe. The harness intercepts it
+/// (the agent never reaches the raw engine registrar) so it can stamp the
+/// trusted session and bind the trigger to `harness::notify_agent`.
+pub const REGISTER_TRIGGER_ID: &str = "engine::register_trigger";
+
+/// The engine function the agent calls to unsubscribe. The harness intercepts it
+/// so it resolves the caller's subscription, enforces ownership, and unregisters
+/// the underlying engine trigger.
+pub const UNREGISTER_TRIGGER_ID: &str = "engine::unregister_trigger";
+
+/// Agent-facing subscription contract.
+#[derive(Debug, Clone, Deserialize, JsonSchema)]
+#[schemars(rename = "SubscribeArgs")]
+pub struct SubscribeRequest {
+ /// The iii trigger type to listen on: `cron`, `state`, `stream`, or another
+ /// worker's custom trigger type (e.g. `approval::pending-resolved`). For an
+ /// ad-hoc signal, subscribe to `state` on a key and have the signaller call
+ /// `state::set` on it (no dedicated emit needed โ the engine fans the trigger
+ /// out to every subscriber).
+ pub trigger_type: String,
+ /// The trigger config, passed verbatim to the engine โ e.g.
+ /// `{ "expression": "0 */5 * * * *" }` for cron, or a `state` scope/key.
+ #[serde(default)]
+ pub config: Value,
+ /// A short human label echoed back in the notification text.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub label: Option,
+ /// Auto-unsubscribe after the first delivered notification. Defaults to true
+ /// for one-shot-ish types (state / stream / custom trigger types), false for
+ /// recurring `cron`.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub once: Option,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
+pub struct SubscribeResponse {
+ pub subscription_id: String,
+ /// The effective `once` flag applied (after the per-type default).
+ pub once: bool,
+}
+
+/// The single per-call invocation chokepoint. Subscription control calls
+/// (`engine::register_trigger` / `engine::unregister_trigger`) are handled inline
+/// with the trusted owning session injected โ the model can never widen the
+/// target; everything else invokes the target normally. Every call site (the
+/// turn loop, `harness::function::trigger`, and the hook-held release path) routes
+/// through here so the trusted injection can't be bypassed.
+pub async fn invoke(
+ deps: &Deps,
+ engine: &EngineClient,
+ policy: &CompiledPolicy,
+ function_id: &str,
+ arguments: &Value,
+ session_id: &str,
+) -> ResultData {
+ match function_id {
+ REGISTER_TRIGGER_ID => intercept_register(deps, arguments, session_id).await,
+ UNREGISTER_TRIGGER_ID => intercept_unregister(deps, arguments, session_id).await,
+ _ => trigger::invoke_target(engine, policy, function_id, arguments).await,
+ }
+}
+
+fn defaults_recurring(trigger_type: &str) -> bool {
+ trigger_type == "cron"
+}
+
+fn effective_once(req: &SubscribeRequest) -> bool {
+ req.once.unwrap_or(!defaults_recurring(&req.trigger_type))
+}
+
+async fn intercept_register(deps: &Deps, args: &Value, session_id: &str) -> ResultData {
+ let req: SubscribeRequest = match serde_json::from_value(args.clone()) {
+ Ok(r) => r,
+ Err(e) => return error_result(format!("invalid subscribe arguments: {e}")),
+ };
+
+ match handle(deps, req, session_id).await {
+ Ok(resp) => ok_result(&resp),
+ Err(e) => error_result(e.to_string()),
+ }
+}
+
+async fn intercept_unregister(deps: &Deps, args: &Value, session_id: &str) -> ResultData {
+ let id = match unregister_subscription_id(args) {
+ Ok(id) => id,
+ Err(e) => return error_result(e),
+ };
+
+ if let Some(owner) = deps.subscriptions.session_of(id) {
+ if owner != session_id {
+ return error_result("subscription belongs to a different session".to_string());
+ }
+ }
+
+ let removed = match deps.subscriptions.take(id) {
+ Some((_session, trigger_id)) => {
+ if let Some(trigger_id) = trigger_id {
+ unregister_engine_trigger(deps, &trigger_id).await;
+ }
+ true
+ }
+ None => false,
+ };
+ ok_result(&json!({ "removed": removed }))
+}
+
+fn unregister_subscription_id(args: &Value) -> Result<&str, String> {
+ args.get("id")
+ .and_then(Value::as_str)
+ .ok_or_else(|| "engine::unregister_trigger requires an `id`".to_string())
+}
+
+async fn handle(
+ deps: &Deps,
+ req: SubscribeRequest,
+ session_id: &str,
+) -> Result {
+ if subscriptions::is_forbidden_trigger_type(&req.trigger_type) {
+ return Err(HarnessError::InvalidRequest(format!(
+ "cannot bind harness-internal trigger type `{}` (self-notification guard)",
+ req.trigger_type
+ )));
+ }
+
+ let once = effective_once(&req);
+
+ let sub_id = format!("sub_{}", uuid::Uuid::new_v4().simple());
+
+ deps.subscriptions
+ .try_insert(
+ &sub_id,
+ session_id,
+ subscriptions::MAX_SUBSCRIPTIONS_PER_SESSION,
+ )
+ .map_err(|_| {
+ HarnessError::InvalidRequest(format!(
+ "subscription cap reached ({} active for this session); unsubscribe first",
+ subscriptions::MAX_SUBSCRIPTIONS_PER_SESSION
+ ))
+ })?;
+
+ let resp = deps
+ .iii
+ .trigger(register_trigger_request(
+ &req,
+ &sub_id,
+ session_id,
+ once,
+ deps.cfg().await.dispatch_timeout_ms,
+ ))
+ .await;
+
+ match resp
+ .ok()
+ .and_then(|v| v.get("id").and_then(Value::as_str).map(str::to_string))
+ {
+ Some(trigger_id) => {
+ if !deps.subscriptions.set_trigger_id(&sub_id, &trigger_id) {
+ unregister_engine_trigger(deps, &trigger_id).await;
+ }
+ }
+ None => {
+ deps.subscriptions.take(&sub_id);
+ return Err(HarnessError::Dependency(format!(
+ "{REGISTER_TRIGGER_ID} `{}` failed",
+ req.trigger_type
+ )));
+ }
+ }
+
+ Ok(SubscribeResponse {
+ subscription_id: sub_id,
+ once,
+ })
+}
+
+fn register_trigger_request(
+ req: &SubscribeRequest,
+ sub_id: &str,
+ session_id: &str,
+ once: bool,
+ timeout_ms: u64,
+) -> TriggerRequest {
+ TriggerRequest {
+ function_id: REGISTER_TRIGGER_ID.to_string(),
+ payload: json!({
+ "trigger_type": req.trigger_type,
+ "function_id": NOTIFY_AGENT_ID,
+ "config": req.config.clone(),
+ "metadata": {
+ "subscription_id": sub_id,
+ "session_id": session_id,
+ "label": req.label.clone(),
+ "once": once,
+ },
+ }),
+ action: None,
+ timeout_ms: Some(timeout_ms),
+ }
+}
+
+pub async fn unregister_engine_trigger(deps: &Deps, trigger_id: &str) {
+ if let Err(e) = deps
+ .iii
+ .trigger(TriggerRequest {
+ function_id: UNREGISTER_TRIGGER_ID.to_string(),
+ payload: json!({ "id": trigger_id }),
+ action: Some(TriggerAction::Void),
+ timeout_ms: None,
+ })
+ .await
+ {
+ tracing::warn!(trigger_id, error = %e, "subscription trigger unregister failed");
+ }
+}
+
+fn ok_result(value: &T) -> ResultData {
+ let details = serde_json::to_value(value).unwrap_or(Value::Null);
+ ResultData {
+ content: vec![ContentBlock::text(
+ serde_json::to_string(&details).unwrap_or_default(),
+ )],
+ is_error: false,
+ details,
+ }
+}
+
+fn error_result(msg: String) -> ResultData {
+ ResultData {
+ content: vec![ContentBlock::text(msg.clone())],
+ is_error: true,
+ details: json!({ "error": msg }),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn register_request_stamps_trusted_target_and_session() {
+ let req: SubscribeRequest = serde_json::from_value(json!({
+ "trigger_type": "state",
+ "config": { "scope": "job", "key": "42" },
+ "label": "done",
+ "once": false,
+ "function_id": "evil::handler",
+ "metadata": { "session_id": "s_attacker" }
+ }))
+ .unwrap();
+
+ let request = register_trigger_request(&req, "sub_trusted", "s_trusted", false, 123);
+
+ assert_eq!(request.function_id, REGISTER_TRIGGER_ID);
+ assert_eq!(request.timeout_ms, Some(123));
+ assert_eq!(request.payload["trigger_type"], "state");
+ assert_eq!(request.payload["function_id"], NOTIFY_AGENT_ID);
+ assert_eq!(
+ request.payload["config"],
+ json!({ "scope": "job", "key": "42" })
+ );
+ assert_eq!(
+ request.payload["metadata"]["subscription_id"],
+ "sub_trusted"
+ );
+ assert_eq!(request.payload["metadata"]["session_id"], "s_trusted");
+ assert_eq!(request.payload["metadata"]["label"], "done");
+ assert_eq!(request.payload["metadata"]["once"], false);
+ }
+
+ #[test]
+ fn once_defaults_to_recurring_only_for_cron() {
+ let state: SubscribeRequest =
+ serde_json::from_value(json!({ "trigger_type": "state" })).unwrap();
+ let cron: SubscribeRequest =
+ serde_json::from_value(json!({ "trigger_type": "cron" })).unwrap();
+ let explicit: SubscribeRequest =
+ serde_json::from_value(json!({ "trigger_type": "cron", "once": true })).unwrap();
+
+ assert!(effective_once(&state));
+ assert!(!effective_once(&cron));
+ assert!(effective_once(&explicit));
+ }
+
+ #[test]
+ fn unregister_requires_string_subscription_id() {
+ assert_eq!(
+ unregister_subscription_id(&json!({})).unwrap_err(),
+ "engine::unregister_trigger requires an `id`"
+ );
+ assert_eq!(
+ unregister_subscription_id(&json!({ "id": 42 })).unwrap_err(),
+ "engine::unregister_trigger requires an `id`"
+ );
+ assert_eq!(
+ unregister_subscription_id(&json!({ "id": "sub_1" })).unwrap(),
+ "sub_1"
+ );
+ }
+}
diff --git a/harness/src/lib.rs b/harness/src/lib.rs
index ddcdf787d..67a6f58a6 100644
--- a/harness/src/lib.rs
+++ b/harness/src/lib.rs
@@ -25,6 +25,7 @@ pub mod policy;
pub mod prompt;
pub mod state;
pub mod subagent;
+pub mod subscriptions;
pub mod surface;
pub mod trigger;
pub mod turn_loop;
diff --git a/harness/src/main.rs b/harness/src/main.rs
index 224a2c121..f4e81f028 100644
--- a/harness/src/main.rs
+++ b/harness/src/main.rs
@@ -124,11 +124,12 @@ async fn main() -> Result<()> {
functions::register_all(&iii, &deps);
- // Bind the cron pending-sweep; retain the handle so a sweep_expression
- // change re-binds it live.
- let handles = Arc::new(TriggerHandles {
- sweep: std::sync::Mutex::new(configuration::bind_sweep(&iii, &cfg)),
- });
+ // Bind lifecycle triggers and retain their handles for the worker lifetime.
+ // The sweep binding is hot-reloaded when sweep_expression changes.
+ let handles = Arc::new(TriggerHandles::new(
+ configuration::bind_sweep(&iii, &cfg),
+ configuration::bind_session_deleted(&iii),
+ ));
discovery::seed(&iii, &functions_cell, cfg.dispatch_timeout_ms).await;
discovery::register_functions_trigger(&iii, functions_cell, cfg.dispatch_timeout_ms);
@@ -138,7 +139,7 @@ async fn main() -> Result<()> {
.context("registering the configuration change trigger")?;
tracing::info!(
- "harness ready: 8 harness::* functions + turn events + hook points + reactive function-registry cache"
+ "harness ready: harness::* functions + subscriptions + turn events + hook points + reactive function-registry cache"
);
tokio::signal::ctrl_c().await?;
diff --git a/harness/src/subscriptions/mod.rs b/harness/src/subscriptions/mod.rs
new file mode 100644
index 000000000..63d2c4d4a
--- /dev/null
+++ b/harness/src/subscriptions/mod.rs
@@ -0,0 +1,72 @@
+//! Agent-facing ephemeral event subscriptions (harness.md ยง Subscriptions).
+//!
+//! The agent SUBSCRIBES by calling `engine::register_trigger`, which the harness
+//! INTERCEPTS (see [`crate::functions::subscribe::invoke`]) to
+//! register an EPHEMERAL listener on any iii trigger type (`cron`, `state`,
+//! `stream`, or another worker's custom trigger type) and be NOTIFIED when it
+//! fires โ instead of polling. When it fires, the shared `harness::notify_agent`
+//! handler injects a `user`-role notification into the owning session, which
+//! wakes (idle) or steers (running) a turn. This is the opposite of
+//! `harness::spawn`: the turn is never parked.
+//!
+//! There is no harness-owned emit: the engine's trigger registry already fans a
+//! fired trigger out to every bound function, so "emitting" is just whatever
+//! already produces the trigger โ e.g. for an ad-hoc signal, subscribe to
+//! `state` on a key and have the signaller call the existing `state::set`.
+//!
+//! Routing: the interceptor registers a trigger (via `engine::register_trigger`)
+//! bound to the ONE shared `harness::notify_agent` function. The engine's
+//! per-subscription proxy round-trips the registration metadata into the fired
+//! payload under [`TRIGGER_META_KEY`] (`__metadata`). That metadata carries the
+//! subscription id, owning session, label, and `once`; the local registry
+//! validates liveness/ownership for cleanup. The owning session is injected by
+//! the harness when it intercepts the agent's `engine::register_trigger` call,
+//! never trusted from model arguments.
+
+pub mod notify_agent;
+pub mod registry;
+
+pub use registry::{CapExceeded, SubscriptionRegistry};
+
+/// Hard cap on active subscriptions per session โ a cheap DoS floor. Lifecycle
+/// otherwise leans on unsubscribe / `once` / `session::deleted` / process exit.
+pub const MAX_SUBSCRIPTIONS_PER_SESSION: usize = 64;
+
+/// The single shared subscription fire handler id. Every subscription's trigger
+/// binds to this (via `engine::register_trigger`); kept OFF the agent-facing catalog.
+pub const NOTIFY_AGENT_ID: &str = "harness::notify_agent";
+pub const NOTIFY_AGENT_DESC: &str =
+ "Internal: the shared subscription fire handler โ injects a notification into the owning \
+ session (resolved from the local subscription registry). Not called directly.";
+
+/// Reserved payload key under which the engine's subscription proxy round-trips a
+/// trigger's registration metadata into the fired payload. MUST match the engine
+/// constant `iii::trigger::TRIGGER_META_KEY`.
+pub const TRIGGER_META_KEY: &str = "__metadata";
+
+/// Internal `session::deleted` cleanup handler id.
+pub const ON_SESSION_DELETED_ID: &str = "harness::on-session-deleted";
+pub const ON_SESSION_DELETED_DESC: &str =
+ "Internal: drop a deleted session's ephemeral subscriptions. Not called directly.";
+
+/// Trigger types the agent may NOT subscribe to: the harness's own trigger
+/// types (turn events, hook points, internal sub handlers). Subscribing to
+/// these risks a self-notification loop โ a notify wakes a turn whose
+/// completion re-fires the trigger โ and would expose harness internals.
+pub fn is_forbidden_trigger_type(trigger_type: &str) -> bool {
+ trigger_type.starts_with("harness::")
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn forbids_harness_trigger_types() {
+ assert!(is_forbidden_trigger_type("harness::turn-completed"));
+ assert!(is_forbidden_trigger_type("harness::notify_agent"));
+ assert!(!is_forbidden_trigger_type("cron"));
+ assert!(!is_forbidden_trigger_type("subscribe"));
+ assert!(!is_forbidden_trigger_type("approval::pending-resolved"));
+ }
+}
diff --git a/harness/src/subscriptions/notify_agent.rs b/harness/src/subscriptions/notify_agent.rs
new file mode 100644
index 000000000..04cca8820
--- /dev/null
+++ b/harness/src/subscriptions/notify_agent.rs
@@ -0,0 +1,193 @@
+//! `harness::notify_agent` โ the ONE shared subscription fire handler.
+//!
+//! Every subscription binds (via `engine::register_trigger`) to this single
+//! function. The engine's per-subscription proxy injects the registration
+//! `metadata` into the fired payload under [`TRIGGER_META_KEY`]
+//! (`__metadata`). That trusted metadata carries the subscription id, owning
+//! session, label, and one-shot semantics; the local registry validates liveness
+//! and ownership for cleanup.
+
+use std::sync::Arc;
+
+use iii_sdk::errors::Error;
+use iii_sdk::RegisterFunction;
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+
+use crate::deps::Deps;
+use crate::subscriptions::{NOTIFY_AGENT_DESC, NOTIFY_AGENT_ID, TRIGGER_META_KEY};
+use crate::types::message::AgentMessage;
+
+#[derive(Debug, Deserialize, Serialize)]
+struct NotifyMetadata {
+ subscription_id: String,
+ session_id: String,
+ #[serde(default)]
+ label: Option,
+ #[serde(default)]
+ once: bool,
+}
+
+pub fn register(deps: Arc) {
+ let iii = deps.iii.clone();
+ iii.register_function(
+ NOTIFY_AGENT_ID,
+ RegisterFunction::new_async(move |event: Value| {
+ let deps = deps.clone();
+ async move {
+ on_fire(&deps, event).await;
+ Ok::(json!({ "ok": true }))
+ }
+ })
+ .description(NOTIFY_AGENT_DESC),
+ );
+}
+
+async fn on_fire(deps: &Deps, mut event: Value) {
+ let meta = match take_metadata(&mut event) {
+ Ok(meta) => meta,
+ Err(MetadataError::Missing) => {
+ tracing::warn!("notify_agent fire without {TRIGGER_META_KEY}; dropping");
+ return;
+ }
+ Err(MetadataError::Invalid(e)) => {
+ tracing::warn!(error = %e, "notify_agent metadata invalid; dropping");
+ return;
+ }
+ };
+
+ let Some(claim) =
+ deps.subscriptions
+ .claim_fire(&meta.subscription_id, &meta.session_id, meta.once)
+ else {
+ return;
+ };
+ if let Some(trigger_id) = claim.trigger_id.as_deref() {
+ crate::functions::subscribe::unregister_engine_trigger(deps, trigger_id).await;
+ }
+
+ let (message, origin) = notification_message(&meta, &event);
+
+ if let Err(e) = crate::functions::send::inject(
+ deps,
+ &meta.session_id,
+ message,
+ Some(&claim.entry_id),
+ Some(&origin),
+ )
+ .await
+ {
+ tracing::warn!(
+ sub_id = %meta.subscription_id,
+ session_id = %meta.session_id,
+ error = %e,
+ "subscription notification injection failed"
+ );
+ }
+}
+
+#[derive(Debug)]
+enum MetadataError {
+ Missing,
+ Invalid(serde_json::Error),
+}
+
+fn take_metadata(event: &mut Value) -> Result {
+ let meta = event
+ .as_object_mut()
+ .and_then(|o| o.remove(TRIGGER_META_KEY))
+ .ok_or(MetadataError::Missing)?;
+
+ serde_json::from_value(meta).map_err(MetadataError::Invalid)
+}
+
+fn notification_message(meta: &NotifyMetadata, event: &Value) -> (AgentMessage, Value) {
+ let summary = summarize_event(event);
+ let text = match &meta.label {
+ Some(label) => format!("[notification: {label}] {summary}"),
+ None => format!("[notification] {summary}"),
+ };
+ (
+ AgentMessage::user_text(text),
+ json!({ "notification": true }),
+ )
+}
+
+fn summarize_event(event: &Value) -> String {
+ const MAX: usize = 600;
+ let rendered = match event {
+ Value::Null => "event fired".to_string(),
+ Value::String(s) => s.clone(),
+ other => serde_json::to_string(other).unwrap_or_else(|_| "event fired".to_string()),
+ };
+ if rendered.chars().count() > MAX {
+ let mut s: String = rendered.chars().take(MAX).collect();
+ s.push_str(" โฆ(truncated)");
+ s
+ } else {
+ rendered
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ use crate::types::content::ContentBlock;
+
+ fn text_of(message: AgentMessage) -> String {
+ let AgentMessage::User(user) = message else {
+ panic!("expected user message");
+ };
+ ContentBlock::join_text(&user.content)
+ }
+
+ #[test]
+ fn metadata_is_required_valid_and_stripped() {
+ let mut missing = json!({ "event_type": "set" });
+ assert!(matches!(
+ take_metadata(&mut missing),
+ Err(MetadataError::Missing)
+ ));
+
+ let mut invalid = json!({ TRIGGER_META_KEY: { "session_id": "s" } });
+ assert!(matches!(
+ take_metadata(&mut invalid),
+ Err(MetadataError::Invalid(_))
+ ));
+
+ let mut event = json!({
+ "event_type": "set",
+ TRIGGER_META_KEY: {
+ "subscription_id": "sub_1",
+ "session_id": "s_secret",
+ "label": "done"
+ }
+ });
+ let meta = take_metadata(&mut event).unwrap();
+
+ assert_eq!(meta.subscription_id, "sub_1");
+ assert_eq!(meta.session_id, "s_secret");
+ assert_eq!(meta.label.as_deref(), Some("done"));
+ assert!(event.get(TRIGGER_META_KEY).is_none());
+ }
+
+ #[test]
+ fn notification_message_uses_label_origin_and_stripped_event() {
+ let meta = NotifyMetadata {
+ subscription_id: "sub_1".to_string(),
+ session_id: "s_1".to_string(),
+ label: Some("job finished".to_string()),
+ once: true,
+ };
+ let event = json!({ "event_type": "set", "value": { "done": true } });
+
+ let (message, origin) = notification_message(&meta, &event);
+
+ assert_eq!(origin, json!({ "notification": true }));
+ assert_eq!(
+ text_of(message),
+ r#"[notification: job finished] {"event_type":"set","value":{"done":true}}"#
+ );
+ }
+}
diff --git a/harness/src/subscriptions/registry.rs b/harness/src/subscriptions/registry.rs
new file mode 100644
index 000000000..19b936ab3
--- /dev/null
+++ b/harness/src/subscriptions/registry.rs
@@ -0,0 +1,302 @@
+//! In-memory ephemeral subscription index (harness.md ยง Subscriptions).
+//!
+//! The engine owns each subscription's trigger AND its delivery proxy; the
+//! harness keeps only the small bookkeeping it needs LOCALLY: the owning session
+//! (per-session cap, ownership check, session-deleted cleanup), a fire counter
+//! (for idempotent notification entry ids), and the engine-returned trigger id
+//! (to unregister). Fire context such as label / once is carried by the
+//! engine's registration metadata proxy, not duplicated here. No iii handles are
+//! held here.
+//!
+//! Subscriptions are intentionally NOT persisted: they live for the harness
+//! process only (the "ephemeral" contract). Lifecycle is covered by explicit
+//! `unsubscribe`, `once` self-teardown, `session::deleted`, and the engine's
+//! worker-disconnect GC on process exit โ so there is no TTL/sweep here.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, MutexGuard};
+
+/// One live subscription's local bookkeeping. Held in an `Arc` so a firing
+/// handler reads it without holding the registry lock.
+pub struct SubEntry {
+ /// Owning session โ captured at registration from the trusted dispatch
+ /// layer; the agent can never widen the target.
+ pub session_id: String,
+ /// Monotonic fire count, used only for an idempotent notification entry id.
+ seq: AtomicU64,
+ /// The engine-returned trigger id; attached just after registration (see
+ /// [`set_trigger_id`](SubscriptionRegistry::set_trigger_id)) and used to
+ /// `engine::unregister_trigger` on teardown.
+ trigger_id: Mutex>,
+}
+
+impl SubEntry {
+ fn new(session_id: String) -> Self {
+ Self {
+ session_id,
+ seq: AtomicU64::new(0),
+ trigger_id: Mutex::new(None),
+ }
+ }
+}
+
+/// Returned by [`SubscriptionRegistry::try_insert`] when the per-session cap is hit.
+#[derive(Debug)]
+pub struct CapExceeded;
+
+/// A liveness/ownership-checked fired subscription.
+#[derive(Debug, PartialEq, Eq)]
+pub struct FireClaim {
+ pub entry_id: String,
+ pub trigger_id: Option,
+}
+
+#[derive(Default)]
+struct Inner {
+ by_id: HashMap>,
+ by_session: HashMap>,
+}
+
+/// The process-wide index of active ephemeral subscriptions.
+pub struct SubscriptionRegistry {
+ inner: Mutex,
+}
+
+impl SubscriptionRegistry {
+ pub fn new() -> Self {
+ Self {
+ inner: Mutex::new(Inner::default()),
+ }
+ }
+
+ fn lock(&self) -> MutexGuard<'_, Inner> {
+ self.inner.lock().unwrap_or_else(|p| p.into_inner())
+ }
+
+ /// Atomically enforce the per-session cap and insert a placeholder entry
+ /// (its trigger id is attached later via [`set_trigger_id`](Self::set_trigger_id)).
+ /// Inserting BEFORE the engine bind means an immediate fire still finds it.
+ pub fn try_insert(
+ &self,
+ sub_id: &str,
+ session_id: &str,
+ max: usize,
+ ) -> Result<(), CapExceeded> {
+ let mut inner = self.lock();
+ let count = inner
+ .by_session
+ .get(session_id)
+ .map(HashSet::len)
+ .unwrap_or(0);
+ if count >= max {
+ return Err(CapExceeded);
+ }
+ inner
+ .by_session
+ .entry(session_id.to_string())
+ .or_default()
+ .insert(sub_id.to_string());
+ inner.by_id.insert(
+ sub_id.to_string(),
+ Arc::new(SubEntry::new(session_id.to_string())),
+ );
+ Ok(())
+ }
+
+ /// Attach the engine-returned trigger id after a successful bind. Returns
+ /// `false` if the entry is already gone (a `once` fire won the race in the
+ /// bind window) โ the caller must then unregister the orphan trigger.
+ pub fn set_trigger_id(&self, sub_id: &str, trigger_id: &str) -> bool {
+ match self.lock().by_id.get(sub_id) {
+ Some(entry) => {
+ *entry.trigger_id.lock().unwrap_or_else(|p| p.into_inner()) =
+ Some(trigger_id.to_string());
+ true
+ }
+ None => false,
+ }
+ }
+
+ /// Claim a fired subscription against local liveness/ownership. One-shot
+ /// fires remove the entry and return the trigger id for teardown; recurring
+ /// fires keep the entry and return a monotonic notification entry id.
+ pub fn claim_fire(&self, sub_id: &str, session_id: &str, once: bool) -> Option {
+ let mut inner = self.lock();
+ let entry = inner.by_id.get(sub_id).cloned()?;
+ if entry.session_id != session_id {
+ return None;
+ }
+
+ if once {
+ let entry = remove_entry(&mut inner, sub_id)?;
+ return Some(FireClaim {
+ entry_id: format!("e_notify_{sub_id}"),
+ trigger_id: trigger_id(&entry),
+ });
+ }
+
+ Some(FireClaim {
+ entry_id: format!(
+ "e_notify_{sub_id}_{}",
+ entry.seq.fetch_add(1, Ordering::SeqCst) + 1
+ ),
+ trigger_id: None,
+ })
+ }
+
+ /// The owning session of a subscription (unsubscribe ownership check).
+ pub fn session_of(&self, sub_id: &str) -> Option {
+ self.lock().by_id.get(sub_id).map(|e| e.session_id.clone())
+ }
+
+ /// Atomically remove and return `(session_id, trigger_id)`. Winner-takes-all:
+ /// only the first caller for a given id gets `Some` โ this is the `once` /
+ /// unsubscribe claim that guarantees at-most-once delivery + teardown.
+ pub fn take(&self, sub_id: &str) -> Option<(String, Option)> {
+ let mut inner = self.lock();
+ let entry = remove_entry(&mut inner, sub_id)?;
+ Some((entry.session_id.clone(), trigger_id(&entry)))
+ }
+
+ /// Remove every subscription owned by a session and return each
+ /// `(sub_id, trigger_id)` so the caller can `engine::unregister_trigger` them.
+ pub fn take_session(&self, session_id: &str) -> Vec<(String, Option)> {
+ let ids: Vec = {
+ let inner = self.lock();
+ inner
+ .by_session
+ .get(session_id)
+ .map(|s| s.iter().cloned().collect())
+ .unwrap_or_default()
+ };
+ ids.into_iter()
+ .filter_map(|id| {
+ self.take(&id)
+ .map(|(_session, trigger_id)| (id, trigger_id))
+ })
+ .collect()
+ }
+}
+
+fn trigger_id(entry: &SubEntry) -> Option {
+ entry
+ .trigger_id
+ .lock()
+ .unwrap_or_else(|p| p.into_inner())
+ .clone()
+}
+
+fn remove_entry(inner: &mut Inner, sub_id: &str) -> Option> {
+ let entry = inner.by_id.remove(sub_id)?;
+ if let Some(set) = inner.by_session.get_mut(&entry.session_id) {
+ set.remove(sub_id);
+ if set.is_empty() {
+ inner.by_session.remove(&entry.session_id);
+ }
+ }
+ Some(entry)
+}
+
+impl Default for SubscriptionRegistry {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn try_insert_enforces_cap_atomically() {
+ let reg = SubscriptionRegistry::new();
+ assert!(reg.try_insert("sub_1", "s", 2).is_ok());
+ assert!(reg.try_insert("sub_2", "s", 2).is_ok());
+ assert!(reg.try_insert("sub_3", "s", 2).is_err());
+ // A different session has its own budget.
+ assert!(reg.try_insert("sub_4", "s2", 2).is_ok());
+ }
+
+ #[test]
+ fn set_trigger_id_then_take_returns_it() {
+ let reg = SubscriptionRegistry::new();
+ reg.try_insert("sub_1", "s", 8).unwrap();
+ assert!(reg.set_trigger_id("sub_1", "trig_1"));
+ let (session, trigger_id) = reg.take("sub_1").expect("entry present");
+ assert_eq!(session, "s");
+ assert_eq!(trigger_id.as_deref(), Some("trig_1"));
+ }
+
+ #[test]
+ fn set_trigger_id_false_after_take() {
+ let reg = SubscriptionRegistry::new();
+ reg.try_insert("sub_1", "s", 8).unwrap();
+ assert!(reg.take("sub_1").is_some());
+ // Entry gone (a once-fire won the window): caller must unregister the orphan.
+ assert!(!reg.set_trigger_id("sub_1", "trig_1"));
+ }
+
+ #[test]
+ fn claim_recurring_fire_is_monotonic_and_owner_checked() {
+ let reg = SubscriptionRegistry::new();
+ reg.try_insert("sub_1", "owner", 8).unwrap();
+ assert_eq!(reg.claim_fire("sub_1", "other", false), None);
+ assert_eq!(
+ reg.claim_fire("sub_1", "owner", false),
+ Some(FireClaim {
+ entry_id: "e_notify_sub_1_1".to_string(),
+ trigger_id: None,
+ })
+ );
+ assert_eq!(
+ reg.claim_fire("sub_1", "owner", false),
+ Some(FireClaim {
+ entry_id: "e_notify_sub_1_2".to_string(),
+ trigger_id: None,
+ })
+ );
+ reg.take("sub_1");
+ assert_eq!(reg.claim_fire("sub_1", "owner", false), None);
+ }
+
+ #[test]
+ fn take_is_idempotent_winner_takes_all() {
+ let reg = SubscriptionRegistry::new();
+ reg.try_insert("sub_1", "s", 8).unwrap();
+ assert!(reg.take("sub_1").is_some());
+ assert!(reg.take("sub_1").is_none());
+ }
+
+ #[test]
+ fn session_of_and_take_session() {
+ let reg = SubscriptionRegistry::new();
+ reg.try_insert("sub_1", "s", 8).unwrap();
+ reg.try_insert("sub_2", "s", 8).unwrap();
+ reg.set_trigger_id("sub_1", "trig_1");
+ assert_eq!(reg.session_of("sub_1").as_deref(), Some("s"));
+ let mut dropped = reg.take_session("s");
+ dropped.sort();
+ assert_eq!(dropped.len(), 2);
+ assert!(reg.session_of("sub_1").is_none());
+ assert!(reg.take_session("s").is_empty());
+ }
+
+ #[test]
+ fn claim_once_fire_removes_only_for_matching_owner() {
+ let reg = SubscriptionRegistry::new();
+ reg.try_insert("sub_1", "owner", 8).unwrap();
+ reg.set_trigger_id("sub_1", "trig_1");
+
+ assert_eq!(reg.claim_fire("sub_1", "other", true), None);
+ assert_eq!(reg.session_of("sub_1").as_deref(), Some("owner"));
+ assert_eq!(
+ reg.claim_fire("sub_1", "owner", true),
+ Some(FireClaim {
+ entry_id: "e_notify_sub_1".to_string(),
+ trigger_id: Some("trig_1".to_string()),
+ })
+ );
+ assert!(reg.session_of("sub_1").is_none());
+ }
+}
diff --git a/harness/src/turn_loop.rs b/harness/src/turn_loop.rs
index aafe26978..4db4eb4ad 100644
--- a/harness/src/turn_loop.rs
+++ b/harness/src/turn_loop.rs
@@ -475,10 +475,18 @@ pub async fn run_step(
let scoped_args =
crate::workspace_inject::inject(&call.function_id, eff_args, working_dir);
- // Invoke the target, then run the post_trigger chain over the
- // result before it is appended (redaction / truncation).
- let raw =
- trigger::invoke_target(&engine, &policy, &call.function_id, &scoped_args).await;
+ // Single invocation chokepoint: subscription control calls are
+ // intercepted (trusted session injected); everything else invokes the
+ // target. Then the post_trigger chain runs over the result.
+ let raw = crate::functions::subscribe::invoke(
+ deps,
+ &engine,
+ &policy,
+ &call.function_id,
+ &scoped_args,
+ &record.session_id,
+ )
+ .await;
let (data, post_ann) = deps
.hooks
.run_post_trigger(&record, payload.step, &call.id, &call.function_id, raw)