Bound receive observer lifetime during native stream reads

bend2/src/coordinator/logs.bend

@@ -197,7 +197,7 @@ def codex_kind(+kind: String) -> Bool:
   Bool.or(String.eq(kind,"codex_command_start"),String.eq(kind,"codex_command_end"))
 
 def frame_id_sql(+line: String) -> String:
-  "WITH frame(value) AS (VALUES(" ++ C.q(line) ++ ")) SELECT coalesce(json_extract(value,'$.toolCallId'),json_extract(value,'$.messageId'),json_extract(value,'$.message.id'),'') FROM frame;"
+  "WITH frame(value) AS (VALUES(" ++ C.q(line) ++ ")) SELECT CASE WHEN json_extract(value,'$.type')='message_update' AND json_type(value,'$.assistantMessageEvent.partial') IS NULL THEN '' ELSE coalesce(json_extract(value,'$.toolCallId'),json_extract(value,'$.messageId'),json_extract(value,'$.message.responseId'),json_extract(value,'$.message.id'),'') END FROM frame;"
 
 def frame_id(db: String, line: String, omp: Bool) -> IO(String):
   match omp:

bend2/src/coordinator/observer-checkpoint.bend

@@ -103,20 +103,19 @@ def changed_when(enabled: Bool, previous: String, next: String) -> Bool:
     case False{}: False{}
     case True{}: Bool.not(String.eq(previous,next))
 
-# Turn supplies the payload fields its reducer can change on this frame.
-# Cursor, command and filter changes always participate in the decision.
+# Output cursor changes are replayable from the prior durable spool offset.
+# Persist the cursor with the next reducer-state change or final checkpoint.
 def checkpoint_due_fields(+previous: State, +next: State, message_enabled: Bool, terminal_enabled: Bool, codex_enabled: Bool) -> Bool:
   +episode_changed = Bool.not(String.eq(episode(previous),episode(next)))
   +guidance_changed = Bool.not(String.eq(cursor(previous),cursor(next)))
   +message_changed = changed_when(message_enabled,last_message(previous),last_message(next))
-  +output_changed = Bool.not(String.eq(output(previous),output(next)))
   +activity_changed = Bool.not(String.eq(omp_activity(previous),omp_activity(next)))
   +primary_changed = Bool.not(String.eq(muse_primary(previous),muse_primary(next)))
   +current_changed = Bool.not(String.eq(muse_current(previous),muse_current(next)))
   +terminal_changed = changed_when(terminal_enabled,terminal(previous),terminal(next))
   +filter_changed = Bool.not(String.eq(filter_mode(previous),filter_mode(next)))
   +codex_changed = changed_when(codex_enabled,codex_log(previous),codex_log(next))
-  Bool.or(episode_changed,Bool.or(output_changed,Bool.or(activity_changed,Bool.or(guidance_changed,Bool.or(message_changed,Bool.or(primary_changed,Bool.or(current_changed,Bool.or(terminal_changed,Bool.or(filter_changed,codex_changed)))))))))
+  Bool.or(episode_changed,Bool.or(activity_changed,Bool.or(guidance_changed,Bool.or(message_changed,Bool.or(primary_changed,Bool.or(current_changed,Bool.or(terminal_changed,Bool.or(filter_changed,codex_changed))))))))
 
 def checkpoint_due(previous: State, next: State) -> Bool:
   checkpoint_due_fields(previous,next,True{},True{},True{})