feat(harness): agent trigger subscriptions — be notified instead of polling - #333
Merged
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughThis PR adds an ephemeral subscription system: a ChangesEphemeral subscription flow
Console notification UI
Sequence Diagram(s)sequenceDiagram
participant Agent
participant SubscribeInvoke as subscribe::invoke
participant SubscriptionRegistry
participant iii
participant NotifyAgent as notify_agent
participant SendInject as send::inject
Agent->>SubscribeInvoke: engine::register_trigger(trigger_type, config)
SubscribeInvoke->>SubscriptionRegistry: try_insert(sub_id, session_id, max)
SubscribeInvoke->>iii: trigger(TriggerRequest to harness::notify_agent)
iii-->>SubscribeInvoke: trigger_id
SubscribeInvoke->>SubscriptionRegistry: set_trigger_id(sub_id, trigger_id)
SubscribeInvoke-->>Agent: subscription_id, once
iii->>NotifyAgent: fired event + metadata
NotifyAgent->>SubscriptionRegistry: claim_fire(sub_id, session_id, once)
SubscriptionRegistry-->>NotifyAgent: FireClaim(entry_id, trigger_id)
NotifyAgent->>iii: unregister_trigger (if one-shot)
NotifyAgent->>SendInject: inject(session_id, notification message)
SendInject-->>Agent: notification appears in session
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Suggested reviewers
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
skill-check — worker0 verified, 27 skipped (no docs/).
Four for four. Nicely done. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@harness/src/functions/unsubscribe.rs`:
- Around line 28-39: The owner check in unsubscribe handling can be bypassed
when req.session_id is missing, so make the guard fail closed in
unsubscribe::handle by rejecting requests without an injected session_id before
calling deps.subscriptions.remove. Keep the existing ownership comparison
against entry.session_id, but require a present session_id for any unsubscribe
path and return HarnessError::InvalidRequest when it is absent.
In `@harness/src/main.rs`:
- Around line 141-143: The readiness log in main should no longer hardcode an
outdated function count; update the message in harness::main so it matches the
12 register calls performed by register_all in harness::functions::mod, keeping
the displayed count aligned with the actual harness::* registrations.
In `@harness/src/subscriptions/handler.rs`:
- Around line 67-81: The subscription teardown in the notification handler
removes entries even when `crate::functions::send::inject` fails, which can drop
the only `once` delivery. Update the logic in the subscription handler so
`registry.remove(sub_id)` is only executed after a successful inject, and keep
failed injections in the registry for retry or later delivery. Use the existing
`decision.remove_after`, `registry.remove`, and `inject` flow in the handler to
gate removal on the success path.
In `@harness/src/subscriptions/registry.rs`:
- Around line 122-155: The race in record_and_decide is caused by separately
loading last_delivered_at, checking min_interval_ms, and then storing the new
timestamp, which lets concurrent callers all pass the gate and deliver. Update
registry::record_and_decide to atomically claim delivery with a compare_exchange
loop on last_delivered_at (or equivalent CAS-based gate) so only one caller can
win per interval, preserving once and coalescing behavior. Also remove the i64
narrowing from min_interval_ms by keeping the interval comparison in a
consistent unsigned type and using the existing now/last values safely so long
intervals cannot overflow or flip negative.
- Around line 85-104: The subscription cap check in `Registry::count_for` and
`Registry::insert` is split across two locks, so concurrent `subscribe::handle`
calls can exceed `max_subscriptions_per_session`. Add a single registry method
that performs the count check and insertion atomically under the same lock, then
update `subscribe::handle` to call that new method instead of calling
`count_for()` and `insert()` separately. Keep the existing `Registry` internals
(`by_session`, `by_id`) as the place where the combined check-and-insert logic
lives.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: ec5744fb-2244-4c54-80ea-027c0ddfd68f
📒 Files selected for processing (24)
harness/prompts/anthropic.txtharness/prompts/gpt.txtharness/prompts/kimi.txtharness/src/config.rsharness/src/deps.rsharness/src/functions/function_trigger.rsharness/src/functions/mod.rsharness/src/functions/on_session_deleted.rsharness/src/functions/send.rsharness/src/functions/subscribe.rsharness/src/functions/subscriptions_list.rsharness/src/functions/sweep_pending.rsharness/src/functions/unsubscribe.rsharness/src/lib.rsharness/src/main.rsharness/src/subscriptions/handler.rsharness/src/subscriptions/mod.rsharness/src/subscriptions/registry.rsharness/src/surface.rsharness/src/turn_loop.rsharness/tests/golden/schemas/harness.subscribe.jsonharness/tests/golden/schemas/harness.subscriptions.jsonharness/tests/golden/schemas/harness.unsubscribe.jsonharness/tests/schemas.rs
Comment on lines
+28
to
+39
| // Owner check: only the owning session may tear down its subscription. | ||
| if let (Some(entry), Some(caller)) = ( | ||
| deps.subscriptions.get(&req.subscription_id), | ||
| req.session_id.as_ref(), | ||
| ) { | ||
| if &entry.session_id != caller { | ||
| return Err(HarnessError::InvalidRequest( | ||
| "subscription belongs to a different session".to_string(), | ||
| )); | ||
| } | ||
| } | ||
| let removed = deps.subscriptions.remove(&req.subscription_id); |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
Owner check is bypassed when session_id is None.
The ownership guard only fires when req.session_id is Some. If a call reaches this handler without an injected session_id, the check is skipped entirely and remove() runs against any subscription_id. The trusted dispatch layer (inject_owner_session) stamps the owner for harness::unsubscribe, so this is currently defense-in-depth rather than an active hole — but failing closed is cheap insurance against a future path that doesn't inject.
🛡️ Proposed fail-closed guard
- // Owner check: only the owning session may tear down its subscription.
- if let (Some(entry), Some(caller)) = (
- deps.subscriptions.get(&req.subscription_id),
- req.session_id.as_ref(),
- ) {
- if &entry.session_id != caller {
- return Err(HarnessError::InvalidRequest(
- "subscription belongs to a different session".to_string(),
- ));
- }
- }
+ // Owner check: only the owning session may tear down its subscription.
+ if let Some(entry) = deps.subscriptions.get(&req.subscription_id) {
+ match req.session_id.as_ref() {
+ Some(caller) if &entry.session_id == caller => {}
+ _ => {
+ return Err(HarnessError::InvalidRequest(
+ "subscription belongs to a different session".to_string(),
+ ));
+ }
+ }
+ }Confirm no trusted/internal caller invokes harness::unsubscribe without an injected session_id:
#!/bin/bash
rg -nP --type=rust -C3 'UNSUBSCRIBE_ID|unsubscribe::handle|inject_owner_session'🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/functions/unsubscribe.rs` around lines 28 - 39, The owner check
in unsubscribe handling can be bypassed when req.session_id is missing, so make
the guard fail closed in unsubscribe::handle by rejecting requests without an
injected session_id before calling deps.subscriptions.remove. Keep the existing
ownership comparison against entry.session_id, but require a present session_id
for any unsubscribe path and return HarnessError::InvalidRequest when it is
absent.
Comment on lines
+67
to
+81
| if let Err(e) = | ||
| crate::functions::send::inject(deps, &entry.session_id, message, Some(&entry_id)).await | ||
| { | ||
| tracing::warn!( | ||
| sub_id, | ||
| session_id = %entry.session_id, | ||
| error = %e, | ||
| "subscription notification injection failed" | ||
| ); | ||
| } | ||
|
|
||
| if decision.remove_after { | ||
| registry.remove(sub_id); | ||
| } | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
fd registry.rs --full-path 'harness/src/subscriptions' --exec sed -n '1,260p'Repository: iii-hq/workers
Length of output: 9195
once subscriptions removed even when injection fails, potentially losing the sole notification.
Lines 78-80 trigger registry.remove(sub_id) based on decision.remove_after without verifying if the inject call at Line 67 succeeded. If injection fails (e.g., missing session state), the once subscription is torn down and the event is silently dropped. Defer the teardown until after successful injection to ensure the agent receives the notification:
Relevant code
if let Err(e) =
crate::functions::send::inject(deps, &entry.session_id, message, Some(&entry_id)).await
{
tracing::warn!(
sub_id,
session_id = %entry.session_id,
error = %e,
"subscription notification injection failed"
);
}
if decision.remove_after {
registry.remove(sub_id);
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/subscriptions/handler.rs` around lines 67 - 81, The subscription
teardown in the notification handler removes entries even when
`crate::functions::send::inject` fails, which can drop the only `once` delivery.
Update the logic in the subscription handler so `registry.remove(sub_id)` is
only executed after a successful inject, and keep failed injections in the
registry for retry or later delivery. Use the existing `decision.remove_after`,
`registry.remove`, and `inject` flow in the handler to gate removal on the
success path.
Comment on lines
+122
to
+155
| /// Coalesce + expiry gate for one fire. Always increments `fire_count`; | ||
| /// returns `None` only when the subscription is already gone. | ||
| pub fn record_and_decide(&self, sub_id: &str, now: i64) -> Option<FireDecision> { | ||
| let entry = self.get(sub_id)?; | ||
| let fire_count = entry.fire_count.fetch_add(1, Ordering::SeqCst) + 1; | ||
|
|
||
| if let Some(exp) = entry.expires_at { | ||
| if now >= exp { | ||
| return Some(FireDecision { | ||
| deliver: false, | ||
| remove_after: true, | ||
| fire_count, | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| let last = entry.last_delivered_at.load(Ordering::SeqCst); | ||
| if entry.min_interval_ms > 0 | ||
| && last > 0 | ||
| && now.saturating_sub(last) < entry.min_interval_ms as i64 | ||
| { | ||
| return Some(FireDecision { | ||
| deliver: false, | ||
| remove_after: false, | ||
| fire_count, | ||
| }); | ||
| } | ||
|
|
||
| entry.last_delivered_at.store(now, Ordering::SeqCst); | ||
| Some(FireDecision { | ||
| deliver: true, | ||
| remove_after: entry.once, | ||
| fire_count, | ||
| }) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Expectation after the fix: no `min_interval_ms as i64` remains, and the fire gate
# claims delivery with a CAS or an equivalent single critical section.
rg -n -C2 'min_interval_ms as i64|record_and_decide|compare_exchange' \
harness/src/subscriptions/registry.rs \
harness/src/subscriptions/handler.rs \
harness/src/functions/subscribe.rsRepository: iii-hq/workers
Length of output: 2239
🏁 Script executed:
#!/bin/bash
# Verify the specific lines mentioned in the original concern
sed -n '139,145p' harness/src/subscriptions/registry.rs
sed -n '262,268p' harness/src/subscriptions/registry.rsRepository: iii-hq/workers
Length of output: 624
Implement atomic delivery claiming and fix integer narrowing in interval logic.
Two critical issues impact correctness and stability:
- Race Condition on Fire Gate: The logic loads
last_delivered_at, checks the interval, and then unconditionally stores the new timestamp. This allows concurrent requests to read the same old value, pass the interval check simultaneously, and all deliver the message. This breaksoncesubscriptions (causing multiple fires) and bypassesmin_interval_mscoalescing. - Integer Narowing: Casting
u64intervals toi64(min_interval_ms as i64) causes overflow to negative numbers for intervals exceeding 292 years. This negates the comparison innow.saturating_sub(last) < ..., effectively disabling the rate limit for valid, long-running subscriptions.
Use a compare_exchange loop to atomically check-and-update the timestamp, and ensure all interval calculations use consistent unsigned types.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/subscriptions/registry.rs` around lines 122 - 155, The race in
record_and_decide is caused by separately loading last_delivered_at, checking
min_interval_ms, and then storing the new timestamp, which lets concurrent
callers all pass the gate and deliver. Update registry::record_and_decide to
atomically claim delivery with a compare_exchange loop on last_delivered_at (or
equivalent CAS-based gate) so only one caller can win per interval, preserving
once and coalescing behavior. Also remove the i64 narrowing from min_interval_ms
by keeping the interval comparison in a consistent unsigned type and using the
existing now/last values safely so long intervals cannot overflow or flip
negative.
ytallo
force-pushed
the
feat/agent-trigger-subscriptions
branch
from
June 25, 2026 14:26
423267d to
a0a9394
Compare
ytallo
marked this pull request as draft
June 25, 2026 14:52
ytallo
force-pushed
the
feat/agent-trigger-subscriptions
branch
from
June 25, 2026 20:59
0655db0 to
bc339f2
Compare
ytallo
force-pushed
the
feat/agent-trigger-subscriptions
branch
from
June 26, 2026 00:50
bc339f2 to
5d2a77d
Compare
ytallo
changed the title
ytallo
force-pushed
the
feat/agent-trigger-subscriptions
branch
from
June 26, 2026 01:05
5d2a77d to
281112a
Compare
ytallo
changed the title
ytallo
force-pushed
the
feat/agent-trigger-subscriptions
branch
from
June 30, 2026 02:58
18ee054 to
08516a7
Compare
ytallo
marked this pull request as ready for review
June 30, 2026 02:58
…olling Give the agent a way to be NOTIFIED about an event instead of polling for it. harness::subscribe registers an ephemeral iii trigger of any type (cron, state, stream, or another worker's custom trigger type); when it fires, an internal per-subscription handler injects a user-role notification into the owning session, waking or steering a turn — non-blocking, the opposite of harness::spawn's park-and-resume. harness::unsubscribe tears one down. - Per-subscription handler: built-in triggers fire the bound function with only the engine event payload (no config/metadata redelivery), so each subscription registers a unique harness::sub::<uuid> whose closure captures the owning session — the engine routing IS the addressing. A single shared handler is impossible without engine changes. - Owner confinement: the owning session is stamped onto subscribe/unsubscribe by the trusted dispatch layer (the turn loop, harness::function::trigger, and the approval hold->release path), never trusted from model arguments. - Notification bridge: send::inject appends a user-role message reusing the session's last turn options (model / provider / dispatch policy), so a woken turn keeps the agent's capabilities. - Ephemeral, session-scoped lifecycle: explicit unsubscribe, one-shot `once` auto-teardown, and session::deleted cleanup, with a per-session cap as a DoS floor. Not persisted across harness restart — the engine GCs a worker's triggers on disconnect. - No harness-owned emit: the engine fans a fired trigger out to every subscriber, so an ad-hoc signal is just subscribing to a `state` key and calling the existing state::set. The system prompts point the agent at this. Tests: unit coverage for owner-session injection, the forbidden-trigger guard, the per-type `once` default, and the event-summary excerpt; wire-schema goldens for subscribe and unsubscribe.
Enhance the Message component to display notifications for user messages. Introduced a new NotificationMessage component that renders a notification with an optional label. Updated the entry-mapper to include notification data in user messages and modified the useConversations hook to handle the new origin property for events. Updated types to accommodate the new notification structure.
…engine::register_trigger/unregister_trigger Updated the subscription mechanism to utilize engine::register_trigger and engine::unregister_trigger instead of the previous harness::subscribe and harness::unsubscribe functions. This change enhances ownership checks and integrates a shared notification handler, harness::notify_agent, for managing subscription notifications. The new approach ensures that the owning session is injected during the registration process, improving security and reliability. Removed deprecated functions and updated related documentation to reflect the new subscription model.
Cleaned up the code by removing outdated comments related to the previous subscription functions harness::subscribe and harness::unsubscribe. This aligns with the recent transition to using engine::register_trigger and engine::unregister_trigger, streamlining the codebase and improving clarity.
…calls Use deps.iii.trigger directly for engine::register_trigger / engine::unregister_trigger instead of the EngineClient::dispatch wrapper, which renamed the iii.trigger term. register stays synchronous with the configured dispatch timeout (it needs the returned trigger id); the fire-and-forget unregisters use TriggerAction::Void.
…cture Refactor the subscription mechanism to streamline the handling of notifications. The `UserMessage` type now uses a boolean for the notification field instead of an object, simplifying the structure. The entry-mapper has been updated accordingly to reflect this change. Additionally, the subscription registry has been enhanced to manage session IDs and labels more effectively, ensuring that the owning session is correctly associated with each subscription. This update improves clarity and consistency in the notification process across the harness.
…ation structure Refactor the subscription mechanism to improve the handling of metadata associated with notifications. The `NotifyMetadata` struct has been introduced to encapsulate subscription details, including session ID, label, and one-shot semantics. Updates to the subscription registry ensure that ownership checks are enforced, and the notification process is streamlined. This change enhances clarity and consistency in how notifications are processed and improves the overall security of the subscription system.
Enhance the entry-mapper by adding an optional origin parameter to the userItem function, allowing for more flexible notification handling. Update the entrySegments function to correctly identify trusted notification entries based on the new structure. Additionally, refactor the subscription notification process to streamline metadata extraction and improve clarity in the notification message generation. This update ensures better handling of user messages and enhances the overall notification system.
The rebase onto main landed on a newer iii_sdk whose export surface moved: TriggerRequest is now under iii_sdk::protocol, the root IIIError alias is gone in favor of iii_sdk::errors::Error, and the III client type is IIIClient. Update the subscription files git auto-merged so they compile against the current SDK.
ytallo
force-pushed
the
feat/agent-trigger-subscriptions
branch
from
July 1, 2026 00:59
08516a7 to
21f9f59
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
harness/src/turn_loop.rs (1)
469-474: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate
working_dirextraction.The same
record.options.metadata.as_ref().and_then(|m| m.get("working_dir")).and_then(Value::as_str)extraction appears both here (forbase_dirstamping) and inwith_working_dir_aidbelow. Consider factoring it into a small helper to avoid the two copies drifting.♻️ Proposed refactor
+fn turn_working_dir(record: &TurnRecord) -> Option<&str> { + record + .options + .metadata + .as_ref() + .and_then(|m| m.get("working_dir")) + .and_then(Value::as_str) +} + // ... at the call.function_id scoping site: - let working_dir = record - .options - .metadata - .as_ref() - .and_then(|m| m.get("working_dir")) - .and_then(Value::as_str); + let working_dir = turn_working_dir(record); // ... in with_working_dir_aid: - let working_dir = record - .options - .metadata - .as_ref() - .and_then(|m| m.get("working_dir")) - .and_then(Value::as_str); + let working_dir = turn_working_dir(record);Also applies to: 1042-1047
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@harness/src/turn_loop.rs` around lines 469 - 474, The `working_dir` lookup is duplicated in `turn_loop.rs` in both the `base_dir` stamping logic and `with_working_dir_aid`, so factor that `record.options.metadata.as_ref().and_then(|m| m.get("working_dir")).and_then(Value::as_str)` chain into a small shared helper and use it in both places. Keep the helper near the existing `TurnLoop`/record handling code so the extraction stays consistent and the two call sites cannot drift apart.console/web/src/components/chat/Message.tsx (1)
131-138: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider an accessible cue for notification messages.
Visually distinguished via border/icon/font, but nothing conveys "this is a system notification" to screen readers — same experience as a plain user message audibly. Consider
role="status"(or ansr-onlyprefix like "Notification:") so assistive tech can tell it apart, and optionally to announce it as it streams in.♿ Proposed accessibility tweak
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"> + <article + role="status" + 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="sr-only">Notification:</span> <span className="break-words">{message.content}</span> </article> ) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@console/web/src/components/chat/Message.tsx` around lines 131 - 138, The NotificationMessage component is only visually distinct, so screen readers can’t tell it apart from a normal chat message. Update the NotificationMessage JSX to expose its purpose to assistive tech, for example by adding an appropriate live/status role or an accessible text prefix. Keep the change localized to NotificationMessage in Message.tsx and preserve the existing visual styling.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@harness/src/functions/subscribe.rs`:
- Around line 223-236: The cleanup path in unregister_engine_trigger currently
only emits a warning on trigger unregister failure, which can leave engine-side
triggers orphaned with no reconciliation. Update unregister_engine_trigger to
either persist failed unregisters for later retry via a sweep/backoff mechanism
like the existing sweep_pending pattern, or at minimum escalate the tracing call
to error and add a metric so failures are observable; use the existing
unregister_engine_trigger and TriggerRequest flow as the entry point for the
fix.
In `@harness/src/subscriptions/registry.rs`:
- Around line 164-179: Make take_session atomic with respect to the registry
state: it currently snapshots ids from by_session under lock and then calls take
after releasing the lock, which allows a concurrent try_insert to sneak in and
survive cleanup. Update Registry::take_session to hold the lock across both
collecting and removing the session’s subscriptions, or otherwise remove entries
from by_session and the related maps in one locked operation using the existing
Registry::lock, take, and by_session members.
---
Nitpick comments:
In `@console/web/src/components/chat/Message.tsx`:
- Around line 131-138: The NotificationMessage component is only visually
distinct, so screen readers can’t tell it apart from a normal chat message.
Update the NotificationMessage JSX to expose its purpose to assistive tech, for
example by adding an appropriate live/status role or an accessible text prefix.
Keep the change localized to NotificationMessage in Message.tsx and preserve the
existing visual styling.
In `@harness/src/turn_loop.rs`:
- Around line 469-474: The `working_dir` lookup is duplicated in `turn_loop.rs`
in both the `base_dir` stamping logic and `with_working_dir_aid`, so factor that
`record.options.metadata.as_ref().and_then(|m|
m.get("working_dir")).and_then(Value::as_str)` chain into a small shared helper
and use it in both places. Keep the helper near the existing `TurnLoop`/record
handling code so the extraction stays consistent and the two call sites cannot
drift apart.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: c3786385-03f1-48a3-8790-0237728b1cb4
⛔ Files ignored due to path filters (1)
harness/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (23)
console/web/src/components/chat/Message.tsxconsole/web/src/hooks/use-conversations.tsconsole/web/src/lib/sessions/entry-mapper.test.tsconsole/web/src/lib/sessions/entry-mapper.tsconsole/web/src/lib/sessions/types.tsconsole/web/src/types/chat.tsharness/prompts/anthropic.txtharness/prompts/gpt.txtharness/prompts/kimi.txtharness/src/configuration.rsharness/src/deferred.rsharness/src/deps.rsharness/src/functions/function_trigger.rsharness/src/functions/mod.rsharness/src/functions/on_session_deleted.rsharness/src/functions/send.rsharness/src/functions/subscribe.rsharness/src/lib.rsharness/src/main.rsharness/src/subscriptions/mod.rsharness/src/subscriptions/notify_agent.rsharness/src/subscriptions/registry.rsharness/src/turn_loop.rs
✅ Files skipped from review due to trivial changes (2)
- harness/src/lib.rs
- harness/prompts/anthropic.txt
🚧 Files skipped from review as they are similar to previous changes (1)
- harness/src/deps.rs
Comment on lines
+223
to
+236
| 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"); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Unregister failures silently leak engine-side triggers with no reconciliation.
unregister_engine_trigger is the sole cleanup path for explicit unsubscribe, session-deletion cleanup, and post-fire teardown of once subscriptions (per the on_session_deleted.rs/notify_agent.rs context snippets). On any transient failure it only logs a warning — there's no retry, backoff, or sweep to reconcile. Once the harness-side registry entry is gone (explicit unregister, or once fire), a failed engine-side unregister leaves the underlying trigger bound forever: for once triggers, every future fire becomes a silent no-op (claim_fire finds nothing) that nonetheless keeps invoking notify_agent indefinitely; for recurring/explicit-unregister cases, this is an unbounded resource leak on the engine side.
Given the codebase already has a sweep_pending-style cron reconciliation pattern for other pending state, consider applying the same idea here (e.g., record failed unregisters and periodically retry), or at minimum escalate the log to error! with a metric so operators can detect drift.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/functions/subscribe.rs` around lines 223 - 236, The cleanup path
in unregister_engine_trigger currently only emits a warning on trigger
unregister failure, which can leave engine-side triggers orphaned with no
reconciliation. Update unregister_engine_trigger to either persist failed
unregisters for later retry via a sweep/backoff mechanism like the existing
sweep_pending pattern, or at minimum escalate the tracing call to error and add
a metric so failures are observable; use the existing unregister_engine_trigger
and TriggerRequest flow as the entry point for the fix.
Comment on lines
+164
to
+179
| pub fn take_session(&self, session_id: &str) -> Vec<(String, Option<String>)> { | ||
| let ids: Vec<String> = { | ||
| 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() | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Make session cleanup atomic.
Line 165 snapshots the session’s IDs, then Lines 173-177 remove them after releasing the registry lock. A concurrent try_insert for the same session can land after the snapshot and survive session::deleted cleanup.
Proposed fix
pub fn take_session(&self, session_id: &str) -> Vec<(String, Option<String>)> {
- let ids: Vec<String> = {
- 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()
+ let mut inner = self.lock();
+ let Some(ids) = inner.by_session.remove(session_id) else {
+ return Vec::new();
+ };
+
+ ids.into_iter()
+ .filter_map(|id| {
+ let entry = inner.by_id.remove(&id)?;
+ Some((id, trigger_id(&entry)))
+ })
+ .collect()
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/subscriptions/registry.rs` around lines 164 - 179, Make
take_session atomic with respect to the registry state: it currently snapshots
ids from by_session under lock and then calls take after releasing the lock,
which allows a concurrent try_insert to sneak in and survive cleanup. Update
Registry::take_session to hold the lock across both collecting and removing the
session’s subscriptions, or otherwise remove entries from by_session and the
related maps in one locked operation using the existing Registry::lock, take,
and by_session members.
2 tasks
This was referenced Jul 2, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why
Agents need a way to wait for "something happened" without polling in a loop. iii already has the right primitive for this: engine triggers. This PR lets an agent subscribe to an existing trigger type and receive a session notification when it fires.
The important constraint is that the caller that fires the trigger often does not know the agent session. The harness owns that mapping: when the agent registers a trigger, the harness binds the trusted
session_idto the subscription and uses it later when the trigger fires.What
This adds agent trigger subscriptions through the existing engine API:
engine::register_trigger { trigger_type, config, label?, once? }subscription_idplus the effectiveonceflagharness::notify_agenthandlersubscription_id,session_id,label,once) that the engine stores on theTriggeroncedefaults to true for one-shot-ish types (state/stream/ custom trigger types) and false for recurringcronengine::unregister_trigger { id }idas the returnedsubscription_idExample flow:
No harness-owned emit/bus is added. Emitting is whatever already produces the trigger, such as
state::setfor an ad-hoc state signal.How it works
functions::subscribe::invoke, soengine::register_triggerandengine::unregister_triggercan be intercepted consistently from the inline turn loop,harness::function::trigger, and deferred approval release paths.session_id. The harness injects the owning session from the current turn/request context and stamps it into the trigger's registration metadata — the agent-supplied arguments can never widen the target function or the metadata.Triggerand delivers it toharness::notify_agentat fire time as a distinct invocation argument, separate from the fired payload — so the payload can never spoof the subscription id or session.harness::notify_agentfunction. On fire, the handler validates the delivered metadata, claims the fire against the local registry (liveness, ownership, and an idempotent notification entry id), tears downoncesubscriptions, and injects auser-role notification into the owning session — waking an idle session or steering a running turn, reusing the session's last turn options.session::deletedcleanup), a fire counter for idempotent entry ids, and the engine-returned trigger id for teardown.onceis true, onsession::deleted, or by the engine's worker-disconnect GC on process exit. A per-session cap (64) limits active subscriptions.{ notification: true }origin so the chat UI renders them distinctly (bell indicator) instead of as ordinary user text.Test plan
cargo testcargo clippy --all-targets -- -D warningspnpm typecheckgit diff --checkNotes / follow-ups
engine::register_trigger/engine::unregister_triggerinstead of addingharness::subscribe/harness::unsubscribewrappers.harness::*trigger types so agents cannot bind harness internals or create self-notification loops.Summary by CodeRabbit
New Features
Bug Fixes