Repository navigation
refactor(infinity-agent-core): record interrupts as history entries instead of prefixing user text - #170
Closed
akainth015 wants to merge 1 commit into
Closed
refactor(infinity-agent-core): record interrupts as history entries instead of prefixing user text#170akainth015 wants to merge 1 commit into
akainth015 wants to merge 1 commit into
Conversation
Deploying infinity with
|
| Latest commit: |
ae72ccf
|
| Status: | ✅ Deploy successful! |
| Preview URL: | https://118a1ed7.infinity-dc7.pages.dev |
| Branch Preview URL: | https://pr-interrupt-history-kind.infinity-dc7.pages.dev |
akainth015
force-pushed
the
pr/interrupt-history-kind
branch
from
October 20, 2026 00:16
d7c64e1 to
b2c57e5
Compare
shadaj
reviewed
Oct 20, 2026
shadaj
left a comment
shadaj
left a comment
Member
There was a problem hiding this comment.
Can we also use this for tool interrupts instead of inserting "interrupted" tool results? The idea is that we can durably resume an agent by checking if the last input was an interrupt or assistant output.
| SyntheticKind::Tagged(TaggedSyntheticKind::ParentMessage { tool_call_id }) => { | ||
| tool_call_id | ||
| } | ||
| SyntheticKind::Tagged(TaggedSyntheticKind::Compaction) => ToolCallId::from_ref(""), |
Member
There was a problem hiding this comment.
We should file an issue to turn this into an Option returning method.
akainth015
force-pushed
the
pr/interrupt-history-kind
branch
from
October 20, 2026 17:42
b2c57e5 to
d2870d3
Compare
…nstead of prefixing user text
When user text interrupted an in-flight completion, the driver rewrote the
text to `<interrupt>…` before folding it into history, so the token the
model sees was baked into the stored user message. Likewise, an interrupted
tool call was settled by storing a synthetic "interrupted" tool result.
Interrupts are now first-class history entries:
* `TaggedSyntheticKind::Interrupt` is an input marker; `prepare_input`
folds it into `InfinityMessage::Interrupt` (settling a trailing
unanswered tool call like any other non-result entry). The driver
enqueues a marker ahead of the interrupting text instead of mutating it.
* The model-facing `<interrupt>` token is derived at render time by
`message::render_history` (used by `HistoryManager::get_history`): it is
prefixed onto the following user text (byte-for-byte what the model saw
before) or rendered standalone if no user text follows. Consecutive
interrupts collapse into one token.
* An unanswered tool call is settled by an `Interrupt` entry instead of a
stored synthetic "Tool call interrupted by user" result. `render_history`
renders an interrupt right after a tool call as that result, so the model
sees the same request (when text interrupts an in-flight synchronous
tool, the result alone now conveys it; no `<interrupt>` prefix follows).
History thus ends on an interrupt when the thread
was stopped, and the late real result is still dropped as stale.
* Stored user text is no longer rewritten. Rows written by older versions
(literal `<interrupt>` prefix, synthetic interrupted results) load and
render unchanged; replay keeps stripping the legacy prefix and hides
interrupt entries (as the live stream does).
Note: an older daemon cannot parse histories containing the new
`{"kind":"interrupt"}` entry (forward-incompatible on rollback).
akainth015
force-pushed
the
pr/interrupt-history-kind
branch
from
October 22, 2026 01:18
d2870d3 to
ae72ccf
Compare
akainth015
added a commit
that referenced
this pull request
Oct 22, 2026
…ing can't wake zombie sessions After a restart, viewing a stale session in the web UI flipped it permanently to Running (even once #161 reports it Idle): attaching a client wakes the thread's driver (`Live` clears the idle flag), and the driver parks forever because `awaiting_tool_result()` is true for its trailing unanswered call. Two causes: * **Orphaned calls**: the RAP server, input queue or timer that would deliver the result died with the old process. * **`sleep_until_event_or_input` held its driver**: its no-op call is never answered by design, but only `close_thread` was exempt from `awaiting_tool_result`, so any woken sleeping thread showed Running, restart or not. ## Fix Builds on #170 (`InfinityMessage::Interrupt`). * **core**: exported `tool_call_expects_result(name)` exempts `close_thread` and `sleep_until_event_or_input`; sleeping drivers now exit and are respawned by the router on the next event or input. * **daemon**: `PersistentConversationStore::settle_orphaned_tool_calls` runs at cold start, before the agent system starts, and settles every trailing call that expects a result with an `InfinityMessage::Interrupt` entry, as an interrupting message would (the model sees the usual "interrupted" result; the entry carries no cause, so the restart is recorded only in its `-orphaned-by-restart` message ID). History then ends on the interrupt: the thread is stopped, not waiting. A tail probe (`RawValue`, so only the last message is deserialized) keeps threads that don't end in such a call unloaded. ## Tests * `driver_idles_after_sleep_until_event_or_input_call` (core residency). * `settle_orphaned_tool_calls_answers_only_orphaned_calls` (per-tool selection, persistence, idempotence). * `startup_settles_orphaned_tool_calls` (the sweep runs at startup). Co-authored-by: Infinity 🤖 <infinity@hydro.run>
shadaj
added a commit
that referenced
this pull request
Oct 27, 2026
… persistence
Make the disk-backed daemon durable against being killed at any point, and
stop rewriting whole history files on every completion.
## Conversation log (`threads/{thread_id}.jsonl`)
* Each thread's conversation is now an append-only JSONL log (`persist::ThreadLog`).
The first line is a `header` carrying the format version and thread id; every
later line is a `message` (with the dedup id it was appended under) or a
`compaction_summary`. Appends are a single `write_all` + `fdatasync`, so a
crash can tear at most the final line.
* Loading recovers the **longest valid prefix**: a torn trailing line is
dropped (with a warning) and the file truncated to the last complete line
before anything new is appended. A complete-but-unparseable line, or a header
from a newer version, is a loud error rather than a silent reset.
* Legacy whole-snapshot `{thread_id}.json` files are migrated into the log on
first load (the log is written atomically, so a crash mid-migration just
retries); the legacy file is left in place and ignored thereafter.
## Versioned, atomically written documents
* `{thread_id}.meta.json`, `{thread_id}.views.json`, `state/{thread_id}.state.json`
and `sessions.json` now carry a `version` field and are written via
`tempfile::NamedTempFile` in the same directory + fsync + `persist` (rename)
(`persist::write_atomic`), so readers only ever see the old or the new
contents and abandoned temp files are removed on drop. Unversioned files from
older daemons read as version 0 and are upgraded eagerly on first load; files
from a newer daemon are refused instead of being overwritten.
* Load/save errors are no longer swallowed: trait methods propagate them, and
the synchronous getters log them and skip any save.
## Processed ids are looked up in history, not stored separately
* `StateStore::get_processed_ids` / `add_processed_message_ids` are **removed**.
`ConversationStore` gains `is_message_processed(thread_id, message_id)`, a point
lookup against the persisted history (each row/line already stores its dedup
id). This eliminates the partial-write hazard of a message landing in one
store but its dedup id not landing in the other, and the DSQL backend no
longer needs DynamoDB for this (it runs a `SELECT EXISTS` with a new
`(session_id, message_id)` index). The daemon's `.state.json` drops the
`processed_message_ids` field.
* `InMemoryConversationStore` keeps a per-thread `HashSet` of dedup ids
alongside the entries (same lock, same mutation paths) so the lookup is O(1).
* `HistoryManager::handle_input` is the new async entry point for incoming
inputs: it checks ids handled in this process, then asks the store; the in-
memory set is no longer preloaded. `handle_content` stays for in-process
synthetic content with fresh ids.
## Incorporated PR #170
Interrupts are recorded as a first-class `InfinityMessage::Interrupt` history
entry (rendered to the model at request time) instead of rewriting user text.
This means a persisted history that ends on an unanswered tool call *without*
an interrupt entry was cut off by a crash — the signal a future auto-resume
can use.
## Breaking changes
BREAKING CHANGE: `StateStore::get_processed_ids` and `add_processed_message_ids`
are removed; `ConversationStore::is_message_processed` is a new required method.
BREAKING CHANGE: `stores::ThreadState` no longer has `processed_message_ids`.
BREAKING CHANGE: daemons older than this cannot read the new `.jsonl` logs or
histories containing `interrupt` entries.
## Not covered (by design, for now)
Inputs the model has not yet validated are still held only in memory and are
lost on a crash (the existing context-overflow protection); a durable inbox is
the next step toward auto-resume.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #178
shadaj
added a commit
that referenced
this pull request
Oct 27, 2026
… persistence
Make the disk-backed daemon durable against being killed at any point, and
stop rewriting whole history files on every completion.
## Conversation log (`threads/{thread_id}.jsonl`)
* Each thread's conversation is now an append-only JSONL log (`persist::ThreadLog`).
The first line is a `header` carrying the format version and thread id; every
later line is a `message` (with the dedup id it was appended under) or a
`compaction_summary`. Appends are a single `write_all` + `fdatasync`, so a
crash can tear at most the final line.
* Loading recovers the **longest valid prefix**: a torn trailing line is
dropped (with a warning) and the file truncated to the last complete line
before anything new is appended. A complete-but-unparseable line, or a header
from a newer version, is a loud error rather than a silent reset.
* Legacy whole-snapshot `{thread_id}.json` files are migrated into the log on
first load (the log is written atomically, so a crash mid-migration just
retries); the legacy file is left in place and ignored thereafter.
## Versioned, atomically written documents
* `{thread_id}.meta.json`, `{thread_id}.views.json`, `state/{thread_id}.state.json`
and `sessions.json` now carry a `version` field and are written via
`tempfile::NamedTempFile` in the same directory + fsync + `persist` (rename)
(`persist::write_atomic`), so readers only ever see the old or the new
contents and abandoned temp files are removed on drop. Unversioned files from
older daemons read as version 0 and are upgraded eagerly on first load; files
from a newer daemon are refused instead of being overwritten.
* Load/save errors are no longer swallowed: trait methods propagate them, and
the synchronous getters log them and skip any save.
## Processed ids are looked up in history, not stored separately
* `StateStore::get_processed_ids` / `add_processed_message_ids` are **removed**.
`ConversationStore` gains `is_message_processed(thread_id, message_id)`, a point
lookup against the persisted history (each row/line already stores its dedup
id). This eliminates the partial-write hazard of a message landing in one
store but its dedup id not landing in the other, and the DSQL backend no
longer needs DynamoDB for this (it runs a `SELECT EXISTS` with a new
`(session_id, message_id)` index). The daemon's `.state.json` drops the
`processed_message_ids` field.
* `InMemoryConversationStore` keeps a per-thread `HashSet` of dedup ids
alongside the entries (same lock, same mutation paths) so the lookup is O(1).
* `HistoryManager::handle_input` is the new async entry point for incoming
inputs: it checks ids handled in this process, then asks the store; the in-
memory set is no longer preloaded. `handle_content` stays for in-process
synthetic content with fresh ids.
## Interruptions (incorporates and refines PR #170)
* User interruptions are recorded as a first-class `InfinityMessage::Interrupt`
history entry (rendered to the model at request time) instead of rewriting
user text. When user text settles a pending tool call, the marker is
committed **known-safe** (sound: a `ToolCall` at the tail implies no
unvalidated inputs, asserted by `append_known_safe`), so "the user stopped
the agent" is durable even if the text is later dropped as oversized.
* `step_no_defer` gains an unconditional commit barrier (`sync()`) between the
prepare and completion phases, so anything committed while folding inputs is
durable before tools are cancelled or the model is invoked.
* **Event** interruptions (subscription events, child reports) record no
marker: the event itself is the record and `render_history` infers the
interruption, answering the call with `TOOL_CALL_INTERRUPTED_BY_EVENT_TEXT`.
A history that ends on a bare tool call therefore means "cut off mid-flight"
— the signal a future auto-resume needs — rather than a deliberate stop.
* Interrupt markers that arrive while unvalidated inputs precede them stay
unvalidated but are never dropped by overflow recovery (fixes a #170 edge
case where a kept subscription event could follow a stranded tool call).
## Breaking changes
BREAKING CHANGE: `StateStore::get_processed_ids` and `add_processed_message_ids`
are removed; `ConversationStore::is_message_processed` is a new required method.
BREAKING CHANGE: `stores::ThreadState` no longer has `processed_message_ids`.
BREAKING CHANGE: daemons older than this cannot read the new `.jsonl` logs or
histories containing `interrupt` entries.
## Not covered (by design, for now)
Inputs the model has not yet validated are still held only in memory and are
lost on a crash (the existing context-overflow protection); a durable inbox is
the next step toward auto-resume.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #178
shadaj
added a commit
that referenced
this pull request
Nov 3, 2026
… persistence
Make the disk-backed daemon durable against being killed at any point, and
stop rewriting whole history files on every completion.
## Conversation log (`threads/{thread_id}.jsonl`)
* Each thread's conversation is now an append-only JSONL log (`persist::ThreadLog`).
The first line is a `header` carrying the format version and thread id; every
later line is a `message` (inputs carry the dedup id they arrived under) or a
`compaction_summary`. Appends are a single `write_all` + `fdatasync`, so a
crash can tear at most the final line.
* Loading recovers the **longest valid prefix**: a torn trailing line is
dropped (with a warning) and the file truncated to the last complete line
before anything new is appended. A complete-but-unparseable line, or a header
from a newer version, is a loud error rather than a silent reset.
* Legacy whole-snapshot `{thread_id}.json` files are migrated into the log on
first load (the log is written atomically, so a crash mid-migration just
retries); the legacy file is left in place and ignored thereafter.
## Versioned, atomically written documents
* `{thread_id}.meta.json`, `{thread_id}.views.json`, `state/{thread_id}.state.json`
and `sessions.json` now carry a `version` field and are written via
`tempfile::NamedTempFile` in the same directory + fsync + `persist` (rename)
(`persist::write_atomic`), so readers only ever see the old or the new
contents and abandoned temp files are removed on drop. Unversioned files from
older daemons read as version 0 and are upgraded eagerly on first load; files
from a newer daemon are refused instead of being overwritten.
* Load/save errors are no longer swallowed: trait methods propagate them, and
the synchronous getters log them and skip any save.
## Dedup ids live only on inputs, and are looked up in history
* `StateStore::get_processed_ids` / `add_processed_message_ids` are **removed**.
`ConversationStore` gains `is_message_processed(thread_id, message_id)`, a point
lookup against the persisted history. This eliminates the partial-write
hazard of a message landing in one store but its dedup id not landing in the
other, and the DSQL backend no longer needs DynamoDB for this (`SELECT EXISTS`
with a new `(session_id, message_id)` index). The daemon's `.state.json`
drops the `processed_message_ids` field.
* Only inputs from the input queue carry a dedup id: `append_messages` takes
`Vec<(InfinityMessage, Option<String>)>`, and the synthetic ids the runtime
used to mint for model output, synchronous/unknown-tool results, interrupt
markers and child-thread seeds are gone. `HistoryManager::handle_input` is
the single place dedup ids are checked and recorded (in-process set first,
then the store); `handle_content(message)` folds runtime-minted content with
no id. `handle_completion` and `run_completion` lose their id parameters.
* DSQL: `message_id` becomes nullable (`ALTER COLUMN … DROP NOT NULL`, which
Aurora DSQL supports, run idempotently at startup).
* `InMemoryConversationStore` keeps a per-thread `HashSet` of input dedup ids
alongside the entries (same lock, same mutation paths) so the lookup is O(1).
## Interruptions (incorporates and refines PR #170)
* User interruptions are recorded as a first-class `InfinityMessage::Interrupt`
history entry (rendered to the model at request time) instead of rewriting
user text. When user text settles a pending tool call, the marker is
committed **known-safe** (sound: a `ToolCall` at the tail implies no
unvalidated inputs, asserted by `append_known_safe`), so "the user stopped
the agent" is durable even if the text is later dropped as oversized.
* `step_no_defer` gains an unconditional commit barrier (`sync()`) between the
prepare and completion phases, so anything committed while folding inputs is
durable before tools are cancelled or the model is invoked.
* **Event** interruptions (subscription events, child reports) record no
marker: the event itself is the record and `render_history` infers the
interruption, answering the call with `TOOL_CALL_INTERRUPTED_BY_EVENT_TEXT`.
A history that ends on a bare tool call therefore means "cut off mid-flight"
— the signal a future auto-resume needs — rather than a deliberate stop.
* Interrupt markers that arrive while unvalidated inputs precede them stay
unvalidated but are never dropped by overflow recovery (fixes a #170 edge
case where a kept subscription event could follow a stranded tool call).
## Breaking changes
BREAKING CHANGE: `StateStore::get_processed_ids` and `add_processed_message_ids`
are removed; `ConversationStore::is_message_processed` is a new required method
and `append_messages` takes `Option<String>` dedup ids.
BREAKING CHANGE: `stores::ThreadState` no longer has `processed_message_ids`.
BREAKING CHANGE: `HistoryManager::handle_content` / `handle_completion` and
`run_completion` no longer take message/completion ids.
BREAKING CHANGE: daemons older than this cannot read the new `.jsonl` logs,
histories containing `interrupt` entries, or migrated sessions with id-less
entries.
## Not covered (by design, for now)
Inputs the model has not yet validated are still held only in memory and are
lost on a crash (the existing context-overflow protection); a durable inbox is
the next step toward auto-resume.
Co-authored-by: Infinity 🤖 <infinity@hydro.run>
PR: #178
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
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.
Prep for #165, per review. Interrupts become a first-class
InfinityMessage::Interrupthistory entry (input markerTaggedSyntheticKind::Interrupt, folded inprepare_input), and the driver no longer rewrites user text.The model-facing
<interrupt>token is derived at render time inmessage::render_history: it is prefixed onto the following user text (byte-identical provider output to today) or emitted standalone if no text follows; consecutive interrupts collapse. Legacy rows with a literal<interrupt>prefix load and render unchanged (no migration); replay hides interrupt entries.Rollback caveat: older daemons can't parse a history containing an interrupt entry.