Skip to content

refactor(infinity-daemon)!: crash-safe, versioned, append-only thread persistence - #178

Open
shadaj wants to merge 1 commit into
mainfrom
sandbox-85094202-d192-4d34-8a56-cbb5baf08cf0
Open

shadaj wants to merge 1 commit into
mainfrom
sandbox-85094202-d192-4d34-8a56-cbb5baf08cf0

Conversation

@shadaj

@shadaj shadaj commented Oct 27, 2026

Copy link
Copy Markdown
Member

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
    temp-file + fsync + rename (persist::write_atomic), so readers only ever see
    the old or the new contents. 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.

@cloudflare-workers-and-pages

cloudflare-workers-and-pages Bot commented Oct 27, 2026 •

Copy link
Copy Markdown

Deploying infinity with  Cloudflare Pages  Cloudflare Pages

Latest commit: a2cba45
Status: ✅  Deploy successful!
Preview URL: https://458f9f09.infinity-dc7.pages.dev
Branch Preview URL: https://sandbox-85094202-d192-4d34-8.infinity-dc7.pages.dev

View logs

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
shadaj force-pushed the sandbox-85094202-d192-4d34-8a56-cbb5baf08cf0 branch 2 times, most recently from d0494c5 to e9dad5c Compare October 27, 2026 22:48
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
shadaj marked this pull request as ready for review November 3, 2026 19:13
@shadaj
shadaj requested a review from a team November 3, 2026 19:13
… 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
@shadaj
shadaj force-pushed the sandbox-85094202-d192-4d34-8a56-cbb5baf08cf0 branch from e9dad5c to a2cba45 Compare November 3, 2026 19:32
@shadaj
shadaj requested a review from akainth015 November 3, 2026 19:32
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant