fix(infinity-agent-core): return tool failures to the agent - #88
Merged
Conversation
* Add a shared `send_tool_error` helper that enqueues an error `ToolResult` with the original tool and call IDs * Convert thread-tool argument and relationship validation failures, including missing `close_thread.thread_id`, from propagated errors or duplicated message construction into queued tool results * Make the batch processor enqueue a generic `Error: Tool call failed` result when asynchronous tool execution returns an error, while retaining the detailed display event and logging delivery failures * Add regressions covering missing `close_thread` arguments and generic failed-tool fallback delivery Co-authored-by: Infinity 🤖 <infinity@hydro.run> PR: #88
shadaj
force-pushed
the
sandbox-86595edd-0387-40e8-aa40-3f1862988c82
branch
from
July 29, 2026 00:06
65a0f98 to
127f90b
Compare
Deploying infinity with
|
| Latest commit: |
127f90b
|
| Status: | ✅ Deploy successful! |
| Preview URL: | https://5088ef94.infinity-dc7.pages.dev |
| Branch Preview URL: | https://sandbox-86595edd-0387-40e8-a.infinity-dc7.pages.dev |
shadaj
marked this pull request as ready for review
July 29, 2026 21:37
MingweiSamuel
approved these changes
Aug 5, 2026
shadaj
added a commit
that referenced
this pull request
Aug 11, 2026
…on + lambda onto it Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for running agents, rebuild both production embeddings on top of it, unify the parallel implementations that had grown around the old architecture, and rewrite the `infinity-runtime` docs around the new API. Core (`infinity_agent_core::system`): - `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem` for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem` with an internal in-process queue (`ChannelSender`). - Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the batch may span multiple threads (SQS FIFO batches can interleave message groups); `step` partitions by `group_id`, applies the deferral policy per thread, joins the per-thread slices concurrently (prepare → completion → history sync → observer commit barrier → tool dispatch), and returns each thread's `StepOutcome`. `&mut self` serializes calls per system; the internal `Thread` type is not public and `AgentSystem` is not `Clone`, so two live in-memory views of one thread cannot exist through the public API. Per-thread configuration resolves lazily on first step so loading a thread never boots tool servers. - `LocalAgentSystem::start` runs an actor-system-style runtime: a router plus one driver per active thread, with input batching, `<interrupt>` handling, deferral of synthetic events while a non-passive tool call is pending, auto-compaction above 75% of the context window, and race-free idle exit/respawn. `RunningSystem` exposes senders, ack'd `subscribe`, `active_threads`, a `thread_exits` channel for per-conversation resource teardown, and `begin_shutdown` for process exit. - `ThreadHandle` (`start_with_handles()` + `thread_handle(id)`): attach to a thread (even before it exists), get `replay()` plus an exactly-once `events` stream, and send input; subscriptions survive driver respawns via a shared registry. - `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative to `on_subscribe` replays) plus awaited durability hooks (`on_user_choice_required` /`_dismissed`, `on_commit` before tool dispatch). `EventCollector` buffers `(thread_id, event)` pairs. - `ModelSource` (per-round resolution; enables mid-session model switching), `ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`), `Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`). - **`batch_processor` deleted**: `process_batch` had no production callers and its semantics violated the documented durability barrier (swallowed sync errors, then dispatched; emitted ResponseDone before sync). The documented low-level API is now `HistoryManager` + `prepare_input`/`run_completion`/`execute_action`; `execute_action_with_error_result` (the #88 error-fallback) lives in `event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its terminal-rendering enum. All unique regression coverage ported. - **Shared slice internals**: `event_processor::input_echo` is the single computation of accepted-input echoes (was duplicated between the system path and the old batch path); `SyntheticKind::is_compaction_complete()`; `InputMessage::user_text()`; driver's in-flight step modeled as `InFlightStep { fut, cancel_tx }` with a consuming `cancel()`. - **`rap_callback` module**: the RAP callback → `InputMessage` conversion (multimodal content, display segments, tagged synthetics incl. subscription final/associative flags, user_choice, oauth) now lives in core with 11 unit tests; the daemon delegates to it. - In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a serializable snapshot/restore API; the daemon's stores are now thin wrappers (extras + JSON persistence) so the subtle ancestor/compaction/dedup semantics exist in exactly one place. `spawn_thread_with_id` supports deterministic IDs. - Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder (Lambda sleep tools carry it as a field). - All 20 daemon agent-loop tests migrated into core `system/tests.rs` with byte-identical snapshots, plus new coverage: `ThreadHandle` streaming across tool-call rounds (CompletionFinished is per round; consumers keep pulling while a call is pending), multi-group step batches, failed-dispatch fallback, OAuth surfacing, prepare-level user-choice, handle respawn survival and pruning. Daemon (`infinity-daemon`): - One daemon-lifetime agent system replaces per-session agent loops; sessions are root threads; a `thread_exits` watcher (via the single `session_has_active_threads` predicate) marks sessions idle and shuts down their RAP servers when no keep-alive client is attached. - `rap_servers.rs`: `ManagedRapServer` boots lazily per session via `ThreadConfigSource` and reboots transparently after idle shutdown; `Up` state caches the manifest (fetch failure tears the boot down — no half-up state); MCP proxy tasks are owned (stdio children `kill_on_drop`) but proxies stay up across idles since MCP servers may be stateful; migration flows ride the same `collect_server_specs` + `ensure_up` path (`MigrationServer`; `boot_rap_servers`/`BootedRapServers`/`rap_tools.rs` deleted); RAP config reading/merging unified in `config::load_merged_rap_config`. - `DaemonObserver` implements the observer hooks; subscriber broadcasting unified in `observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update` never pruning dead subscribers); re-sent Connects replace rather than stack subscriptions; replay rendering now matches live rendering (strips `<interrupt>` prefixes, uses `rap_protocol::build_display_segments`). - `CatalogModelSource` resolves each thread's persisted model per round. Fix the pre-existing double `ModelSwitched` delivery (`switch_model` tracks whether the broadcast reached the requester via `same_channel`). - `send_input` takes `user_driven: bool`; `SessionManager` methods that only touch interior-mutable state take `&self`. - Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens captured an eager-boot layout artifact). - Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`, `InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`. Lambda (`infinity-agent-lambda`): - Event handler rewritten on the new API: one system per invocation with a `ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest cache) plus platform sleep tools; one batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`. Sleep tools share a `WakeupScheduler` (SQS delay / EventBridge dispatch deduplicated). - **New `rap-receiver` binary replaces `agent/lib/infinity-agents/rap-receiver/ index.mjs`** (deleted): same Function-URL contract, but full parity with the daemon's conversion via `infinity_agent_core::rap_callback` — user_choice callbacks accepted (JS returned 400), multimodal content and `display_as` preserved, subscription `final` flag and tagged synthetics honored, and stable dedup IDs for results/oauth/choices (per-delivery IDs only for subscription events, which share a tool_call_id). CDK switches the receiver to a cargo-lambda `RustFunction`. Deploy note: the Function URL is recreated, so persisted callback URLs from long-lived subscriptions must be re-subscribed. Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two sections rewritten in a usage-oriented style (example-first walkthroughs rather than API inventories): "The Agent System API" (overview, building-a-system, running-locally, step-mode, observers) and "The Low-Level API" (overview, history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc` is warning-free. BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`, `process_input_item`, `DisplayEvent`); `Thread` is no longer public (use `AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`; `EventCollector::take` returns `(thread_id, event)` pairs; builder tools are stored as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`; `ToolContext`/builder lose `input_queue_arn`. `infinity-daemon` no longer exports `sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`, `Session`, `InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`; `SessionManager::switch_model` takes the requester's sender and returns `Result<(), String>`; `send_input` takes `user_driven: bool`; `SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` now lives in `infinity_agent_cli::display`. Co-authored-by: Infinity 🤖 <infinity@hydro.run> PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 11, 2026
…on + lambda onto it Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for running agents, rebuild both production embeddings on top of it, unify the parallel implementations that had grown around the old architecture, and rewrite the `infinity-runtime` docs around the new API. Core (`infinity_agent_core::system`): - `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem` for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem` with an internal in-process queue (`ChannelSender`). - Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the batch may span multiple threads (SQS FIFO batches can interleave message groups); `step` partitions by `group_id`, applies the deferral policy per thread, joins the per-thread slices concurrently (prepare → completion → history sync → observer commit barrier → tool dispatch), and returns each thread's `StepOutcome`. `&mut self` serializes calls per system; the internal `Thread` type is not public and `AgentSystem` is not `Clone`, so two live in-memory views of one thread cannot exist through the public API. Per-thread configuration resolves lazily on first step so loading a thread never boots tool servers. - `LocalAgentSystem::start` runs an actor-system-style runtime: a router plus one driver per active thread, with input batching, `<interrupt>` handling, deferral of synthetic events while a non-passive tool call is pending, auto-compaction above 75% of the context window, and race-free idle exit/respawn. `RunningSystem` exposes senders, ack'd `subscribe`, `active_threads`, a `thread_exits` channel for per-conversation resource teardown, and `begin_shutdown` for process exit. - `ThreadHandle` (`start_with_handles()` + `thread_handle(id)`): attach to a thread (even before it exists), get `replay()` plus an exactly-once `events` stream, and send input; subscriptions survive driver respawns via a shared registry. - `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative to `on_subscribe` replays) plus awaited durability hooks (`on_user_choice_required` /`_dismissed`, `on_commit` before tool dispatch). `EventCollector` buffers `(thread_id, event)` pairs. - `ModelSource` (per-round resolution; enables mid-session model switching), `ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`), `Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`). - **`batch_processor` deleted**: `process_batch` had no production callers and its semantics violated the documented durability barrier (swallowed sync errors, then dispatched; emitted ResponseDone before sync). The documented low-level API is now `HistoryManager` + `prepare_input`/`run_completion`/`execute_action`; `execute_action_with_error_result` (the #88 error-fallback) lives in `event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its terminal-rendering enum. All unique regression coverage ported. - **Shared slice internals**: `event_processor::input_echo` is the single computation of accepted-input echoes (was duplicated between the system path and the old batch path); `SyntheticKind::is_compaction_complete()`; `InputMessage::user_text()`; driver's in-flight step modeled as `InFlightStep { fut, cancel_tx }` with a consuming `cancel()`. - **`rap_callback` module**: the RAP callback → `InputMessage` conversion (multimodal content, display segments, tagged synthetics incl. subscription final/associative flags, user_choice, oauth) now lives in core with 11 unit tests; the daemon delegates to it. - In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a serializable snapshot/restore API; the daemon's stores are now thin wrappers (extras + JSON persistence) so the subtle ancestor/compaction/dedup semantics exist in exactly one place. `spawn_thread_with_id` supports deterministic IDs. - Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder (Lambda sleep tools carry it as a field). - All 20 daemon agent-loop tests migrated into core `system/tests.rs` with byte-identical snapshots, plus new coverage: `ThreadHandle` streaming across tool-call rounds (CompletionFinished is per round; consumers keep pulling while a call is pending), multi-group step batches, failed-dispatch fallback, OAuth surfacing, prepare-level user-choice, handle respawn survival and pruning. Daemon (`infinity-daemon`): - One daemon-lifetime agent system replaces per-session agent loops; sessions are root threads; a `thread_exits` watcher (via the single `session_has_active_threads` predicate) marks sessions idle and shuts down their RAP servers when no keep-alive client is attached. - `rap_servers.rs`: `ManagedRapServer` boots lazily per session via `ThreadConfigSource` and reboots transparently after idle shutdown; `Up` state caches the manifest (fetch failure tears the boot down — no half-up state); MCP proxy tasks are owned (stdio children `kill_on_drop`) but proxies stay up across idles since MCP servers may be stateful; migration flows ride the same `collect_server_specs` + `ensure_up` path (`MigrationServer`; `boot_rap_servers`/`BootedRapServers`/`rap_tools.rs` deleted); RAP config reading/merging unified in `config::load_merged_rap_config`. - `DaemonObserver` implements the observer hooks; subscriber broadcasting unified in `observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update` never pruning dead subscribers); re-sent Connects replace rather than stack subscriptions; replay rendering now matches live rendering (strips `<interrupt>` prefixes, uses `rap_protocol::build_display_segments`). - `CatalogModelSource` resolves each thread's persisted model per round. Fix the pre-existing double `ModelSwitched` delivery (`switch_model` tracks whether the broadcast reached the requester via `same_channel`). - `send_input` takes `user_driven: bool`; `SessionManager` methods that only touch interior-mutable state take `&self`. - Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens captured an eager-boot layout artifact). - Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`, `InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`. Lambda (`infinity-agent-lambda`): - Event handler rewritten on the new API: one system per invocation with a `ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest cache) plus platform sleep tools; one batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`. Sleep tools share a `WakeupScheduler` (SQS delay / EventBridge dispatch deduplicated). - **New `rap-receiver` binary replaces `agent/lib/infinity-agents/rap-receiver/ index.mjs`** (deleted): same Function-URL contract, but full parity with the daemon's conversion via `infinity_agent_core::rap_callback` — user_choice callbacks accepted (JS returned 400), multimodal content and `display_as` preserved, subscription `final` flag and tagged synthetics honored, and stable dedup IDs for results/oauth/choices (per-delivery IDs only for subscription events, which share a tool_call_id). CDK switches the receiver to a cargo-lambda `RustFunction`. Deploy note: the Function URL is recreated, so persisted callback URLs from long-lived subscriptions must be re-subscribed. Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two sections rewritten in a usage-oriented style (example-first walkthroughs rather than API inventories): "The Agent System API" (overview, building-a-system, running-locally, step-mode, observers) and "The Low-Level API" (overview, history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc` is warning-free. BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`, `process_input_item`, `DisplayEvent`); `Thread` is no longer public (use `AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`; `EventCollector::take` returns `(thread_id, event)` pairs; builder tools are stored as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`; `ToolContext`/builder lose `input_queue_arn`. `infinity-daemon` no longer exports `sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`, `Session`, `InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`; `SessionManager::switch_model` takes the requester's sender and returns `Result<(), String>`; `send_input` takes `user_driven: bool`; `SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` now lives in `infinity_agent_cli::display`. Co-authored-by: Infinity 🤖 <infinity@hydro.run> PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → observer
commit barrier → tool dispatch), and returns each thread's `StepOutcome`. `&mut
self` serializes calls per system; the internal `Thread` type is not public and
`AgentSystem` is not `Clone`, so two live in-memory views of one thread cannot
exist through the public API. Per-thread configuration resolves lazily on first
step so loading a thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools and
prompt, unioned onto the system-wide configuration (static or
`ThreadConfigSource`) via an internal `UnionConfigSource`; `thread_handle(id)`
re-attaches to existing threads only (launched in-process or with history).
Launch configurations are process-lifetime (not persisted).
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited durability hooks (`on_user_choice_required`
/`_dismissed`, `on_commit` before tool dispatch). `EventCollector` buffers
`(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt unioning and
existing-only attachment, `ThreadHandle` streaming across tool-call rounds,
multi-group step batches, failed-dispatch fallback, OAuth surfacing, prepare-level
user-choice, handle respawn survival and pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops; sessions are
root threads; a `thread_exits` watcher marks sessions idle and shuts down their
RAP servers when no keep-alive client is attached.
- `rap_servers.rs`: `ManagedRapServer` boots lazily per session via
`ThreadConfigSource` and reboots transparently after idle shutdown; `Up` caches
the manifest (fetch failure tears the boot down); MCP proxy tasks are owned
(stdio children `kill_on_drop`) but proxies stay up across idles since MCP servers
may be stateful; migration flows ride the same path (`MigrationServer`;
`boot_rap_servers`/`rap_tools.rs` deleted); config merging unified in
`config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); `EventCollector::take` returns `(thread_id, event)` pairs; builder tools
are stored as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`;
`ToolContext`/builder lose `input_queue_arn`. `infinity-daemon` no longer exports
`sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`, `Session`,
`InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → observer
commit barrier → tool dispatch), and returns each thread's `StepOutcome`. `&mut
self` serializes calls per system; the internal `Thread` type is not public and
`AgentSystem` is not `Clone`, so two live in-memory views of one thread cannot
exist through the public API. Per-thread configuration resolves lazily on first
step so loading a thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools and
prompt, unioned onto the system-wide configuration (static or
`ThreadConfigSource`) via an internal `UnionConfigSource`; `thread_handle(id)`
re-attaches to existing threads only (launched in-process or with history).
Launch configurations are process-lifetime (not persisted).
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited durability hooks (`on_user_choice_required`
/`_dismissed`, `on_commit` before tool dispatch). `EventCollector` buffers
`(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt unioning and
existing-only attachment, `ThreadHandle` streaming across tool-call rounds,
multi-group step batches, failed-dispatch fallback, OAuth surfacing, prepare-level
user-choice, handle respawn survival and pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops; sessions are
root threads; a `thread_exits` watcher marks sessions idle and shuts down their
RAP servers when no keep-alive client is attached.
- `rap_servers.rs`: `ManagedRapServer` boots lazily per session via
`ThreadConfigSource` and reboots transparently after idle shutdown; `Up` caches
the manifest (fetch failure tears the boot down); MCP proxy tasks are owned
(stdio children `kill_on_drop`) but proxies stay up across idles since MCP servers
may be stateful; migration flows ride the same path (`MigrationServer`;
`boot_rap_servers`/`rap_tools.rs` deleted); config merging unified in
`config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); `EventCollector::take` returns `(thread_id, event)` pairs; builder tools
are stored as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`;
`ToolContext`/builder lose `input_queue_arn`. `infinity-daemon` no longer exports
`sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`, `Session`,
`InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → observer
commit barrier → tool dispatch), and returns each thread's `StepOutcome`. `&mut
self` serializes calls per system; the internal `Thread` type is not public and
`AgentSystem` is not `Clone`, so two live in-memory views of one thread cannot
exist through the public API. Per-thread configuration resolves lazily on first
step so loading a thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools and
prompt, unioned onto the system-wide configuration (static or
`ThreadConfigSource`) via an internal `UnionConfigSource`; `thread_handle(id)`
re-attaches to existing threads only (launched in-process or with history).
Launch configurations are process-lifetime (not persisted).
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited durability hooks (`on_user_choice_required`
/`_dismissed`, `on_commit` before tool dispatch). `EventCollector` buffers
`(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt unioning and
existing-only attachment, `ThreadHandle` streaming across tool-call rounds,
multi-group step batches, failed-dispatch fallback, OAuth surfacing, prepare-level
user-choice, handle respawn survival and pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops; sessions are
root threads; a `thread_exits` watcher marks sessions idle and shuts down their
RAP servers when no keep-alive client is attached.
- `rap_servers.rs`: `ManagedRapServer` boots lazily per session via
`ThreadConfigSource` and reboots transparently after idle shutdown; `Up` caches
the manifest (fetch failure tears the boot down); MCP proxy tasks are owned
(stdio children `kill_on_drop`) but proxies stay up across idles since MCP servers
may be stateful; migration flows ride the same path (`MigrationServer`;
`boot_rap_servers`/`rap_tools.rs` deleted); config merging unified in
`config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); `EventCollector::take` returns `(thread_id, event)` pairs; builder tools
are stored as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`;
`ToolContext`/builder lose `input_queue_arn`. `infinity-daemon` no longer exports
`sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`, `Session`,
`InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`, resident runtime under `system::local`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → observer
commit barrier → tool dispatch), and returns each thread's `StepOutcome`. `&mut
self` serializes calls per system; the internal `Thread` type is not public and
`AgentSystem` is not `Clone`, so two live in-memory views of one thread cannot
exist through the public API. Per-thread configuration resolves lazily on first
step so loading a thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools and
prompt, unioned onto the system-wide configuration (static or
`ThreadConfigSource`); `thread_handle(id)` re-attaches to existing threads only.
Launch configurations are process-lifetime (not persisted).
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited durability hooks (`on_user_choice_required`
/`_dismissed`, `on_commit` before tool dispatch). `EventCollector` buffers
`(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt unioning and
existing-only attachment, `ThreadHandle` streaming across tool-call rounds,
multi-group step batches, failed-dispatch fallback, OAuth surfacing, prepare-level
user-choice, handle respawn survival and pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops; sessions are
root threads; a `thread_exits` watcher marks sessions idle and shuts down their
RAP servers when no keep-alive client is attached.
- `rap_servers.rs`: `ManagedRapServer` boots lazily per session via
`ThreadConfigSource` and reboots transparently after idle shutdown; `Up` caches
the manifest (fetch failure tears the boot down); MCP proxy tasks are owned
(stdio children `kill_on_drop`) but proxies stay up across idles since MCP servers
may be stateful; migration flows ride the same path (`MigrationServer`;
`boot_rap_servers`/`rap_tools.rs` deleted); config merging unified in
`config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); the resident-runtime types (`RunningSystem`, `SubscribeHandle`,
`ThreadHandle`, `HandleObserver`, `HandleSubscribeRequest`, `LaunchingSystem`,
`ThreadBuilder`, `ChannelSender`, `ChannelSendError`, `ActiveThreads`) live under
`system::local`; `EventCollector::take` returns `(thread_id, event)` pairs; builder
tools are stored as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`;
`ToolContext`/builder lose `input_queue_arn`. `infinity-daemon` no longer exports
`sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`, `Session`,
`InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`, resident runtime under `system::local`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → tool
dispatch), and returns each thread's `StepOutcome`. `&mut self` serializes calls
per system; the internal `Thread` type is not public and `AgentSystem` is not
`Clone`, so two live in-memory views of one thread cannot exist through the
public API. Per-thread configuration resolves lazily on first step so loading a
thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools and
prompt, unioned onto the system-wide configuration; launch configurations
attach to the launched (root) thread, so spawned subagent threads inherit them
via root resolution. `thread_handle(id)` re-attaches to existing threads only.
Launch configurations are process-lifetime (not persisted).
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited user-choice durability hooks
(`on_user_choice_required`/`_dismissed`). Tool dispatch happens strictly after
history sync, so turns are durable before their effects are observable.
`EventCollector` buffers `(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
Unused refactor leftovers pruned: `ToolSet`/`VecToolSet`,
`ToolsConfig::{add_command, toolset_commands, mcp_servers, http_mcp_servers}`,
`HistoryManager::state_store()`.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt unioning,
launched-thread child inheritance, existing-only attachment, `ThreadHandle`
streaming across tool-call rounds, multi-group step batches, failed-dispatch
fallback, OAuth surfacing, prepare-level user-choice, handle respawn survival and
pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops and the
restart-on-send path (`needs_restart` → `start_session`): idle threads respawn a
driver transparently instead. Sessions are root threads; a `thread_exits` watcher
marks sessions idle and shuts down their RAP servers when no keep-alive client is
attached.
- `rap_servers.rs`: RAP server lifecycle moves out of session start — boot happens
lazily when a thread's config first resolves (loading/replaying a thread never
boots servers), shutdown is driven by session idleness, and a shut-down server
reboots on the next tool interaction. `Up` caches the manifest (fetch failure
tears the boot down); MCP proxy tasks are owned (stdio children `kill_on_drop`)
but proxies stay up across idles since MCP servers may be stateful; migration
flows ride the same path (`MigrationServer`; `boot_rap_servers`/`rap_tools.rs`
deleted); config merging unified in `config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); the resident-runtime types (`RunningSystem`, `SubscribeHandle`,
`ThreadHandle`, `HandleObserver`, `HandleSubscribeRequest`, `LaunchingSystem`,
`ThreadBuilder`, `ChannelSender`, `ChannelSendError`, `ActiveThreads`) live under
`system::local`; `ThreadObserver` has no `on_commit` (persist in `on_event` or
externally; the turn is already durable before dispatch); `EventCollector::take`
returns `(thread_id, event)` pairs; builder tools are stored as `Rc<dyn Tool<M>>`;
`Tool` gains defaulted `is_passive`; `ToolContext`/builder lose `input_queue_arn`;
`ToolSet`/`VecToolSet` and `ToolsConfig::{add_command, toolset_commands,
mcp_servers, http_mcp_servers}` are removed. `infinity-daemon` no longer exports
`sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`, `Session`,
`InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`, resident runtime under `system::local`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → tool
dispatch), and returns each thread's `StepOutcome`. `&mut self` serializes calls
per system; the internal `Thread` type is not public and `AgentSystem` is not
`Clone`, so two live in-memory views of one thread cannot exist through the
public API. Per-thread configuration resolves lazily on first step so loading a
thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools and
prompt, unioned onto the system-wide configuration; launch configurations
attach to the launched (root) thread, so spawned subagent threads inherit them
via root resolution. `thread_handle(id)` re-attaches to existing threads only.
Launch configurations are process-lifetime (not persisted).
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited user-choice durability hooks
(`on_user_choice_required`/`_dismissed`). Tool dispatch happens strictly after
history sync, so turns are durable before their effects are observable.
`EventCollector` buffers `(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
Unused refactor leftovers pruned: `ToolSet`/`VecToolSet`,
`ToolsConfig::{add_command, toolset_commands, mcp_servers, http_mcp_servers}`,
`HistoryManager::state_store()`.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt unioning,
launched-thread child inheritance, existing-only attachment, `ThreadHandle`
streaming across tool-call rounds, multi-group step batches, failed-dispatch
fallback, OAuth surfacing, prepare-level user-choice, handle respawn survival and
pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops and the
restart-on-send path (`needs_restart` → `start_session`): idle threads respawn a
driver transparently instead. Sessions are root threads; a `thread_exits` watcher
marks sessions idle and shuts down their RAP servers when no keep-alive client is
attached.
- `rap_servers.rs`: RAP server lifecycle moves out of session start — boot happens
lazily when a thread's config first resolves (loading/replaying a thread never
boots servers), shutdown is driven by session idleness, and a shut-down server
reboots on the next tool interaction. `Up` caches the manifest (fetch failure
tears the boot down); MCP proxy tasks are owned (stdio children `kill_on_drop`)
but proxies stay up across idles since MCP servers may be stateful; migration
flows ride the same path (`MigrationServer`; `boot_rap_servers`/`rap_tools.rs`
deleted); config merging unified in `config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); the resident-runtime types (`RunningSystem`, `SubscribeHandle`,
`ThreadHandle`, `HandleObserver`, `HandleSubscribeRequest`, `LaunchingSystem`,
`ThreadBuilder`, `ChannelSender`, `ChannelSendError`, `ActiveThreads`) live under
`system::local`; `ThreadObserver` has no `on_commit` (persist in `on_event` or
externally; the turn is already durable before dispatch); `EventCollector::take`
returns `(thread_id, event)` pairs; builder tools are stored as `Rc<dyn Tool<M>>`;
`Tool` gains defaulted `is_passive`; `ToolContext`/builder lose `input_queue_arn`;
`ToolSet`/`VecToolSet` and `ToolsConfig::{add_command, toolset_commands,
mcp_servers, http_mcp_servers}` are removed. `infinity-daemon` no longer exports
`sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`, `Session`,
`InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`, resident runtime under `system::local`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → tool
dispatch), and returns each thread's `StepOutcome`. `&mut self` serializes calls
per system; the internal `Thread` type is not public and `AgentSystem` is not
`Clone`, so two live in-memory views of one thread cannot exist through the
public API. Per-thread configuration resolves lazily on first step so loading a
thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools,
prompt, and optionally their own `ModelSource`, unioned onto the system-wide
configuration; launch configurations attach to the launched (root) thread, so
spawned subagent threads inherit tools, prompt, and model via root resolution.
`thread_handle(id)` re-attaches to existing threads only (`None` means the
thread does not exist). Launch configurations are process-lifetime.
- Shutdown is ownership-based: `RunningSystem::shutdown(self)` (and
`LaunchingSystem::shutdown(self)`) cancels every driver, flushes in-flight turns,
and awaits the router. Because shutting down consumes the system, `&self` methods
are infallible: `thread_handle`/`launch` return handles directly, `subscribe`
resolves once installed, and `send`/`send_user_text` return `()`. Independent
handles that can outlive the system (`SubscribeHandle`, `ThreadHandle`) keep
fallible signatures.
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited user-choice durability hooks
(`on_user_choice_required`/`_dismissed`). Tool dispatch happens strictly after
history sync, so turns are durable before their effects are observable.
`EventCollector` buffers `(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
Unused refactor leftovers pruned: `ToolSet`/`VecToolSet`,
`ToolsConfig::{add_command, toolset_commands, mcp_servers, http_mcp_servers}`,
`HistoryManager::state_store()`.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt/model unioning
and child inheritance, existing-only attachment, `ThreadHandle` streaming across
tool-call rounds, multi-group step batches, failed-dispatch fallback, OAuth
surfacing, prepare-level user-choice, handle respawn survival and pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops and the
restart-on-send path (`needs_restart` → `start_session`): idle threads respawn a
driver transparently instead. Sessions are root threads; a `thread_exits` watcher
marks sessions idle and shuts down their RAP servers when no keep-alive client is
attached.
- `rap_servers.rs`: RAP server lifecycle moves out of session start — boot happens
lazily when a thread's config first resolves (loading/replaying a thread never
boots servers), shutdown is driven by session idleness, and a shut-down server
reboots on the next tool interaction. `Up` caches the manifest (fetch failure
tears the boot down); MCP proxy tasks are owned (stdio children `kill_on_drop`)
but proxies stay up across idles since MCP servers may be stateful; migration
flows ride the same path (`MigrationServer`; `boot_rap_servers`/`rap_tools.rs`
deleted); config merging unified in `config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); the resident-runtime types (`RunningSystem`, `SubscribeHandle`,
`ThreadHandle`, `HandleObserver`, `HandleSubscribeRequest`, `LaunchingSystem`,
`ThreadBuilder`, `ChannelSender`, `ChannelSendError`, `ActiveThreads`) live under
`system::local`; `RunningSystem::begin_shutdown`/`task` are replaced by owned
`shutdown(self)`, and `RunningSystem::{send, send_user_text, subscribe,
thread_handle}` are infallible; `ThreadObserver` has no `on_commit` (persist in
`on_event` or externally; the turn is already durable before dispatch);
`EventCollector::take` returns `(thread_id, event)` pairs; builder tools are stored
as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`; `ToolContext`/builder
lose `input_queue_arn`; `ToolSet`/`VecToolSet` and `ToolsConfig::{add_command,
toolset_commands, mcp_servers, http_mcp_servers}` are removed. `infinity-daemon` no
longer exports `sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`,
`Session`, `InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`, resident runtime under `system::local`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → tool
dispatch), and returns each thread's `StepOutcome`. `&mut self` serializes calls
per system; the internal `Thread` type is not public and `AgentSystem` is not
`Clone`, so two live in-memory views of one thread cannot exist through the
public API. Per-thread configuration resolves lazily on first step so loading a
thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools,
prompt, and optionally their own `ModelSource`, unioned onto the system-wide
configuration; launch configurations attach to the launched (root) thread, so
spawned subagent threads inherit tools, prompt, and model via root resolution.
`thread_handle(id)` re-attaches to existing threads only (`None` means the
thread does not exist). Launch configurations are process-lifetime.
- Shutdown is ownership-based: `RunningSystem::shutdown(self)` (and
`LaunchingSystem::shutdown(self)`) cancels every driver, flushes in-flight turns,
and awaits the router. Because shutting down consumes the system, `&self` methods
are infallible: `thread_handle`/`launch` return handles directly, `subscribe`
resolves once installed, and `send`/`send_user_text` return `()`. Independent
handles that can outlive the system (`SubscribeHandle`, `ThreadHandle`) keep
fallible signatures.
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited user-choice durability hooks
(`on_user_choice_required`/`_dismissed`). Tool dispatch happens strictly after
history sync, so turns are durable before their effects are observable.
`EventCollector` buffers `(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
Unused refactor leftovers pruned: `ToolSet`/`VecToolSet`,
`ToolsConfig::{add_command, toolset_commands, mcp_servers, http_mcp_servers}`,
`HistoryManager::state_store()`.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt/model unioning
and child inheritance, existing-only attachment, `ThreadHandle` streaming across
tool-call rounds, multi-group step batches, failed-dispatch fallback, OAuth
surfacing, prepare-level user-choice, handle respawn survival and pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops and the
restart-on-send path (`needs_restart` → `start_session`): idle threads respawn a
driver transparently instead. Sessions are root threads; a `thread_exits` watcher
marks sessions idle and shuts down their RAP servers when no keep-alive client is
attached.
- `rap_servers.rs`: RAP server lifecycle moves out of session start — boot happens
lazily when a thread's config first resolves (loading/replaying a thread never
boots servers), shutdown is driven by session idleness, and a shut-down server
reboots on the next tool interaction. `Up` caches the manifest (fetch failure
tears the boot down); MCP proxy tasks are owned (stdio children `kill_on_drop`)
but proxies stay up across idles since MCP servers may be stateful; migration
flows ride the same path (`MigrationServer`; `boot_rap_servers`/`rap_tools.rs`
deleted); config merging unified in `config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); the resident-runtime types (`RunningSystem`, `SubscribeHandle`,
`ThreadHandle`, `HandleObserver`, `HandleSubscribeRequest`, `LaunchingSystem`,
`ThreadBuilder`, `ChannelSender`, `ChannelSendError`, `ActiveThreads`) live under
`system::local`; `RunningSystem::begin_shutdown`/`task` are replaced by owned
`shutdown(self)`, and `RunningSystem::{send, send_user_text, subscribe,
thread_handle}` are infallible; `ThreadObserver` has no `on_commit` (persist in
`on_event` or externally; the turn is already durable before dispatch);
`EventCollector::take` returns `(thread_id, event)` pairs; builder tools are stored
as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`; `ToolContext`/builder
lose `input_queue_arn`; `ToolSet`/`VecToolSet` and `ToolsConfig::{add_command,
toolset_commands, mcp_servers, http_mcp_servers}` are removed. `infinity-daemon` no
longer exports `sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`,
`Session`, `InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`, resident runtime under `system::local`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → tool
dispatch), and returns each thread's `StepOutcome`. `&mut self` serializes calls
per system; the internal `Thread` type is not public and `AgentSystem` is not
`Clone`, so two live in-memory views of one thread cannot exist through the
public API. Per-thread configuration resolves lazily on first step so loading a
thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools,
prompt, and optionally their own `ModelSource`, unioned onto the system-wide
configuration; launch configurations attach to the launched (root) thread, so
spawned subagent threads inherit tools, prompt, and model via root resolution.
`thread_handle(id)` re-attaches to existing threads only (`None` means the
thread does not exist). Launch configurations are process-lifetime.
- Shutdown is ownership-based: `RunningSystem::shutdown(self)` (and
`LaunchingSystem::shutdown(self)`) cancels every driver, flushes in-flight turns,
and awaits the router. Because shutting down consumes the system, `&self` methods
are infallible: `thread_handle`/`launch` return handles directly, `subscribe`
resolves once installed, and `send`/`send_user_text` return `()`. Independent
handles that can outlive the system (`SubscribeHandle`, `ThreadHandle`) keep
fallible signatures.
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited user-choice durability hooks
(`on_user_choice_required`/`_dismissed`). Tool dispatch happens strictly after
history sync, so turns are durable before their effects are observable.
`EventCollector` buffers `(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
Unused refactor leftovers pruned: `ToolSet`/`VecToolSet`,
`ToolsConfig::{add_command, toolset_commands, mcp_servers, http_mcp_servers}`,
`HistoryManager::state_store()`.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt/model unioning
and child inheritance, existing-only attachment, `ThreadHandle` streaming across
tool-call rounds, multi-group step batches, failed-dispatch fallback, OAuth
surfacing, prepare-level user-choice, handle respawn survival and pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops and the
restart-on-send path (`needs_restart` → `start_session`): idle threads respawn a
driver transparently instead. Sessions are root threads; a `thread_exits` watcher
marks sessions idle and shuts down their RAP servers when no keep-alive client is
attached.
- `rap_servers.rs`: RAP server lifecycle moves out of session start — boot happens
lazily when a thread's config first resolves (loading/replaying a thread never
boots servers), shutdown is driven by session idleness, and a shut-down server
reboots on the next tool interaction. `Up` caches the manifest (fetch failure
tears the boot down); MCP proxy tasks are owned (stdio children `kill_on_drop`)
but proxies stay up across idles since MCP servers may be stateful; migration
flows ride the same path (`MigrationServer`; `boot_rap_servers`/`rap_tools.rs`
deleted); config merging unified in `config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); the resident-runtime types (`RunningSystem`, `SubscribeHandle`,
`ThreadHandle`, `HandleObserver`, `HandleSubscribeRequest`, `LaunchingSystem`,
`ThreadBuilder`, `ChannelSender`, `ChannelSendError`, `ActiveThreads`) live under
`system::local`; `RunningSystem::begin_shutdown`/`task` are replaced by owned
`shutdown(self)`, and `RunningSystem::{send, send_user_text, subscribe,
thread_handle}` are infallible; `ThreadObserver` has no `on_commit` (persist in
`on_event` or externally; the turn is already durable before dispatch);
`EventCollector::take` returns `(thread_id, event)` pairs; builder tools are stored
as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`; `ToolContext`/builder
lose `input_queue_arn`; `ToolSet`/`VecToolSet` and `ToolsConfig::{add_command,
toolset_commands, mcp_servers, http_mcp_servers}` are removed. `infinity-daemon` no
longer exports `sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`,
`Session`, `InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`, resident runtime under `system::local`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → tool
dispatch), and returns each thread's `StepOutcome`. `&mut self` serializes calls
per system; the internal `Thread` type is not public and `AgentSystem` is not
`Clone`, so two live in-memory views of one thread cannot exist through the
public API. Per-thread configuration resolves lazily on first step so loading a
thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools,
prompt, and optionally their own `ModelSource`, unioned onto the system-wide
configuration; launch configurations attach to the launched (root) thread, so
spawned subagent threads inherit tools, prompt, and model via root resolution.
`thread_handle(id)` re-attaches to existing threads only (`None` means the
thread does not exist). Launch configurations are process-lifetime.
- Shutdown is ownership-based: `RunningSystem::shutdown(self)` (and
`LaunchingSystem::shutdown(self)`) cancels every driver, flushes in-flight turns,
and awaits the router. Because shutting down consumes the system, `&self` methods
are infallible: `thread_handle`/`launch` return handles directly, `subscribe`
resolves once installed, and `send`/`send_user_text` return `()`. Independent
handles that can outlive the system (`SubscribeHandle`, `ThreadHandle`) keep
fallible signatures.
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited user-choice durability hooks
(`on_user_choice_required`/`_dismissed`). Tool dispatch happens strictly after
history sync, so turns are durable before their effects are observable.
`EventCollector` buffers `(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
Unused refactor leftovers pruned: `ToolSet`/`VecToolSet`,
`ToolsConfig::{add_command, toolset_commands, mcp_servers, http_mcp_servers}`,
`HistoryManager::state_store()`.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt/model unioning
and child inheritance, existing-only attachment, `ThreadHandle` streaming across
tool-call rounds, multi-group step batches, failed-dispatch fallback, OAuth
surfacing, prepare-level user-choice, handle respawn survival and pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops and the
restart-on-send path (`needs_restart` → `start_session`): idle threads respawn a
driver transparently instead. Sessions are root threads; a `thread_exits` watcher
marks sessions idle and shuts down their RAP servers when no keep-alive client is
attached.
- `rap_servers.rs`: RAP server lifecycle moves out of session start — boot happens
lazily when a thread's config first resolves (loading/replaying a thread never
boots servers), shutdown is driven by session idleness, and a shut-down server
reboots on the next tool interaction. `Up` caches the manifest (fetch failure
tears the boot down); MCP proxy tasks are owned (stdio children `kill_on_drop`)
but proxies stay up across idles since MCP servers may be stateful; migration
flows ride the same path (`MigrationServer`; `boot_rap_servers`/`rap_tools.rs`
deleted); config merging unified in `config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); the resident-runtime types (`RunningSystem`, `SubscribeHandle`,
`ThreadHandle`, `HandleObserver`, `HandleSubscribeRequest`, `LaunchingSystem`,
`ThreadBuilder`, `ChannelSender`, `ChannelSendError`, `ActiveThreads`) live under
`system::local`; `RunningSystem::begin_shutdown`/`task` are replaced by owned
`shutdown(self)`, and `RunningSystem::{send, send_user_text, subscribe,
thread_handle}` are infallible; `ThreadObserver` has no `on_commit` (persist in
`on_event` or externally; the turn is already durable before dispatch);
`EventCollector::take` returns `(thread_id, event)` pairs; builder tools are stored
as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`; `ToolContext`/builder
lose `input_queue_arn`; `ToolSet`/`VecToolSet` and `ToolsConfig::{add_command,
toolset_commands, mcp_servers, http_mcp_servers}` are removed. `infinity-daemon` no
longer exports `sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`,
`Session`, `InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`, resident runtime under `system::local`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → tool
dispatch), and returns each thread's `StepOutcome`. `&mut self` serializes calls
per system; the internal `Thread` type is not public and `AgentSystem` is not
`Clone`, so two live in-memory views of one thread cannot exist through the
public API. Per-thread configuration resolves lazily on first step so loading a
thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools,
prompt, and optionally their own `ModelSource`, unioned onto the system-wide
configuration; launch configurations attach to the launched (root) thread, so
spawned subagent threads inherit tools, prompt, and model via root resolution.
`thread_handle(id)` re-attaches to existing threads only (`None` means the
thread does not exist). Launch configurations are process-lifetime.
- Shutdown is ownership-based: `RunningSystem::shutdown(self)` (and
`LaunchingSystem::shutdown(self)`) cancels every driver, flushes in-flight turns,
and awaits the router. Because shutting down consumes the system, `&self` methods
are infallible: `thread_handle`/`launch` return handles directly, `subscribe`
resolves once installed, and `send`/`send_user_text` return `()`. Independent
handles that can outlive the system (`SubscribeHandle`, `ThreadHandle`) keep
fallible signatures.
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited user-choice durability hooks
(`on_user_choice_required`/`_dismissed`). Tool dispatch happens strictly after
history sync, so turns are durable before their effects are observable.
`EventCollector` buffers `(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
Unused refactor leftovers pruned: `ToolSet`/`VecToolSet`,
`ToolsConfig::{add_command, toolset_commands, mcp_servers, http_mcp_servers}`,
`HistoryManager::state_store()`.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt/model unioning
and child inheritance, existing-only attachment, `ThreadHandle` streaming across
tool-call rounds, multi-group step batches, failed-dispatch fallback, OAuth
surfacing, prepare-level user-choice, handle respawn survival and pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops and the
restart-on-send path (`needs_restart` → `start_session`): idle threads respawn a
driver transparently instead. Sessions are root threads; a `thread_exits` watcher
marks sessions idle and shuts down their RAP servers when no keep-alive client is
attached.
- `rap_servers.rs`: RAP server lifecycle moves out of session start — boot happens
lazily when a thread's config first resolves (loading/replaying a thread never
boots servers), shutdown is driven by session idleness, and a shut-down server
reboots on the next tool interaction. `Up` caches the manifest (fetch failure
tears the boot down); MCP proxy tasks are owned (stdio children `kill_on_drop`)
but proxies stay up across idles since MCP servers may be stateful; migration
flows ride the same path (`MigrationServer`; `boot_rap_servers`/`rap_tools.rs`
deleted); config merging unified in `config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); the resident-runtime types (`RunningSystem`, `SubscribeHandle`,
`ThreadHandle`, `HandleObserver`, `HandleSubscribeRequest`, `LaunchingSystem`,
`ThreadBuilder`, `ChannelSender`, `ChannelSendError`, `ActiveThreads`) live under
`system::local`; `RunningSystem::begin_shutdown`/`task` are replaced by owned
`shutdown(self)`, and `RunningSystem::{send, send_user_text, subscribe,
thread_handle}` are infallible; `ThreadObserver` has no `on_commit` (persist in
`on_event` or externally; the turn is already durable before dispatch);
`EventCollector::take` returns `(thread_id, event)` pairs; builder tools are stored
as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`; `ToolContext`/builder
lose `input_queue_arn`; `ToolSet`/`VecToolSet` and `ToolsConfig::{add_command,
toolset_commands, mcp_servers, http_mcp_servers}` are removed. `infinity-daemon` no
longer exports `sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`,
`Session`, `InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
shadaj
added a commit
that referenced
this pull request
Aug 12, 2026
…on + lambda onto it
Introduce `infinity_agent_core::system`, a builder-based, object-oriented API for
running agents, rebuild both production embeddings on top of it, unify the parallel
implementations that had grown around the old architecture, and rewrite the
`infinity-runtime` docs around the new API.
Core (`infinity_agent_core::system`, resident runtime under `system::local`):
- `AgentSystemBuilder`: `new(stores, model, sender)` builds a step-mode `AgentSystem`
for serverless platforms; `new_local(stores, model)` builds a `LocalAgentSystem`
with an internal in-process queue (`ChannelSender`).
- Step mode's public surface is `AgentSystem::step(inputs, observer, defer)`: the
batch may span multiple threads (SQS FIFO batches can interleave message groups);
`step` partitions by `group_id`, applies the deferral policy per thread, joins the
per-thread slices concurrently (prepare → completion → history sync → tool
dispatch), and returns each thread's `StepOutcome`. `&mut self` serializes calls
per system; the internal `Thread` type is not public and `AgentSystem` is not
`Clone`, so two live in-memory views of one thread cannot exist through the
public API. Per-thread configuration resolves lazily on first step so loading a
thread never boots tool servers.
- `LocalAgentSystem` is mode-typed (`Handles` by default, `Launcher` after
`with_thread_launcher()`), with one `start()` per mode:
- Handles mode: `start()` returns a `RunningSystem` running an actor-system-style
runtime (router + one driver per active thread, input batching, `<interrupt>`
handling, deferral while a non-passive tool call is pending, auto-compaction
above 75% of the context window, race-free idle exit/respawn) with ack'd
`subscribe`, a `thread_exits` channel, and `thread_handle(id)` attaching to any
thread. `start_with_observer(factory)` swaps in a custom `ThreadObserver`.
- Launcher mode: `start()` returns a `LaunchingSystem`. `thread_builder()`
configures and launches new threads (generated IDs) with their own tools,
prompt, and optionally their own `ModelSource`, unioned onto the system-wide
configuration; launch configurations attach to the launched (root) thread, so
spawned subagent threads inherit tools, prompt, and model via root resolution.
`thread_handle(id)` re-attaches to existing threads only (`None` means the
thread does not exist). Launch configurations are process-lifetime.
- Shutdown is ownership-based: `RunningSystem::shutdown(self)` (and
`LaunchingSystem::shutdown(self)`) cancels every driver, flushes in-flight turns,
and awaits the router. Because shutting down consumes the system, `&self` methods
are infallible: `thread_handle`/`launch` return handles directly, `subscribe`
resolves once installed, and `send`/`send_user_text` return `()`. Independent
handles that can outlive the system (`SubscribeHandle`, `ThreadHandle`) keep
fallible signatures.
- `ThreadHandle`: `replay()` plus an exactly-once `events` stream and input sending;
subscriptions survive driver respawns via a shared registry.
- `ThreadObserver`: synchronous `on_event(AgentEvent)` fan-out (exactly-once relative
to `on_subscribe` replays) plus awaited user-choice durability hooks
(`on_user_choice_required`/`_dismissed`). Tool dispatch happens strictly after
history sync, so turns are durable before their effects are observable.
`EventCollector` buffers `(thread_id, event)` pairs.
- `ModelSource` (per-round resolution; enables mid-session model switching),
`ThreadConfigSource` (per-thread tools/prompt/notifier; enables per-session RAP
servers and toolsets), `DeferQueue` (`InMemoryDeferQueue`/`NoDeferral`),
`Tool::is_passive()`, tokio sleep tools (`with_tokio_sleep_tools`).
- `batch_processor` deleted: `process_batch` had no production callers and its
semantics violated the documented durability barrier (swallowed sync errors, then
dispatched; emitted ResponseDone before sync). The documented low-level API is now
`HistoryManager` + `prepare_input`/`run_completion`/`execute_action`;
`execute_action_with_error_result` (the #88 error-fallback) lives in
`event_processor`. `DisplayEvent` moved out of core: `AgentEvent` is core's only
event vocabulary; the CLI keeps a local, non-generic `DisplayEvent` as its
terminal-rendering enum. All unique regression coverage ported.
- Shared slice internals: `event_processor::input_echo` is the single computation of
accepted-input echoes; `SyntheticKind::is_compaction_complete()`;
`InputMessage::user_text()`; driver's in-flight step modeled as
`InFlightStep { fut, cancel_tx }` with a consuming `cancel()`.
- `rap_callback` module: the RAP callback → `InputMessage` conversion (multimodal
content, display segments, tagged synthetics incl. subscription final/associative
flags, user_choice, oauth) lives in core with 11 unit tests; the daemon delegates.
- In-memory `ConversationStore`/`StateStore` in `infinity_agent_core::stores` with a
serializable snapshot/restore API; the daemon's stores are thin wrappers (extras +
JSON persistence) so the ancestor/compaction/dedup semantics exist in exactly one
place. `spawn_thread_with_id` supports deterministic IDs.
- Platform-neutral core: `input_queue_arn` removed from `ToolContext`/builder.
Unused refactor leftovers pruned: `ToolSet`/`VecToolSet`,
`ToolsConfig::{add_command, toolset_commands, mcp_servers, http_mcp_servers}`,
`HistoryManager::state_store()`.
- All 20 daemon agent-loop tests migrated into core `system/tests.rs` with
byte-identical snapshots, plus new coverage: launcher tool/prompt/model unioning
and child inheritance, existing-only attachment, `ThreadHandle` streaming across
tool-call rounds, multi-group step batches, failed-dispatch fallback, OAuth
surfacing, prepare-level user-choice, handle respawn survival and pruning.
Daemon (`infinity-daemon`):
- One daemon-lifetime agent system replaces per-session agent loops and the
restart-on-send path (`needs_restart` → `start_session`): idle threads respawn a
driver transparently instead. Sessions are root threads; a `thread_exits` watcher
marks sessions idle and shuts down their RAP servers when no keep-alive client is
attached.
- `rap_servers.rs`: RAP server lifecycle moves out of session start — boot happens
lazily when a thread's config first resolves (loading/replaying a thread never
boots servers), shutdown is driven by session idleness, and a shut-down server
reboots on the next tool interaction. `Up` caches the manifest (fetch failure
tears the boot down); MCP proxy tasks are owned (stdio children `kill_on_drop`)
but proxies stay up across idles since MCP servers may be stateful; migration
flows ride the same path (`MigrationServer`; `boot_rap_servers`/`rap_tools.rs`
deleted); config merging unified in `config::load_merged_rap_config`.
- `DaemonObserver` implements the observer hooks; broadcasting unified in
`observer::broadcast_pruning`/`broadcast_to_thread` (fixing `handle_view_update`
never pruning dead subscribers); re-sent Connects replace rather than stack
subscriptions; replay rendering matches live rendering.
- `CatalogModelSource` resolves each thread's persisted model per round; fixed the
pre-existing double `ModelSwitched` delivery.
- `send_input` takes `user_driven: bool`; interior-mutability methods take `&self`.
- Web goldens `chat-image-result`/`chat-diff-result` regenerated (old goldens
captured an eager-boot layout artifact).
- Deleted: `session/thread_worker.rs`, `session/agent_loop.rs`, `sleep_tools.rs`,
`InMemoryMessageSender`, `Session`, `spawn_session`, `AgentMessage`.
Lambda (`infinity-agent-lambda`):
- Event handler rewritten on the new API: one system per invocation with a
`ThreadConfigSource` resolving each thread's RAP toolsets (DynamoDB manifest
cache) plus platform sleep tools (shared `WakeupScheduler` backend); one
batch-shaped `step`; per-thread outputs aggregated in a `BTreeMap`.
- New `rap-receiver` binary replaces the drifted JS receiver (deleted): same
Function-URL contract, full parity with the daemon's conversion via
`infinity_agent_core::rap_callback` (user_choice accepted, multimodal and
`display_as` preserved, tagged synthetics honored, stable dedup IDs where the
wire has identity). CDK switches to a cargo-lambda `RustFunction`; the Function
URL is recreated on deploy, so persisted callback URLs must be re-subscribed.
Docs (`docs/docs/infinity-runtime`): the single "Rust API" page is replaced by two
sections, written in the Hydro docs register (definitional openings, bolded key
terms, cause-and-effect guarantees, code-first examples, admonitions for caveats):
"The Agent System API" (overview, building-a-system, running-locally, step-mode,
observers, including launcher mode) and "The Low-Level API" (overview,
history-manager, completion-loop). Cross-references updated repo-wide; `cargo doc`
is warning-free.
BREAKING CHANGE: `infinity-agent-core` removes `batch_processor` (`process_batch`,
`process_input_item`, `DisplayEvent`); `Thread` is no longer public (use
`AgentSystem::step` or `ThreadHandle`); `AgentSystem` is not `Clone`;
`LocalAgentSystem::start_with_handles` is now `start()` and custom observers use
`start_with_observer` (`LocalAgentSystem` gained a mode type parameter, default
`Handles`); the resident-runtime types (`RunningSystem`, `SubscribeHandle`,
`ThreadHandle`, `HandleObserver`, `HandleSubscribeRequest`, `LaunchingSystem`,
`ThreadBuilder`, `ChannelSender`, `ChannelSendError`, `ActiveThreads`) live under
`system::local`; `RunningSystem::begin_shutdown`/`task` are replaced by owned
`shutdown(self)`, and `RunningSystem::{send, send_user_text, subscribe,
thread_handle}` are infallible; `ThreadObserver` has no `on_commit` (persist in
`on_event` or externally; the turn is already durable before dispatch);
`EventCollector::take` returns `(thread_id, event)` pairs; builder tools are stored
as `Rc<dyn Tool<M>>`; `Tool` gains defaulted `is_passive`; `ToolContext`/builder
lose `input_queue_arn`; `ToolSet`/`VecToolSet` and `ToolsConfig::{add_command,
toolset_commands, mcp_servers, http_mcp_servers}` are removed. `infinity-daemon` no
longer exports `sleep_tools`, `AgentMessage`, `thread_worker`, `agent_loop`,
`Session`, `InMemoryMessageSender`, `boot_rap_servers`, or `rap_tools`;
`SessionManager::switch_model` takes the requester's sender and returns
`Result<(), String>`; `send_input` takes `user_driven: bool`;
`SharedSessionManager` is `Rc<tokio::Mutex<...>>`. The CLI's `DisplayEvent` lives in
`infinity_agent_cli::display`.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #92
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.
send_tool_errorhelper that enqueues an errorToolResultwith the original tool and call IDsclose_thread.thread_id, from propagated errors or duplicated message construction into queued tool resultsError: Tool call failedresult when asynchronous tool execution returns an error, while retaining the detailed display event and logging delivery failuresclose_threadarguments and generic failed-tool fallback delivery