flow message bus: message / inbox / post / watch - #93
Conversation
Directed pages (human: native iTerm notification with exponential backoff until acked, wait metrics; session: context delivery via hooks or listen), broadcast posts fanned out on write to watchers, watch subscriptions, page listen for wake-on-mail agents, aggressive hook handlers (session-start, user-prompt-submit ack-on-reply, post-tool-use inbox drain + notify-scan, stop post-nudge), skill §4.18 + references/paging.md, 90d retention sweep. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
…attern) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
flow message (directed, escalation schedule), flow inbox (list / pop [--wait] / ack / due / stats), flow post, flow watch [--me]. All terminal/notification UI removed from core — users script notifiers on top of flow inbox due; agents park a Monitor or background Bash on flow inbox pop --wait. Tables renamed bus_*, kind message|post. Skill §4.18 + references/messaging.md rewritten. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
…ession Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
… longer re-wakes the session every turn end Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
flow done / flow archive now delete the task's undeliverable pending inbox rows, its watches (as watcher and topic), nudge stamp, and listener row. Messages the task sent to a human that are still pending survive — closing a task doesn't un-ask a question the user hasn't seen; those clear on pop/ack, consumed rows age out via the 90d sweep. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
…Stop-nudge backoff Delivery now rides the parked `flow inbox pop --wait` listener: SessionStart teaches park-and-re-arm (and watching tasks the session depends on or creates), pop --wait's own output nudges the re-arm. Prompt-submit drain remains the listener-less fallback. skill update removes stale PostToolUse entries; the hook subcommand stays a silent no-op for old installs. Stop-nudge declines now back off 30m→1h→2h..4h (bus_nudges.attempts, migrated), reset by a post. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
…c pop claims Any assignee's queue is consumable: --me/--self forces the human self queue from a bound session (e.g. a dedicated inbox-monitor task); --as <assignee> targets any human queue (monitor/transport workers). --json on inbox/pop/due emits machine-readable rows. Pops are now atomic pending->acked/delivered claims, so concurrent consumers of one inbox never double-pop. watch gains the same identity overrides. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
…-as override The harness treats any Stop-hook output as blocking the turn and force-ends after repeated blocks, telling hooks to check stop_hook_active — the nudge now reads the stdin payload and stays silent on hook-driven continuation stops (parse failures fail-safe to silent). --me/--self collapsed into the one override that matters: --as <assignee>, with --as self as the user's own queue. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
…silence Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
If the turn ends with pending mail and no live pop --wait listener, the Stop hook drains it into context — the agent handles it before idling instead of the mail waiting for the user's next prompt. Real mail bypasses the post-nudge backoff and self-limits (the drain consumes the rows); a live listener takes precedence (its message is never stolen); stop_hook_active still silences everything. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
Session-start/stop hook guidance, pop --wait re-arm hint, and skill docs now name Monitor first (waits without holding a shell, wakes on exit) with background Bash as the fallback for harnesses without it. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
…loop Monitor streams stdout lines as events, so the loop is required and also removes re-arming entirely: while true; do flow inbox pop --wait --timeout 300 --json || true; done under persistent:true emits one JSON event per message and zero timeout noise. Background Bash stays the single-shot fallback with per-wake re-arm. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
|
Minimum Claude Code version note (verified against the Claude Code CHANGELOG):
Recommendation: document Claude Code ≥ v2.1.163 for full bus behavior. On older versions everything else (message/inbox/pop/watch, the Monitor listener, SessionStart/UserPromptSubmit deliveries) works; only the Stop-hook features degrade. |
…as alias) Stored kind renamed message|broadcast with a table-rebuild migration (CHECK constraint pins the old value); JSON surface now emits kind:broadcast. Stop nudge, skill, stats and usage updated. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
…ntion Atomic pop claims already guarantee exactly-once with concurrent consumers, so the bus_listeners table, heartbeats, pid probes, the Stop-hook precedence check, and the age-based sweeps all go. Stop now always drains stranded mail at turn end (a racing Monitor listener loses harmlessly). Retention: keep the newest 1000 consumed rows; pending never expires. bus_listeners dropped via DDL. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
…fixes Review findings (Anshul) + design directive: 1. Hooks no longer consume mail: MarkDelivered removed from every hook path; hooks emit pending COUNTS only. Consumption happens solely via flow inbox pop / the armed Monitor loop, so dropped hook output can never lose mail. Stop no longer touches the inbox at all. 2. Escalation off-by-one: first re-notify is now 1m (60<<attempts), sequence 1,2,4,8,16m then 30m cap. 3. Pending broadcasts no longer immortal: the row-count roll covers broadcasts of any status; only pending directed messages are exempt. 4. Backoff overflow: exponent clamped before shifting, cap can't be bypassed by unbounded attempts. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
- Ack-on-reply scoped to the replier's own queue (AND to_assignee): replying no longer acks mail addressed to other assignees. - flow message rejects a flag in the address slot (no more immortal '--urgent' queues) and rejects done/archived task addresses (undeliverable — their bus footprint is already cleaned). - CleanupTaskBus matches watcher suffixes exactly in Go instead of LIKE (slugs with _ / %% no longer delete sibling subscriptions). - flow init installs the full hook set (UserPromptSubmit + Stop, retired-hook cleanup) instead of stranding new installs without the ack contract until the next version bump. - Unbound SessionStart also surfaces the user's pending-mail count. - bus kind migration recreates the messages indexes inside the tx (RENAME kept the names on the old table; they self-healed only on the next open). - Deterministic oldest-first: ORDER BY created_at, rowid everywhere (second-granularity timestamps tie-break by insertion order). - Stale messaging.md paragraph promising Stop-hook mail delivery removed; dead MarkDelivered removed (again). Rejected finding: 'Stop hooks do not consume additionalContext' — empirically false on Claude Code >= v2.1.163 (observed live; also in the changelog); documented as the minimum version on PR #93. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
anshulsao
left a comment
anshulsao
left a comment
There was a problem hiding this comment.
Review — message bus
Read the full bus layer, hooks, and command surface (through 4229c16). The core is solid: atomic single-row pop claims give exactly-once with no double-delivery, UserPromptSubmit acks are scoped to the sender's own session, stop_hook_active fail-safes to silent, the post→broadcast table rebuild is transactional, and idempotent DDL is clean. The bus_listeners drop + row-count retention is a nice simplification (it also closed two earlier concerns of mine — a local-vs-UTC skew in the old date sweep, and a listener-registration race).
Four things I'd want addressed. #1 is the merge gate; the rest are small.
🔴 1. Stop-hook drain marks mail delivered but the payload may never reach the model — internal/app/bushooks.go (cmdHookStop → drainTaskInbox)
drainTaskInbox calls MarkDelivered (flips pending→delivered) and returns the bodies via Stop hookSpecificOutput.additionalContext. Classic Claude Code Stop hooks influence a turn only via exit-code-2/stderr or {"decision":"block"} — an exit-0 additionalContext payload is, as far as I know, not injected for Stop. If that's the case, the rows are already consumed (no future pop/due returns them) but the model never sees the text → silent loss.
Since 4229c16 removed the !listenerAlive gate, this now drains unconditionally on every idle turn-end, so the exposure is larger than before. The passing test only checks the CLI's JSON output, not real harness injection, so it can't catch this.
Ask: confirm against the installed Claude Code version that Stop actually injects additionalContext. If it doesn't, don't MarkDelivered in the Stop path (leave rows pending for a real pop).
🟡 2. Escalation backoff is doubled — first reminder is 2m, not 1m — internal/flowdb/bus.go BumpNotifyAttempt
delay := time.Duration(60*(1<<uint(attempts+1))) * time.Secondattempts+1 is an off-by-one: at attempts=0 the first delay is 60*(1<<1)=120s. Sequence becomes 2m,4m,8m,16m,30m — every step double the documented 1m,2m,4m…. Fix: 1<<uint(attempts).
🟡 3. Broadcasts fanned out to the human accumulate forever — internal/app/post.go fan-out + SweepBus
If the human identity watches anything (flow watch <project> / flow watch self), every flow broadcast inserts a kind='broadcast', status='pending' row into the human queue. Those rows never escalate (only kind='message' is due-eligible), are never surfaced (flow inbox due is the only thing notifiers script), and are never swept (retention is WHERE status != 'pending', and pending rows don't count toward the 1000-kept window either). They clear only via a manual flow inbox pop.
Ask: either sweep kind='broadcast' even while pending (they're FYI, losing an old one is fine), or don't durably enqueue broadcasts to the human identity at all — a human has no pop --wait listener anyway.
🟢 4. BumpNotifyAttempt int64 overflow after ~57 escalations — same function
60*(1<<uint(attempts+1)) overflows int64 around attempts+1 ≈ 58; the > 30*time.Minute cap runs after the multiply, so a negative result slips through and sets next_notify_at in the past → the message re-fires on every poll. Clamp attempts (or cap before multiplying). Fixing #2's shift shifts the threshold but doesn't remove it.
Minor
migrateBusKindBroadcast's INSERT INTO bus_messages SELECT … relies on positional column order matching the new schema — correct today, just an easy-to-break coupling if the column list ever drifts. Worth a column-name list.
Nice feature overall — happy-path message→due→pop→ack works end to end. Just want #1 settled before this merges.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
|
Thanks for the review — note it covers through
Your migration column-list nit is fixed in the latest commit. Fresh eyes on current head welcome. 🤖 Generated with Claude Code |
anshulsao
left a comment
anshulsao
left a comment
There was a problem hiding this comment.
LGTM — all four findings addressed in dd6dfbf, verified against the code:
- #1 Stop-hook consume —
drainTaskInbox+MarkDeliveredremoved; hooks inform, never consume. Fair rejection on the premise too — good call documenting the min Claude Code version (≥ v2.1.163) where Stop injectsadditionalContext. - #2 backoff —
60<<uint(attempts)gives the intended 1m,2m,4m,8m,16m,30m. - #3 broadcast pile-up — sweep now rolls broadcasts by count while keeping pending questions exempt. 👍
- #4 overflow — shift bounded by
attempts < 5, can't wrap.
Nice extras beyond the review too — ack scoped by to_assignee, exact-match watcher cleanup, deterministic created_at, rowid ordering, and flow init installing the full hook set. Ship it.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu
2 participants
What
A CLI-only message bus built into flow: sessions call for the user's attention, message each other, and broadcast updates — queued in flow.db, escalated on a schedule, wait times measured. flow ships no notification UI (users script their own on
flow inbox due) and hooks never consume mail (inform-only; the sole consumption paths areflow inbox popand an agent's Monitor loop).Commands
flow message <assignee>[/<task-slug>] "<body>" [--urgent](aliasmsg) — directed. Bare assignee = the human: pending on an escalating notify schedule (due immediately, then 1m→30m cap) until answered.assignee/task-slug= the session bound to that task. Bodies ≤200 chars; done/archived task addresses rejected.flow broadcast "<one-liner>"(aliaspost) — fan-out on write: one row per current watcher of the task/project/assignee; never interrupts, never escalates.flow watch <task|project|assignee> [--as <assignee>] | --list | --rm <target>— subscriptions.flow inbox [--as <assignee>] [--json]/flow inbox pop [--wait] [--timeout <s>] [--as] [--json]— consumption; pop atomically claims the oldest (exactly-once under concurrent consumers).--as self= the user's own queue from inside a bound session;--as <other>serves monitor/transport workers.flow inbox ack [<id>]/flow inbox due [--json]/flow inbox stats— manual answer; the escalation feed for user notifier scripts; wait metrics.Delivery & hooks (inform-only)
while true; do flow inbox pop --wait --timeout 300 --json || true; done— each message is an event that wakes the agent; no re-arming.stop_hook_activeloop guard). Never touches the inbox.Storage & retention
bus_messages/bus_watches/bus_nudgesin flow.db (idempotent DDL; transactional kind migration with in-tx index recreation). Pop = atomic status claim, rows never deleted by consumption. Retention: newest 1000 rollable rows kept (consumed rows + broadcasts of any status); only pending directed messages never expire. Task close-out (done/archive) deletes the task's undeliverable rows, watches, and nudge stamp.Skill
Lean §4.18 in SKILL.md +
references/messaging.md(address grammar, Monitor-loop listener, etiquette: one message per wait, broadcasts are FYI, message whoever must act).Testing
flowdb + command + hook tests, full suite green; live-verified end to end in a real session (message→due→pop→ack with wait metrics, broadcast fan-out, Monitor event delivery, Stop-nudge + backoff + loop guard). Two review rounds (internal multi-angle + Anshul's) addressed in-branch.
🤖 Generated with Claude Code
https://claude.ai/code/session_01PFGNJQo5xNBBrT7rZMuUTu