diff --git a/bend2/src/coordinator/logs.bend b/bend2/src/coordinator/logs.bend index 613d89d51..1fe0b48b4 100644 --- a/bend2/src/coordinator/logs.bend +++ b/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: diff --git a/bend2/src/coordinator/observer-checkpoint.bend b/bend2/src/coordinator/observer-checkpoint.bend index 14cec8513..65ed123ec 100644 --- a/bend2/src/coordinator/observer-checkpoint.bend +++ b/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{}) diff --git a/bend2/src/coordinator/turn.bend b/bend2/src/coordinator/turn.bend index bef376fd1..1ef8b893a 100644 --- a/bend2/src/coordinator/turn.bend +++ b/bend2/src/coordinator/turn.bend @@ -636,12 +636,76 @@ def with_restore_result(-A: Type, result: Result<&1,&1,U32 & String,Unit>, next: # The host returns a right-nested line, notice and subscription status. type WakeRead is Data: AwakeEnd{notice: String, active: Bool} + AwakeInvalid{notice: String, active: Bool} AwakeLine{line: String, notice: String, active: Bool} +type ConsumeCheckpoint is Data: + ConsumeCheckpoint{guidance_cursor: String, last_message: String, muse_primary: String, muse_current: String, terminal: String, filter_mode: String, codex_log: String, held: String, omp_activity: String, output: String, output_done: Bool, episode: String} + +def consume_checkpoint_from(state: Observer.State) -> ConsumeCheckpoint: + match state: + case Observer.State{guidance_cursor,last_message,muse_primary,muse_current,terminal,filter_mode,codex_log,held,omp_activity,output,output_done,episode}: + ConsumeCheckpoint{guidance_cursor,last_message,muse_primary,muse_current,terminal,filter_mode,codex_log,held,omp_activity,output,output_done,episode} + +def consume_checkpoint_to(state: ConsumeCheckpoint) -> Observer.State: + match state: + case ConsumeCheckpoint{guidance_cursor,last_message,muse_primary,muse_current,terminal,filter_mode,codex_log,held,omp_activity,output,output_done,episode}: + Observer.State{guidance_cursor,last_message,muse_primary,muse_current,terminal,filter_mode,codex_log,held,omp_activity,output,output_done,episode} + +type ConsumeDeliveries is Data: + ConsumeNoDeliveries{} + ConsumeDelivery{channel: Chan(Result<&1,&1,U32 & String,String>), rest: ConsumeDeliveries} + +def consume_deliveries_from(values: List)>) -> ConsumeDeliveries: + match values: + case Nil{}: ConsumeNoDeliveries{} + case Con{head,tail}: ConsumeDelivery{head,consume_deliveries_from(tail)} + +def consume_deliveries_to(values: ConsumeDeliveries) -> List)>: + match values: + case ConsumeNoDeliveries{}: Nil{} + case ConsumeDelivery{channel,rest}: Con{channel,consume_deliveries_to(rest)} + +type ConsumeLoopState is Data: + ConsumeLoopState{policy_level: String, held: String, log_failed: Bool, cursor: String, last_message: String, primary: String, current: String, terminal: String, filter: String, requested: String, codex_log: String, first_error: String, checkpoint_state: ConsumeCheckpoint, incomplete_observation: Bool, wake_on: Bool, deliveries: ConsumeDeliveries} + +type ConsumeStep is Data: + ConsumeContinue{state: ConsumeLoopState, wake_on: Bool} + ConsumeComplete{outcome: Outcome, deliveries: ConsumeDeliveries} + +def consume_policy_level(policy: Logs.Policy) -> String: + match policy: + case Logs.Policy{level}: level + +def consume_complete_pair(result: Outcome & List)>) -> ConsumeStep: + match result: + case (outcome,deliveries): ConsumeComplete{outcome,consume_deliveries_from(deliveries)} + +def consume_complete(action: IO(Outcome & List)>)) -> IO(ConsumeStep): + IO.bind(Outcome & List)>, ConsumeStep, action, result => IO.pure(ConsumeStep,consume_complete_pair(result))) + +def native_line_contains_nul_scan(+line: String, +nul: Char, found: Bool) -> Bool: + match line found: + case SNil{} True{}: True{} + case SNil{} False{}: False{} + case SCon{h,t} True{}: True{} + case SCon{h,t} False{}: native_line_contains_nul_scan(t,nul,Char.is_eq(h,nul)) + +def native_line_contains_nul(+line: String) -> Bool: + native_line_contains_nul_scan(line,Char.from_u32(0),False{}) + +def wake_text_result_case(+text: String, notice: String, active: Bool, has_nul: Bool) -> WakeRead: + match has_nul: + case True{}: AwakeInvalid{notice,active} + case False{}: AwakeLine{text,notice,active} + +def wake_text_result(+text: String, notice: String, active: Bool) -> WakeRead: + wake_text_result_case(text,notice,active,native_line_contains_nul(text)) + def wake_read(line: Maybe, notice: String, active: Bool) -> WakeRead: match line: case None{}: AwakeEnd{notice,active} - case Some{text}: AwakeLine{text,notice,active} + case Some{text}: wake_text_result(text,notice,active) def subscribe_fields(opened: Result<&1,&1,U32 & String,U32 & String>) -> Sigma<&2,&2,U32,_ => Bool>: match opened: @@ -714,10 +778,31 @@ def consume_exhausted(+db: String, +player: String, +id: String, +log: String, h +incomplete : Bool <- IO.pure(Bool,Bool.or(incomplete_observation,Bool.or(checkpoint_write_failed(unshared_result(Unit,wake_closed)),Bool.or(wait_failed(unshared_result(String,status)),Bool.or(checkpoint_write_failed(unshared_result(Unit,sent)),Bool.or(checkpoint_write_failed(unshared_result(Unit,flushed)),Bool.or(checkpoint_write_failed(unshared_result(Unit,checkpoint_done)),Bool.or(checkpoint_write_failed(unshared_result(Unit,checkpointed)),observation_store_failed(unshared_result(String,closed)))))))))) IO.pure(Outcome & List)>,(Ended{status_text(unshared_result(String,status)),event,error,incomplete},deliveries)) +def invalid_tail_error() -> String: + "Native output contains an embedded NUL byte; observation stopped at the last validated checkpoint." + +def invalid_tail_terminal_status(+status: String) -> Bool: + Bool.or(Bool.and(Tx.starts_with(status,"exit "),Bool.not(String.eq(status,"exit 0"))),Bool.or(Tx.starts_with(status,"signal "),String.eq(status,"unknown after keeper loss"))) + +def invalid_tail_matches_error(+error: String) -> Bool: + String.eq(error,invalid_tail_error()) + +def invalid_tail_ended(+status: String, event: String, +error: String, deliveries: List)>) -> Outcome & List)>: + terminal : Bool = invalid_tail_terminal_status(status) + only_corruption : Bool = invalid_tail_matches_error(error) + (Ended{status,event,error,Bool.pick(Bool,Bool.and(terminal,only_corruption),False{},True{})},deliveries) + +def invalid_tail_result(result: Outcome & List)>) -> Outcome & List)>: + match result: + case (Ended{status,event,error,incomplete},deliveries): invalid_tail_ended(status,event,error,deliveries) + +def invalid_tail_finish(action: IO(Outcome & List)>)) -> IO(Outcome & List)>): + IO.bind(Outcome & List)>,Outcome & List)>,action,result => IO.pure(Outcome & List)>,invalid_tail_result(result))) + @unsafe -def consume(+db: String, +player: String, +id: String, +log: String, +policy: Logs.Policy, +held: String, +log_failed: Bool, +handle: U32, +harness: String, +cursor: String, +last_message: String, +primary: String, +current: String, +terminal: String, +filter: String, +requested: String, +codex_log: String, +first_error: String, +sender: Chan(Result<&1,&1,U32 & String,Unit>), +attempt: String, +checkpoint_state: Observer.State, +incomplete_observation: Bool, +wake_sub: U32, wake_on: Bool, deliveries: List)>, result: Result<&1,&1,U32 & String,WakeRead>) -> IO(Outcome & List)>): +def consume_frame(+db: String, +player: String, +id: String, +log: String, +policy: Logs.Policy, +held: String, +log_failed: Bool, +handle: U32, +harness: String, +cursor: String, +last_message: String, +primary: String, +current: String, +terminal: String, +filter: String, +requested: String, +codex_log: String, +first_error: String, +sender: Chan(Result<&1,&1,U32 & String,Unit>), +attempt: String, +checkpoint_state: Observer.State, +incomplete_observation: Bool, +wake_sub: U32, wake_on: Bool, deliveries: List)>, result: Result<&1,&1,U32 & String,WakeRead>) -> IO(ConsumeStep): match result: - case Fail{error}: do IO)>>: + case Fail{error}: do IO: wake_closed : Result<&1,&1,U32 & String,Unit> <- wake_close(wake_sub,wake_on) closed : Result<&1,&1,U32 & String,Unit> <- P.ProcessChild.close_stdin(handle) sent : Result<&1,&1,U32 & String,Unit> <- IO.join(Result<&1,&1,U32 & String,Unit>,sender) @@ -732,17 +817,17 @@ def consume(+db: String, +player: String, +id: String, +log: String, +policy: Lo sent_error : String <- IO.pure(String,append_write_error(append_write_error(append_write_error(read_error,wake_closed),closed),sent)) waited_error : String <- IO.pure(String,append_store_error(sent_error,unshared_result(String,waited))) error : String <- IO.pure(String,append_write_error(append_write_error(append_write_error(waited_error,unshared_result(Unit,flushed)),checkpoint_done),named)) - IO.pure(Outcome & List)>,(Ended{status_text(unshared_result(String,waited)),terminal,error,True{}},deliveries)) - case Done{AwakeEnd{+notice,+alive}}: do IO)>>: + IO.pure(ConsumeStep,ConsumeComplete{Ended{status_text(unshared_result(String,waited)),terminal,error,True{}},consume_deliveries_from(deliveries)}) + case Done{AwakeInvalid{+notice,+alive}}: + consume_complete(invalid_tail_finish(consume_exhausted(db,player,id,log,held,log_failed,handle,harness,codex_log,terminal,filter,requested,append_error(invalid_tail_error(),first_error),sender,attempt,checkpoint_state,True{},deliveries,wake_sub,alive))) + case Done{AwakeEnd{+notice,+alive}}: do IO: +drained : String <- Guidance.awake_pending(db,player,handle,cursor,Bool.and(String.eq(harness,"omp\n"),Bool.and(Bool.not(String.is_empty(notice)),String.is_empty(terminal)))) - continuation : (List)> -> IO(Outcome & List)>)) <- IO.pure(List)> -> IO(Outcome & List)>),Bool.pick(List)> -> IO(Outcome & List)>),String.is_empty(notice), - pending => consume_exhausted(db,player,id,log,held,log_failed,handle,harness,codex_log,terminal,filter,requested,first_error,sender,attempt,checkpoint_state,incomplete_observation,pending,wake_sub,alive), - pending => do IO)>>: - next : Result<&1,&1,U32 & String,WakeRead> <- wake_line(handle,wake_sub,alive) - consume(db,player,id,log,policy,held,log_failed,handle,harness,drained,last_message,primary,current,terminal,filter,requested,codex_log,first_error,sender,attempt,checkpoint_state,incomplete_observation,wake_sub,alive,pending,next))) + continuation : (List)> -> IO(ConsumeStep)) <- IO.pure(List)> -> IO(ConsumeStep),Bool.pick(List)> -> IO(ConsumeStep),String.is_empty(notice), + pending => consume_complete(consume_exhausted(db,player,id,log,held,log_failed,handle,harness,codex_log,terminal,filter,requested,first_error,sender,attempt,checkpoint_state,incomplete_observation,pending,wake_sub,alive)), + pending => IO.pure(ConsumeStep,ConsumeContinue{ConsumeLoopState{consume_policy_level(policy),held,log_failed,drained,last_message,primary,current,terminal,filter,requested,codex_log,first_error,consume_checkpoint_from(checkpoint_state),incomplete_observation,alive,consume_deliveries_from(pending)},alive}))) continuation(deliveries) case Done{AwakeLine{+line,notice,+alive}}: - do IO)>>: + do IO: +drained : String <- Guidance.awake_pending(db,player,handle,cursor,Bool.and(String.eq(harness,"omp\n"),Bool.and(Bool.not(String.is_empty(notice)),String.is_empty(terminal)))) +current_policy : Logs.Policy <- Logs.level_open(db,player) +kind : String <- frame_kind(db,line,harness) @@ -754,7 +839,7 @@ def consume(+db: String, +player: String, +id: String, +log: String, +policy: Lo +event : String <- omp_terminal(db,line,message,activity,Bool.and(String.eq(harness,"omp\n"),String.eq(Tx.trim_nl(kind),"agent_end"))) +kept : String <- Logs.retained(db,current_policy,Tx.trim_nl(kind),line,event,log) step : List & String <- IO.pure(List & String,Logs.frame_step(current_policy,Tx.trim_nl(kind),frame_key,kept,held)) - with_frame_step(Outcome & List)>,step,lines => next_held_value => do IO)>>: + with_frame_step(ConsumeStep,step,lines => next_held_value => do IO: +next_held : String <- IO.pure(String,next_held_value) +positioned : Result<&2,&2,Sigma<&2,&2,U32,_ => String>,String> <- shared_result_io(String,Logs.write_positioned(log,lines,Observer.output(checkpoint_state),log_failed,Bool.not(String.is_empty(attempt)),False{})) +logged : Result<&2,&2,Sigma<&2,&2,U32,_ => String>,Unit> <- shared_result_io(Unit,IO.pure(Result<&1,&1,U32 & String,Unit>,Logs.position_write(log,unshared_result(String,positioned)))) @@ -782,7 +867,7 @@ def consume(+db: String, +player: String, +id: String, +log: String, +policy: Lo +mode : String <- filter_mode(db,line,Bool.and(String.eq(harness,"omp\n"),String.eq(Tx.trim_nl(kind),"response")),requested) +next_filter : String <- IO.pure(String,Omp.filter_next(filter,Tx.trim_nl(mode))) +next_checkpoint : Observer.State <- IO.pure(Observer.State,Observer.with_episode(Observer.with_output(Observer.with_omp_activity(Observer.make(next_cursor,message,next_primary,next_current,ended,next_filter,codex_state,next_held),activity),Logs.position(Observer.output(checkpoint_state),unshared_result(String,positioned))),episode)) - NativeRequests.with_observation(Outcome & List)>,questions,wakes => question_error => question_incomplete => do IO)>>: + NativeRequests.with_observation(ConsumeStep,questions,wakes => question_error => question_incomplete => do IO: +question_text : String <- IO.pure(String,question_error) +request_incomplete : Bool <- IO.pure(Bool,question_incomplete) +checkpoint_enabled : Bool <- IO.pure(Bool,Bool.and(Bool.not(incomplete_observation),Bool.and(Bool.not(failed),Bool.and(Bool.not(request_incomplete),Bool.and(Bool.not(String.is_empty(attempt)),Observer.checkpoint_due_fields(checkpoint_state,next_checkpoint,message_frame,terminal_selected,codex_frame)))))) @@ -790,8 +875,26 @@ def consume(+db: String, +player: String, +id: String, +log: String, +policy: Lo checkpoint_notice : Unit <- checkpoint_failure_notice(db,player,id,attempt,checkpoint_enabled,unshared_result(Unit,checkpointed)) +next_error : String <- IO.pure(String,append_write_error(append_write_error(append_store_error(append_write_error(append_error(first_error,question_text),unshared_result(Unit,written)),unshared_result(String,observed)),unshared_result(Unit,closed_input)),unshared_result(Unit,checkpointed))) +next_incomplete : Bool <- IO.pure(Bool,Bool.or(incomplete_observation,Bool.or(request_incomplete,Bool.or(observation_store_failed(unshared_result(String,observed)),Bool.or(checkpoint_write_failed(unshared_result(Unit,closed_input)),Bool.or(checkpoint_write_failed(unshared_result(Unit,checkpointed)),failed)))))) - next : Result<&1,&1,U32 & String,WakeRead> <- wake_line(handle,wake_sub,alive) - consume(db,player,id,log,current_policy,next_held,failed,handle,harness,next_cursor,message,next_primary,next_current,ended,next_filter,requested,codex_state,next_error,sender,attempt,next_checkpoint,next_incomplete,wake_sub,alive,wakes,next))) + IO.pure(ConsumeStep,ConsumeContinue{ConsumeLoopState{consume_policy_level(current_policy),next_held,failed,next_cursor,message,next_primary,next_current,ended,next_filter,requested,codex_state,next_error,consume_checkpoint_from(next_checkpoint),next_incomplete,alive,consume_deliveries_from(wakes)},alive}))) + +# The retained process sends its initial input once. Reattachment reads the same +# attempt from its first frame so native completion can be reconstructed. +def consume_step_resume(step: ConsumeStep, continue_loop: Bool -> ConsumeLoopState -> IO(Outcome & List)>)) -> IO(Outcome & List)>): + match step: + case ConsumeComplete{outcome,deliveries}: IO.pure(Outcome & List)>,(outcome,consume_deliveries_to(deliveries))) + case ConsumeContinue{state,wake_on}: continue_loop(wake_on,state) + +@unsafe +def consume_loop(+db: String, +player: String, +id: String, +log: String, +handle: U32, +harness: String, +sender: Chan(Result<&1,&1,U32 & String,Unit>), +attempt: String, +wake_sub: U32, state: ConsumeLoopState, result: Result<&1,&1,U32 & String,WakeRead>) -> IO(Outcome & List)>): + match state: + case ConsumeLoopState{policy_level,held,log_failed,cursor,last_message,primary,current,terminal,filter,requested,codex_log,first_error,checkpoint_state,incomplete_observation,wake_on,deliveries}: + IO.bind(ConsumeStep, Outcome & List)>, consume_frame(db,player,id,log,Logs.Policy{policy_level},held,log_failed,handle,harness,cursor,last_message,primary,current,terminal,filter,requested,codex_log,first_error,sender,attempt,consume_checkpoint_to(checkpoint_state),incomplete_observation,wake_sub,wake_on,consume_deliveries_to(deliveries),result), step => consume_step_resume(step, next_wake_on => next_state => do IO)>>: + next : Result<&1,&1,U32 & String,WakeRead> <- wake_line(handle,wake_sub,next_wake_on) + consume_loop(db,player,id,log,handle,harness,sender,attempt,wake_sub,next_state,next))) + +@unsafe +def consume(+db: String, +player: String, +id: String, +log: String, +policy: Logs.Policy, +held: String, +log_failed: Bool, +handle: U32, +harness: String, +cursor: String, +last_message: String, +primary: String, +current: String, +terminal: String, +filter: String, +requested: String, +codex_log: String, +first_error: String, +sender: Chan(Result<&1,&1,U32 & String,Unit>), +attempt: String, +checkpoint_state: Observer.State, +incomplete_observation: Bool, +wake_sub: U32, wake_on: Bool, deliveries: List)>, result: Result<&1,&1,U32 & String,WakeRead>) -> IO(Outcome & List)>): + consume_loop(db,player,id,log,handle,harness,sender,attempt,wake_sub,ConsumeLoopState{consume_policy_level(policy),held,log_failed,cursor,last_message,primary,current,terminal,filter,requested,codex_log,first_error,consume_checkpoint_from(checkpoint_state),incomplete_observation,wake_on,consume_deliveries_from(deliveries)},result) def direct_observed(observation: Outcome & List)>) -> IO(Outcome): (outcome,deliveries)=observation @@ -815,9 +918,6 @@ def started(+db: String, +player: String, +id: String, +log: String, prompt: Str observation : Outcome & List)> <- consume(db,player,id,generation,policy,"",False{},handle,harness,"0\n","","","","","",requested,"",restore_error,sender,"",Observer.empty(),restore_failed,0,False{},Nil{},first) direct_observed(observation)) -# The retained process sends its initial input once. Reattachment reads the same -# attempt from its first frame so native completion can be reconstructed. -@unsafe def retained_output(+db: String, +player: String, +id: String, +log: String, +harness: String, +attempt: String, +handle: U32, +supplied_requested: String) -> IO(Outcome & List)>): do IO)>>: +generation : String <- IO.pure(String,generation_log(log,id)) @@ -1046,9 +1146,9 @@ def refusal_read(resumable: Bool, event: String, text: Result<&1,&1,U32 & String def refusal(+log: String, resumable: Bool, +outcome: Outcome) -> IO(Bool): match outcome: case Ended{status,event,error,incomplete}: - do IO: + Bool.pick(IO(Bool),Tx.starts_with(error,invalid_tail_error()),IO.pure(Bool,False{}),do IO: text : Result<&1,&1,U32 & String,String> <- T.Text.read(log) - refusal_read(resumable,event,text) + refusal_read(resumable,event,text)) # The refused conversation is gone, so the pending input runs fresh and names # the state the workspace holds; the fresh turn records its own identity. diff --git a/bend2/test/logs.py b/bend2/test/logs.py index 8b674677b..6d68835dd 100644 --- a/bend2/test/logs.py +++ b/bend2/test/logs.py @@ -760,6 +760,7 @@ def test_unfinished_assistant_message_keeps_latest_update(self): frames = [json.dumps({'type': 'message_start', 'messageId': 'unfinished-message', 'message': {'id': 'unfinished-message', 'role': 'assistant', 'content': []}})] frames.extend(json.dumps({'type': 'message_update', 'messageId': 'unfinished-message', + 'assistantMessageEvent': {'partial': {'type': 'text', 'text': 'partial answer %d' % i}}, 'message': {'id': 'unfinished-message', 'role': 'assistant', 'content': [{'type': 'text', 'text': 'partial answer %d' % i}]}}) for i in range(3)) @@ -767,6 +768,15 @@ def test_unfinished_assistant_message_keeps_latest_update(self): updates = [row for row in map(json.loads, self.lines()) if row.get('type') == 'message_update'] self.assertEqual(updates, [json.loads(frames[-1])]) + def test_update_deltas_with_message_id_are_appended(self): + frames = [json.dumps({'type': 'message_update', 'messageId': 'delta-message', + 'assistantMessageEvent': {'type': 'text_delta', 'delta': 'part %d' % i}, + 'message': {'role': 'assistant'}}) + for i in range(3)] + self.stream(frames + [self.terminal()]) + updates = [row for row in map(json.loads, self.lines()) if row.get('type') == 'message_update'] + self.assertEqual(updates, list(map(json.loads, frames))) + def test_existing_policy_migrates_and_preserves_registered_logs(self): with sqlite3.connect(self.db) as connection: connection.executescript("DROP TABLE log_policies;DROP TABLE log_files;" diff --git a/bend2/test/receive.py b/bend2/test/receive.py index 22108421b..76bc45095 100644 --- a/bend2/test/receive.py +++ b/bend2/test/receive.py @@ -417,10 +417,42 @@ def session_lock_path(db,session): print(json.dumps({'type':'system','subtype':'init','session_id':native}),flush=True) reply({'initialized':native}) continue + if action.get('emit_nul_tail'): + count=int(action['emit_nul_tail']) + os.write(1,b'\x00'*count) + sys.stdout.flush() + reply({'nul_tail_written':count}) + if action.get('exit_after_nul_tail',False): break + continue if action.get('exit_fixture'): if action.get('complete_task',False): confirm_finished_fixture_task('Completed the controlled native observation task.') break + if action.get('emit_delta_batch'): + batch=action['emit_delta_batch'] + count=int(batch['count']) + payload='x'*int(batch['bytes']) + frame=json.dumps({'type':'message_update','assistantMessageEvent':{'type':'text_delta','delta':payload}}) + for _ in range(count): print(frame) + sys.stdout.flush() + reply({'delta_frames_written':count,'batch':batch['id']}) + continue + if action.get('emit_snapshot'): + size=int(action['emit_snapshot']) + payload='s'*size + frame={'type':'message_update', + 'assistantMessageEvent':{'partial':{'type':'text','text':payload}}, + 'message':{'role':'assistant','content':[{'type':'text','text':payload}]}} + line=json.dumps(frame) + print(line,flush=True) + reply({'snapshot_bytes_written':size}) + continue + if action.get('emit_checkpoint_message'): + ident=action['emit_checkpoint_message'] + frame={'type':'message_end','message':{'id':ident,'role':'assistant','content':[{'type':'text','text':'checkpoint '+ident}]}} + print(json.dumps(frame),flush=True) + reply({'checkpoint_message_written':ident}) + continue if action.get('progress'): progress(action) continue @@ -696,6 +728,35 @@ def output_log(self, session): self.fail(f'{session} has no registered output log generation') return pathlib.Path(row[0]) + def checkpoint_covers_output_marker(self, session, marker, checkpoint, raw_output): + """Wait for the observer cursor to cover a valid marker before tail tests.""" + if not checkpoint.is_file() or not raw_output.is_file(): + return False + log = self.output_log(session) + if not log.is_file(): + return False + log_size = log.stat().st_size + with log.open('rb') as stream: + stream.seek(max(0, log_size - 8192)) + tail = stream.read() + marker_at = tail.rfind(marker.encode()) + if marker_at < 0: + return False + line_end = tail.find(b'\n', marker_at) + if line_end < 0: + return False + record = checkpoint.read_bytes() + state_at = record.find(b'{"schema":1,') + if state_at < 0: + return False + try: + state = json.loads(record[state_at:]) + output_cursor = int(state.get('output', -1)) + except (ValueError, TypeError): + return False + absolute_line_end = log_size - len(tail) + line_end + 1 + return output_cursor >= absolute_line_end + def reference_startup_stderr(self, executable, session): """Read the complete platform diagnostic for the harness's fresh launch. @@ -4439,6 +4500,368 @@ def test_guidance_arrives_while_retained_native_is_silent(self): self.assertEqual(len((self.directory / 'native-launches.jsonl').read_text().splitlines()), 1) self.assert_no_start() + def test_nul_tail_after_exit_zero_preserves_raw_checkpoint_and_pending_input(self): + self.player(harness='omp') + config_path = self.directory / 'fixture.json' + config = json.loads(config_path.read_text()) + config['record_launches'] = True + config_path.write_text(json.dumps(config)) + self.prepare_input('nul-exit-zero-task', 'parent', + 'Retain the native attempt when successful exit has invalid output.', + kind='task') + observer = self.spawn(*self.receive_args('parent')) + control, started = self.accept('parent') + native_id = self.eventually(lambda: self.coord('player', 'parent')['native'], + 'The original OMP identity was not recorded.') + self.assertEqual(native_id, started['native']) + with sqlite3.connect(self.db) as database: + old_id, old_directory = database.execute( + 'SELECT id,directory FROM executions WHERE session=?', ('parent',)).fetchone() + attempt = pathlib.Path(old_directory) + checkpoint = attempt / 'checkpoint' + raw_output = attempt / 'stdout' + + self.action(control, emit_checkpoint_message='nul-exit-zero-checkpoint') + self.assertEqual(json.loads(control.readline()), + {'checkpoint_message_written':'nul-exit-zero-checkpoint'}) + self.eventually(lambda: self.checkpoint_covers_output_marker( + 'parent', 'nul-exit-zero-checkpoint', checkpoint, raw_output), + 'The observer did not checkpoint the valid pre-tail message_end.') + checkpoint_before = checkpoint.read_bytes() + checkpoint_stat = checkpoint.stat() + raw_before = raw_output.read_bytes() + raw_stat = raw_output.stat() + tail = b'\x00' * 1642 + self.action(control, emit_nul_tail=len(tail), exit_after_nul_tail=True) + self.assertEqual(json.loads(control.readline()), {'nul_tail_written':len(tail)}) + output, error = self.finish(observer, ok=False) + self.assertIn('Native output contains an embedded NUL byte', output + error) + self.assertEqual(raw_output.read_bytes(), raw_before + tail, + 'The invalid native tail was altered or discarded.') + self.assertEqual((raw_output.stat().st_ino, raw_output.stat().st_size), + (raw_stat.st_ino, raw_stat.st_size + len(tail))) + self.assertEqual(checkpoint.read_bytes(), checkpoint_before, + 'Invalid output moved or rewrote the last valid observer checkpoint.') + self.assertEqual((checkpoint.stat().st_ino, checkpoint.stat().st_size), + (checkpoint_stat.st_ino, checkpoint_stat.st_size)) + self.assertFalse((attempt / 'acknowledged').exists(), + 'A successful process exit must not acknowledge an invalid raw tail.') + self.assertEqual(self.execution('parent')[:2], ('exited', 'exit 0')) + self.assertIsNone(self.coord('delivery', 'nul-exit-zero-task')['receipt'], + 'Invalid output acknowledged the input owned by the retained attempt.') + self.assertEqual([message['id'] for message in self.coord('inbox', 'parent')], + ['nul-exit-zero-task']) + self.assertEqual(self.coord('player', 'parent')['native'], native_id) + self.assertEqual(self.execution('parent')[2], old_directory, + 'A held invalid tail must remain attached to its original attempt.') + self.assertEqual(self.coord('player', 'parent')['native'], native_id) + + def test_nul_tail_after_keeper_loss_preserves_input_and_starts_fresh_attempt(self): + self.player(harness='omp') + config_path = self.directory / 'fixture.json' + config = json.loads(config_path.read_text()) + config['record_launches'] = True + config_path.write_text(json.dumps(config)) + self.prepare_input('nul-tail-task', 'parent', 'Retain the native attempt across invalid output.', kind='task') + observer = self.spawn(*self.receive_args('parent')) + control, started = self.accept('parent') + native_id = self.eventually(lambda: self.coord('player', 'parent')['native'], + 'The original OMP identity was not recorded.') + self.assertEqual(native_id, started['native']) + with sqlite3.connect(self.db) as database: + old_id, old_directory = database.execute( + 'SELECT id,directory FROM executions WHERE session=?', ('parent',)).fetchone() + attempt = pathlib.Path(old_directory) + checkpoint = attempt / 'checkpoint' + raw_output = attempt / 'stdout' + + self.action(control, emit_checkpoint_message='nul-tail-checkpoint') + self.assertEqual(json.loads(control.readline()), + {'checkpoint_message_written':'nul-tail-checkpoint'}) + self.eventually(lambda: self.checkpoint_covers_output_marker( + 'parent', 'nul-tail-checkpoint', checkpoint, raw_output), + 'The observer did not checkpoint the valid pre-tail message_end.') + checkpoint_before = checkpoint.read_bytes() + checkpoint_stat = checkpoint.stat() + raw_before = raw_output.read_bytes() + raw_stat = raw_output.stat() + tail = b'\x00' * 1642 + self.action(control, emit_nul_tail=len(tail)) + self.assertEqual(json.loads(control.readline()), {'nul_tail_written':len(tail)}) + self.eventually(lambda: raw_output.stat().st_size == raw_stat.st_size + len(tail), + 'The shared owner did not spool the invalid tail before keeper loss.') + owner_command = f'{EXE.resolve()} --instance-owner {self.db.resolve()}' + owners = [process for process in self.owned_processes() + if process['command'] == owner_command] + self.assertEqual(len(owners), 1) + selected_owner = self.selected_process(owners[0]['pid']) + selected_native = self.selected_process(started['pid']) + self.assertIsNotNone(selected_owner) + self.assertIsNotNone(selected_native) + self.signal_selected(selected_owner, signal.SIGKILL) + self.signal_selected(selected_native, signal.SIGKILL) + output, error = self.finish(observer, ok=False) + self.assertIn('Native output contains an embedded NUL byte', output + error) + self.assertEqual(raw_output.read_bytes(), raw_before + tail, + 'The invalid native tail was altered or discarded.') + self.assertEqual((raw_output.stat().st_ino, raw_output.stat().st_size), + (raw_stat.st_ino, raw_stat.st_size + len(tail))) + self.assertEqual(checkpoint.read_bytes(), checkpoint_before, + 'Invalid output moved or rewrote the last valid observer checkpoint.') + self.assertEqual((checkpoint.stat().st_ino, checkpoint.stat().st_size), + (checkpoint_stat.st_ino, checkpoint_stat.st_size)) + self.assertTrue((attempt / 'acknowledged').is_file(), + 'The dead process attempt was not settled for a fresh observer attempt.') + self.assertEqual(self.execution('parent')[:2], ('exited', 'unknown after keeper loss')) + self.assertEqual(self.coord('delivery', 'nul-tail-task')['receipt'], None, + 'Invalid output acknowledged the input owned by the retained attempt.') + self.assertEqual([message['id'] for message in self.coord('inbox', 'parent')], + ['nul-tail-task']) + self.assertEqual(self.coord('player', 'parent')['native'], native_id) + + fresh = self.spawn(*self.receive_args('parent')) + _fresh_control, fresh_started = self.accept('parent') + self.assertEqual(fresh_started['native'], native_id) + with sqlite3.connect(self.db) as database: + new_id, new_directory = database.execute( + 'SELECT id,directory FROM executions WHERE session=?', ('parent',)).fetchone() + self.assertNotEqual(new_id, old_id) + self.assertNotEqual(new_directory, old_directory) + self.assertEqual(self.coord('delivery', 'nul-tail-task')['receipt'], None) + fresh.kill() + fresh.wait() + + def test_output_only_frames_replay_from_durable_checkpoint(self): + self.player(harness='omp') + self.prepare_input('output-replay-task', 'parent', 'Replay output from the last semantic checkpoint.') + observer = self.spawn(*self.receive_args('parent')) + control, started = self.accept('parent') + native_id = self.eventually(lambda: self.coord('player', 'parent')['native'], + 'The original OMP identity was not recorded.') + self.assertEqual(native_id, started['native']) + attempt = pathlib.Path(self.execution('parent')[2]) + checkpoint = attempt / 'checkpoint' + log = self.output_log('parent') + + self.action(control, emit_checkpoint_message='durable-output-cursor') + self.assertEqual(json.loads(control.readline()), + {'checkpoint_message_written': 'durable-output-cursor'}) + def checkpoint_state(): + if not checkpoint.is_file(): + return None + record = checkpoint.read_bytes() + state_at = record.find(b'{"schema":1,') + if state_at < 0: + return None + try: + return record, json.loads(record[state_at:]) + except json.JSONDecodeError: + return None + + def durable_semantic_checkpoint(): + saved = checkpoint_state() + return saved is not None and 'durable-output-cursor' in saved[1].get('last_message', '') + + self.eventually(durable_semantic_checkpoint, + 'The semantic frame was not included in the durable checkpoint.') + durable_checkpoint, durable_state = checkpoint_state() + state_at = durable_checkpoint.find(b'{"schema":1,') + self.assertGreaterEqual(state_at, 0, 'Checkpoint has no schema-1 observer state.') + self.assertIn('durable-output-cursor', durable_state['last_message']) + durable_output = int(durable_state['output']) + + self.action(control, emit_delta_batch={'id': 'uncommitted', 'count': 3, 'bytes': 128}) + self.assertEqual(json.loads(control.readline()), + {'delta_frames_written': 3, 'batch': 'uncommitted'}) + expected_lines = 3 + def output_batch_written(): + if not log.exists(): + return False + with log.open() as stream: + return sum(json.loads(line).get('type') == 'message_update' for line in stream) >= expected_lines + self.eventually(output_batch_written, 'The output-only frames were not written before interruption.') + self.assertEqual(checkpoint.read_bytes(), durable_checkpoint, + 'Output-only changes advanced the durable checkpoint.') + self.assertEqual(int(json.loads(checkpoint.read_bytes()[state_at:])['output']), durable_output) + self.assertGreater(log.stat().st_size, durable_output, + 'Output-only frames did not extend beyond the durable cursor.') + + observer.kill() + observer.wait() + self.assertTrue(any(process['pid'] == started['pid'] for process in self.owned_processes()), + 'The native process did not survive observer recovery.') + self.assertEqual(self.coord('player', 'parent')['native'], native_id) + recovered = self.spawn(*self.receive_args('parent')) + self.assertIsNone(recovered.poll(), 'Same-attempt receiver did not adopt the live native.') + + marker = 'semantic checkpoint after output replay' + self.action(control, emit_checkpoint_message=marker) + self.assertEqual(json.loads(control.readline()), {'checkpoint_message_written': marker}) + self.eventually(lambda: marker.encode() in log.read_bytes(), + 'The recovered observer did not consume frames after its saved cursor.') + def replay_checkpointed(): + saved = checkpoint_state() + return saved is not None and marker in saved[1].get('last_message', '') + + self.eventually(replay_checkpointed, + 'The recovered observer did not checkpoint the next semantic change.') + replay_state = checkpoint_state()[1] + self.assertGreaterEqual(int(replay_state['output']), log.stat().st_size, + 'The recovered checkpoint cursor does not cover replayed output.') + self.assert_no_start() + with log.open() as stream: + updates = [json.loads(line) for line in stream + if json.loads(line).get('type') == 'message_update'] + self.assertEqual(sum(update.get('assistantMessageEvent', {}).get('delta') == 'x' * 128 + for update in updates), 3, + 'Replay duplicated or lost output-only frames.') + self.assertEqual(self.coord('player', 'parent')['native'], native_id) + self.action(control, body='Output replay completed.', hold_exit=True) + self.assertEqual(json.loads(control.readline()), {'terminal_written': True}) + self.action(control, exit_fixture=True) + self.assertEqual(control.readline(), b'') + self.finish(recovered) + + def test_long_omp_stream_releases_frames_and_recovers_checkpoint(self): + self.player(harness='omp') + config_path = self.directory / 'fixture.json' + config = json.loads(config_path.read_text()) + config['record_launches'] = True + config_path.write_text(json.dumps(config)) + self.prepare_input('memory-task', 'parent', 'Retain the original native conversation.', kind='task') + observer = self.spawn(*self.receive_args('parent')) + control, started = self.accept('parent') + native_id = self.eventually(lambda: self.coord('player', 'parent')['native'], + 'The original OMP identity was not recorded.') + self.assertEqual(native_id, started['native']) + attempt = pathlib.Path(self.execution('parent')[2]) + log = self.output_log('parent') + + def rss_bytes(): + status = pathlib.Path(f'/proc/{observer.pid}/status').read_text() + value = next(line.split()[1] for line in status.splitlines() if line.startswith('VmRSS:')) + return int(value) * 1024 + + batch_count = 8 + frames_per_batch = 500 + payload_bytes = 8192 + frame = json.dumps({'type': 'message_update', + 'assistantMessageEvent': {'type': 'text_delta', + 'delta': 'x' * payload_bytes}}) + bytes_per_frame = len(frame.encode()) + 1 + samples = [rss_bytes()] + start_size = log.stat().st_size if log.exists() else 0 + for index in range(batch_count): + self.action(control, emit_delta_batch={'id': index, 'count': frames_per_batch, + 'bytes': payload_bytes}) + self.assertEqual(json.loads(control.readline()), + {'delta_frames_written': frames_per_batch, 'batch': index}) + expected = start_size + (index + 1) * frames_per_batch * bytes_per_frame + deadline = time.monotonic() + 30 + while time.monotonic() < deadline and (not log.exists() or log.stat().st_size < expected): + if observer.poll() is not None: + stdout, stderr = self.child_communication(observer) + self.fail(f'Observer exited during long stream: {stdout}{stderr}') + time.sleep(.05) + self.assertTrue(log.exists() and log.stat().st_size >= expected, + f'Observer did not retain batch {index}') + samples.append(rss_bytes()) + print(json.dumps({'memoryBatch': index, 'rssBytes': samples[-1], + 'rssGrowthBytes': max(samples) - samples[0], + 'retainedLogBytes': log.stat().st_size}), flush=True) + self.assertLess(max(samples) - samples[0], 768 * 1024 * 1024, + 'Observer RSS grew with the emitted frame stream') + + last_half_input = (batch_count // 2) * frames_per_batch * bytes_per_frame + last_half_growth = max(0, samples[batch_count] - samples[batch_count // 2]) + print(json.dumps({'smallDeltaLastHalfInputBytes': last_half_input, + 'smallDeltaLastHalfRssGrowthBytes': last_half_growth}), flush=True) + self.assertLess(last_half_growth, max(4 * 1024 * 1024, last_half_input // 4), + 'Observer RSS retained a linear share of the small-delta stream') + + snapshot_bytes = 0 + for size in (1, 2, 4, 8): + payload = 's' * (size * 1024 * 1024) + snapshot = json.dumps({'type': 'message_update', + 'assistantMessageEvent': {'partial': {'type': 'text', 'text': payload}}, + 'message': {'role': 'assistant', + 'content': [{'type': 'text', 'text': payload}]}}) + self.action(control, emit_snapshot=size * 1024 * 1024) + self.assertEqual(json.loads(control.readline()), + {'snapshot_bytes_written': size * 1024 * 1024}) + snapshot_bytes += len(snapshot.encode()) + expected = start_size + batch_count * frames_per_batch * bytes_per_frame + snapshot_bytes + deadline = time.monotonic() + 30 + while time.monotonic() < deadline and log.stat().st_size < expected: + if observer.poll() is not None: + stdout, stderr = self.child_communication(observer) + self.fail(f'Observer exited during cumulative snapshot: {stdout}{stderr}') + time.sleep(.05) + self.assertGreaterEqual(log.stat().st_size, expected, + f'Observer did not retain the {size} MiB snapshot') + samples.append(rss_bytes()) + print(json.dumps({'cumulativeSnapshotMiB': size, 'rssBytes': samples[-1], + 'rssGrowthBytes': max(samples) - samples[0], + 'retainedLogBytes': log.stat().st_size}), flush=True) + self.assertLess(max(samples) - samples[0], 768 * 1024 * 1024, + 'Observer RSS grew with cumulative snapshot size') + + self.action(control, emit_checkpoint_message='memory-checkpoint') + self.assertEqual(json.loads(control.readline()), + {'checkpoint_message_written': 'memory-checkpoint'}) + checkpoint = attempt / 'checkpoint' + self.eventually(lambda: checkpoint.exists(), 'The long-stream observer did not checkpoint.') + record = checkpoint.read_bytes() + state_at = record.find(b'{"schema":1,') + self.assertGreaterEqual(state_at, 0, 'Checkpoint has no schema-1 observer state.') + state = json.loads(record[state_at:]) + tail = log.read_bytes()[-8192:] + marker_at = tail.rfind(b'memory-checkpoint') + self.assertGreaterEqual(marker_at, 0, 'The checkpoint message is absent from the public log.') + line_end = tail.find(b'\n', marker_at) + self.assertGreaterEqual(line_end, 0) + self.assertGreaterEqual(int(state['output']), log.stat().st_size - len(tail) + line_end + 1, + 'Checkpoint output cursor does not cover the observed message_end.') + self.assertLess(max(samples[-5:]) - min(samples[-5:]), 256 * 1024 * 1024, + 'Observer RSS did not settle over the final batches.') + + observer.kill() + observer.wait() + self.assertTrue(any(process['pid'] == started['pid'] for process in self.owned_processes()), + 'The native process did not survive observer recovery.') + self.assertEqual(self.coord('player', 'parent')['native'], native_id) + marker = 'post-checkpoint output from the original native' + self.action(control, progress=marker) + self.assertEqual(json.loads(control.readline()), {'progress_written': marker}) + + def log_tail_contains(value): + if not log.exists(): + return False + with log.open('rb') as stream: + stream.seek(max(0, log.stat().st_size - 8192)) + return value.encode() in stream.read() + + self.eventually(lambda: log_tail_contains(marker), + 'Recovered observer did not continue from the saved checkpoint.') + self.action(control, body='Long native stream completed after observer recovery.', hold_exit=True) + self.assertEqual(json.loads(control.readline()), {'terminal_written': True}) + self.eventually(lambda: log_tail_contains('agent_end'), + 'Recovered observer did not retain the terminal native event.') + self.action(control, exit_fixture=True) + self.assertEqual(control.readline(), b'') + self.finish(observer, ok=False) + self.eventually(lambda: self.coord('turns', 'parent'), + 'The recovered task completion was not recorded.') + self.assertEqual(self.coord('player', 'parent')['native'], native_id) + self.assertEqual(len((self.directory / 'native-launches.jsonl').read_text().splitlines()), 1) + with log.open() as stream: + observed_updates = sum(1 for line in stream + if json.loads(line).get('type') == 'message_update') + self.assertEqual(observed_updates, batch_count * frames_per_batch + 4) + self.assertEqual(self.coord('delivery', 'memory-task')['receipt'], + 'native-turn:' + self.execution('parent')[0]) + self.shutdown_idle_database_owner('The long-stream receive owner did not exit.') + def test_successful_turn_handles_complete_initial_input_without_explicit_ack(self): """A successful native result settles the complete initial input read by that turn and preserves its native conversation and custody.