diff --git a/dag/extdeps/llm/claude_code_stream_json.dag b/dag/extdeps/llm/claude_code_stream_json.dag index 489451c25c8..a05928e1a00 100644 --- a/dag/extdeps/llm/claude_code_stream_json.dag +++ b/dag/extdeps/llm/claude_code_stream_json.dag @@ -76,6 +76,11 @@ data extdeps_model_scope: ExternalModelScope = ExternalModelScope { } data claude_code_stream_event_type_result: String = "result" +data claude_code_stream_event_type_system: String = "system" +data claude_code_stream_event_type_assistant: String = "assistant" +data claude_code_stream_event_type_user: String = "user" +data claude_code_content_block_type_tool_use: String = "tool_use" +data claude_code_content_block_type_tool_result: String = "tool_result" data claude_code_rate_limit_status_rejected: String = "rejected" // The argv words that make the CLI write this wire. --verbose is required by the CLI for @@ -112,6 +117,9 @@ type ClaudeCodeStreamLineGap type ClaudeCodeStreamLine = ClaudeCodeRateLimitEventLine { info: ClaudeCodeRateLimitInfo } | ClaudeCodeResultLine { result: ClaudeCodeTurnResult } + | ClaudeCodeSystemLine + | ClaudeCodeAssistantLine { has_tool_use: Bool } + | ClaudeCodeUserLine { has_tool_result: Bool } | ClaudeCodeOtherLine { event_type: String } | ClaudeCodeLineUnreadable { gap: ClaudeCodeStreamLineGap } @@ -425,6 +433,40 @@ fn claude_result_line(doc: JsonValue) -> ClaudeCodeStreamLine { } } +fn claude_content_blocks_have_type(content: JsonValue, block_type: String) -> Bool { + match content { + JsonArray { elements } => + any(elements, e => + match claude_string_member(v: e, key: "type") { + StringMemberRead { value: s } => s == block_type + StringMemberNull => false + StringMemberAbsent => false + StringMemberMalformed => false + } + ) + JsonNull => false + JsonBool { value: _ } => false + JsonNumber { lexeme: _ } => false + JsonString { value: _ } => false + JsonObject { members: _ } => false + } +} + +fn claude_message_content_has_type(doc: JsonValue, block_type: String) -> Bool { + match json_object_unique_member(v: doc, key: "message") { + JsonMemberFound { value: msg } => + match json_object_unique_member(v: msg, key: "content") { + JsonMemberFound { value: content } => claude_content_blocks_have_type(content: content, block_type: block_type) + JsonMemberAbsent => false + JsonMemberDuplicated { count: _ } => false + JsonMemberNotAnObject => false + } + JsonMemberAbsent => false + JsonMemberDuplicated { count: _ } => false + JsonMemberNotAnObject => false + } +} + fn claude_code_stream_line(line: String) -> ClaudeCodeStreamLine { match parse_json_document(s: line) { JsonDocumentUnreadable { gap: _ } => ClaudeCodeLineUnreadable { gap: StreamLineNotJson } @@ -442,7 +484,23 @@ fn claude_code_stream_line(line: String) -> ClaudeCodeStreamLine { if t == claude_code_stream_event_type_result { claude_result_line(doc: doc) } else { - ClaudeCodeOtherLine { event_type: t } + if t == claude_code_stream_event_type_system { + ClaudeCodeSystemLine + } else { + if t == claude_code_stream_event_type_assistant { + ClaudeCodeAssistantLine { + has_tool_use: claude_message_content_has_type(doc: doc, block_type: claude_code_content_block_type_tool_use), + } + } else { + if t == claude_code_stream_event_type_user { + ClaudeCodeUserLine { + has_tool_result: claude_message_content_has_type(doc: doc, block_type: claude_code_content_block_type_tool_result), + } + } else { + ClaudeCodeOtherLine { event_type: t } + } + } + } } } StringMemberNull => ClaudeCodeLineUnreadable { gap: StreamLineMemberMissing { member: "type" } } diff --git a/dag/extdeps/llm/cli.dag b/dag/extdeps/llm/cli.dag index 0ee78d5949e..5f72c831235 100644 --- a/dag/extdeps/llm/cli.dag +++ b/dag/extdeps/llm/cli.dag @@ -5,6 +5,7 @@ import extdeps.tools.env { env_path_resolved_program } import std.algebra { trim } import std.types { List, Int, Bool, NonEmptyStr, RenderedTerminalText, Unit } +import extdeps.llm.claude_code_stream_json { claude_code_stream_json_print_args } import extdeps.external_authority { ExternalAuthority } import extdeps.uri { Uri, Https } import extdeps.languages.json.grammar { @@ -201,6 +202,61 @@ fn claude_model_args(selection: ProviderModelSelection) -> List { } } +// HEADLESS PRINT ARGV. Interactive `claude` (shape_claude_invoke_argv) is a session in a terminal: +// it takes --session-id, --settings, --effort and a node name. Automation is a different surface of +// the same binary: `claude -p --output-format stream-json --verbose` writes one JSON object per +// stdout line (extdeps.llm.claude_code_stream_json). The Agent SDK's query() spawns that same +// process; this row is that execution without a Node runtime. --verbose is required by the CLI for +// stream-json under -p. Turn bound is a coproduct so "no --max-turns" cannot be confused with a +// sentinel count. Permission flags and the start prompt stay caller-supplied: skip-permissions is +// gunbc policy (dispatch_claude_permission_args), not an upstream default. + +type ClaudePrintTurnBound + = ClaudePrintUnboundedTurns + | ClaudePrintMaxTurns { turns: Int } + +data claude_cli_max_turns_flag: String = "--max-turns" + +data claude_cli_append_system_prompt_flag: String = "--append-system-prompt" + +data claude_cli_effort_flag: String = "--effort" + +fn claude_print_max_turns_args(bound: ClaudePrintTurnBound) -> List { + match bound { + ClaudePrintUnboundedTurns => [] + ClaudePrintMaxTurns { turns } => [claude_cli_max_turns_flag, to_string(turns)] + } +} + +fn shape_claude_print_argv( + append_system_prompt: String, + start_prompt: String, + permission_args: List, + model: ProviderModelSelection, + effort: ReasoningEffort, + turn_bound: ClaudePrintTurnBound, +) -> List { + concat( + [claude_cli_program], + concat( + claude_code_stream_json_print_args, + concat( + claude_print_max_turns_args(bound: turn_bound), + concat( + [claude_cli_effort_flag, reasoning_effort_label(effort: effort)], + concat( + [claude_cli_append_system_prompt_flag, append_system_prompt], + concat( + claude_model_args(selection: model), + concat(permission_args, [start_prompt]), + ), + ), + ), + ), + ), + ) +} + fn codex_model_args(selection: ProviderModelSelection) -> List { match selection { ProviderAccountDefaultModel => [] @@ -226,7 +282,7 @@ fn shape_claude_invoke_argv( claude_model_args(selection: model), [ "--settings", settings_json, - "--effort", reasoning_effort_label(effort: effort), + claude_cli_effort_flag, reasoning_effort_label(effort: effort), "--session-id", session_id, "-n", node_id, "--append-system-prompt", append_system_prompt, diff --git a/dag/gunbc/claude_code_limit_standing.dag b/dag/gunbc/claude_code_limit_standing.dag index dbbb4a553ab..2e08a4e1e88 100644 --- a/dag/gunbc/claude_code_limit_standing.dag +++ b/dag/gunbc/claude_code_limit_standing.dag @@ -9,6 +9,9 @@ import extdeps.llm.claude_code_stream_json { ClaudeCodeRateLimitEventLine, ClaudeCodeResultLine, ClaudeCodeOtherLine, + ClaudeCodeSystemLine, + ClaudeCodeAssistantLine, + ClaudeCodeUserLine, ClaudeCodeLineUnreadable, ClaudeCodeRateLimitInfo, ClaudeCodeRateLimitWindow, @@ -105,6 +108,12 @@ fn claude_code_trip_reading(stdout: String) -> ClaudeCodeTripReading { ClaudeTripFold { index: acc.index + 1, reading: ClaudeCodeTripRead { result: Present { value: result }, rate_limit: l } } ClaudeCodeOtherLine { event_type: _ } => ClaudeTripFold { index: acc.index + 1, reading: acc.reading } + ClaudeCodeSystemLine => + ClaudeTripFold { index: acc.index + 1, reading: acc.reading } + ClaudeCodeAssistantLine { has_tool_use: _ } => + ClaudeTripFold { index: acc.index + 1, reading: acc.reading } + ClaudeCodeUserLine { has_tool_result: _ } => + ClaudeTripFold { index: acc.index + 1, reading: acc.reading } ClaudeCodeLineUnreadable { gap } => ClaudeTripFold { index: acc.index + 1, reading: ClaudeCodeTripUnreadable { line_index: acc.index, gap: gap } } } diff --git a/dag/gunbc/roadmap/roadmap_dispatch_actuator.dag b/dag/gunbc/roadmap/roadmap_dispatch_actuator.dag index bc8a43a2662..977ccc2fcff 100644 --- a/dag/gunbc/roadmap/roadmap_dispatch_actuator.dag +++ b/dag/gunbc/roadmap/roadmap_dispatch_actuator.dag @@ -94,7 +94,8 @@ import extdeps.tmux { import extdeps.llm.cli { ProviderModelSelection, ProviderModelPinned, ProviderAccountDefaultModel, - shape_claude_invoke_argv, + shape_claude_print_argv, + ClaudePrintUnboundedTurns, claude_cli_program, codex_cli_program, shape_codex_exec_argv, @@ -478,7 +479,8 @@ fn dispatch_actuator_selection_for_provider_on( // DECLARED FRONTIER, NOT CODE: automatic selection between the harness and a vendor is a std.decision // select_realization over funded axes (cost, usage-window headroom, latency, quality). Its trigger is // a grounded funded-axis policy once limit, latency and quality readings exist for at least two -// providers; until then nothing here chooses on the operator's behalf. +// providers; until then nothing here chooses on the operator's behalf. A harness refusal therefore +// stays a refusal. Headless Claude is reached by `?executor=claude`, not by widening the default. type DispatchExecutorRequest = ExecutorDefault | ExecutorVendor { provider: DispatchCliProvider } @@ -2355,19 +2357,15 @@ fn git_worktree_add_detached_argv_for_instance( fn claude_spawn_argv( effort: ReasoningEffort, model: ProviderModelSelection, - session_uuid: String, - node_id: RoadmapNodeId, brief: String, ) -> List { - shape_claude_invoke_argv( - session_id: session_uuid, - node_id: node_id, - effort: effort, - model: model, - settings_json: claude_skip_permission_settings_json, + shape_claude_print_argv( append_system_prompt: brief, start_prompt: dispatch_claude_start_prompt, permission_args: dispatch_claude_permission_args, + model: model, + effort: effort, + turn_bound: ClaudePrintUnboundedTurns, ) } @@ -2505,8 +2503,6 @@ fn dispatch_provider_inner_argv( claude_spawn_argv( effort: effort, model: model, - session_uuid: session_uuid, - node_id: node_id, brief: brief, ), ), @@ -2588,8 +2584,6 @@ fn dispatch_provider_inner_argv_for_instance( claude_spawn_argv( effort: effort, model: model, - session_uuid: session_uuid, - node_id: node_id, brief: brief, ), ), @@ -2855,12 +2849,30 @@ fn dispatch_worker_unit_properties( granted: Kibibyte, process_bounds: UnitProcessBounds, exit_hook: NonEmptyStr, +) -> List { + dispatch_worker_unit_stream_properties( + instance: instance, + working_directory: working_directory, + stdout_path: worker_log_path, + stderr_path: worker_log_path, + granted: granted, + process_bounds: process_bounds, + ) +} + +fn dispatch_worker_unit_stream_properties( + instance: HostDashboardInstance, + working_directory: String, + stdout_path: String, + stderr_path: String, + granted: Kibibyte, + process_bounds: UnitProcessBounds, ) -> List { concat( compute_grant_unit_properties(granted: granted, working_directory: working_directory, process_bounds: Present { value: process_bounds }), [ - SystemdRunProperty { property: StandardOutputProperty, value: join(["append:", worker_log_path], "") as NonEmptyStr }, - SystemdRunProperty { property: StandardErrorProperty, value: join(["append:", worker_log_path], "") as NonEmptyStr }, + SystemdRunProperty { property: StandardOutputProperty, value: join(["append:", stdout_path], "") as NonEmptyStr }, + SystemdRunProperty { property: StandardErrorProperty, value: join(["append:", stderr_path], "") as NonEmptyStr }, SystemdRunProperty { property: ProtectSystemProperty, value: "strict" as NonEmptyStr }, SystemdRunProperty { property: ProtectHomeProperty, value: "read-only" as NonEmptyStr }, SystemdRunProperty { property: PrivateTmpProperty, value: "yes" as NonEmptyStr }, @@ -2909,16 +2921,18 @@ type DispatchWorkerUnitPlan { systemd_run_program: FilePath unit: NonEmptyStr working_directory: String - worker_log_path: String + stdout_path: String + stderr_path: String inner_argv: List exit_hook: NonEmptyStr } fn dispatch_worker_unit_properties_for_plan(instance: HostDashboardInstance, plan: DispatchWorkerUnitPlan, granted: Kibibyte, process_bounds: UnitProcessBounds) -> List { - dispatch_worker_unit_properties( + dispatch_worker_unit_stream_properties( instance: instance, working_directory: plan.working_directory, - worker_log_path: plan.worker_log_path, + stdout_path: plan.stdout_path, + stderr_path: plan.stderr_path, granted: granted, process_bounds: process_bounds, exit_hook: plan.exit_hook, @@ -2947,80 +2961,139 @@ type DispatchSessionContainerPlan // not be a degraded success, it would be a guaranteed HostBudgetUnreadable wearing the shape of a // spawn that worked. macbook_local_toolchain answers `none` here because macOS has no systemd at // all, which is why the field is optional rather than a path every toolchain must invent. -fn dispatch_session_container_for_provider( - provider: DispatchCliProvider, +fn dispatch_systemd_unit_container_plan( instance: HostDashboardInstance, node_id: String, attempt_key: String, + pane_working_directory: String, + stdout_path: String, + stderr_path: String, + provider_inner: List, + unavailable_step: NonEmptyStr, + unavailable_why: String, +) -> DispatchSessionContainerPlan { + match instance.toolchain.systemd_run { + Absent => + DispatchSessionContainerUnavailable { + step: unavailable_step, + detail: join([ + unavailable_why, + ", and the toolchain for host ", instance.host_identity as String, + " declares no systemd-run", + ], ""), + } + Present { value: program } => + match attempt_exit_hook(instance_root: instance.instance_root as String, node_id: node_id, attempt_key: attempt_key, role: AttemptWorkerRole) { + AttemptExitHookRefused { reason } => + DispatchSessionContainerUnavailable { step: "worker-exit-hook" as NonEmptyStr, detail: reason as String } + AttemptExitHookRendered { exec_line } => + DispatchSessionContainerReady { + container: SystemdUnitContainer { + plan: DispatchWorkerUnitPlan { + systemd_run_program: program, + unit: dispatch_attempt_unit_name(node_id: node_id, attempt_key: attempt_key), + working_directory: pane_working_directory, + stdout_path: stdout_path, + stderr_path: stderr_path, + inner_argv: provider_inner, + exit_hook: exec_line, + }, + }, + } + } + } +} + +fn dispatch_tmux_session_container_plan( + instance: HostDashboardInstance, session_name: String, events_path: String, - worker_log_path: String, pane_working_directory: String, - tmux_spawn: HostExecArgv, - pipe_command: TmuxPipeCommand, + worktree_path: String, provider_inner: List, ) -> DispatchSessionContainerPlan { - match provider { - GunbcHarnessProvider => - match instance.toolchain.systemd_run { - Absent => - DispatchSessionContainerUnavailable { - step: "harness-transient-unit" as NonEmptyStr, - detail: join([ - "the gunbc harness worker requires a transient unit for an enforceable memory bound, ", - "and the toolchain for host ", instance.host_identity as String, - " declares no systemd-run", - ], ""), - } - Present { value: program } => - match attempt_exit_hook(instance_root: instance.instance_root as String, node_id: node_id, attempt_key: attempt_key, role: AttemptWorkerRole) { - AttemptExitHookRefused { reason } => - DispatchSessionContainerUnavailable { step: "worker-exit-hook" as NonEmptyStr, detail: reason as String } - AttemptExitHookRendered { exec_line } => - DispatchSessionContainerReady { - container: SystemdUnitContainer { - plan: DispatchWorkerUnitPlan { - systemd_run_program: program, - unit: dispatch_attempt_unit_name(node_id: node_id, attempt_key: attempt_key), - working_directory: pane_working_directory, - worker_log_path: worker_log_path, - inner_argv: provider_inner, - exit_hook: exec_line, - }, - }, - } - } + match dispatch_attempt_pipe_emission_for_instance( + instance: instance, + events_path: events_path, + ) { + PipeEmitRefused { detail } => + DispatchSessionContainerUnavailable { + step: "tmux-event-pipe-emit" as NonEmptyStr, + detail: detail, } - ClaudeCodeProvider => + PipeEmitted { command } => DispatchSessionContainerReady { container: dispatch_tmux_container( instance: instance, session_name: session_name, - tmux_spawn: tmux_spawn, - pipe_command: pipe_command, + tmux_spawn: tmux_idle_spawn_argv_for_instance( + instance: instance, + session_name: session_name, + worktree_path: worktree_path, + pane_working_directory: pane_working_directory, + ), + pipe_command: command, provider_inner: provider_inner, ), } + } +} + +fn dispatch_session_container_for_provider( + provider: DispatchCliProvider, + instance: HostDashboardInstance, + node_id: String, + attempt_key: String, + session_name: String, + events_path: String, + worker_log_path: String, + pane_working_directory: String, + worktree_path: String, + provider_inner: List, +) -> DispatchSessionContainerPlan { + match provider { + GunbcHarnessProvider => + dispatch_systemd_unit_container_plan( + instance: instance, + node_id: node_id, + attempt_key: attempt_key, + pane_working_directory: pane_working_directory, + stdout_path: worker_log_path, + stderr_path: worker_log_path, + provider_inner: provider_inner, + unavailable_step: "harness-transient-unit" as NonEmptyStr, + unavailable_why: "the gunbc harness worker requires a transient unit for an enforceable memory bound", + ) + ClaudeCodeProvider => + dispatch_systemd_unit_container_plan( + instance: instance, + node_id: node_id, + attempt_key: attempt_key, + pane_working_directory: pane_working_directory, + stdout_path: events_path, + stderr_path: worker_log_path, + provider_inner: provider_inner, + unavailable_step: "claude-transient-unit" as NonEmptyStr, + unavailable_why: "headless Claude Code requires a transient unit so stream-json stdout is a unit property and the session placement grant is the cgroup bound", + ) CodexDispatchProvider => - DispatchSessionContainerReady { - container: dispatch_tmux_container( - instance: instance, - session_name: session_name, - tmux_spawn: tmux_spawn, - pipe_command: pipe_command, - provider_inner: provider_inner, - ), - } + dispatch_tmux_session_container_plan( + instance: instance, + session_name: session_name, + events_path: events_path, + pane_working_directory: pane_working_directory, + worktree_path: worktree_path, + provider_inner: provider_inner, + ) CursorDispatchProvider => - DispatchSessionContainerReady { - container: dispatch_tmux_container( - instance: instance, - session_name: session_name, - tmux_spawn: tmux_spawn, - pipe_command: pipe_command, - provider_inner: provider_inner, - ), - } + dispatch_tmux_session_container_plan( + instance: instance, + session_name: session_name, + events_path: events_path, + pane_working_directory: pane_working_directory, + worktree_path: worktree_path, + provider_inner: provider_inner, + ) } } @@ -3488,69 +3561,47 @@ fn dispatch_spawn_commands_for_resolved_instance( node_id: node.node, attempt_key: attempt_key, ) - let tmux = tmux_idle_spawn_argv_for_instance( - instance: instance, - session_name: session_name, - worktree_path: worktree_path, - pane_working_directory: dispatch_pane_working_directory( - provider: provider, - instance: instance, - worktree_path: worktree_path, - ), - ) let events_path = dispatch_attempt_events_path_for_instance( instance: instance, node_id: node.node, attempt_key: attempt_key, ) - match dispatch_attempt_pipe_emission_for_instance( + match dispatch_session_container_for_provider( + provider: provider, instance: instance, + node_id: node.node, + attempt_key: attempt_key, + session_name: session_name, events_path: events_path, + worker_log_path: dispatch_attempt_worker_log_path_for_instance( + instance: instance, + node_id: node.node, + attempt_key: attempt_key, + ), + pane_working_directory: dispatch_pane_working_directory( + provider: provider, + instance: instance, + worktree_path: worktree_path, + ), + worktree_path: worktree_path, + provider_inner: provider_inner, ) { - PipeEmitRefused { detail } => - DispatchSpawnRefused { - step: "tmux-event-pipe-emit" as NonEmptyStr, - detail: detail, - } - PipeEmitted { command } => - match dispatch_session_container_for_provider( - provider: provider, - instance: instance, - node_id: node.node, - attempt_key: attempt_key, - session_name: session_name, - events_path: events_path, - worker_log_path: dispatch_attempt_worker_log_path_for_instance( - instance: instance, - node_id: node.node, - attempt_key: attempt_key, - ), - pane_working_directory: dispatch_pane_working_directory( - provider: provider, - instance: instance, - worktree_path: worktree_path, - ), - tmux_spawn: tmux, - pipe_command: command, - provider_inner: provider_inner, - ) { - DispatchSessionContainerUnavailable { step: st, detail: d } => - DispatchSpawnRefused { step: st, detail: d } - DispatchSessionContainerReady { container: c } => - DispatchSpawnReady { - commands: DispatchSpawnCommands { - worktree_add: worktree, - attempt_state_prepare: state_prepare, - attempt_current_pointer: attempt_current_pointer_argv_for_instance( - instance: instance, - node_id: node.node, - attempt_key: attempt_key, - ), - session_container: c, - }, - } + DispatchSessionContainerUnavailable { step: st, detail: d } => + DispatchSpawnRefused { step: st, detail: d } + DispatchSessionContainerReady { container: c } => + DispatchSpawnReady { + commands: DispatchSpawnCommands { + worktree_add: worktree, + attempt_state_prepare: state_prepare, + attempt_current_pointer: attempt_current_pointer_argv_for_instance( + instance: instance, + node_id: node.node, + attempt_key: attempt_key, + ), + session_container: c, + }, } - } + } } } } diff --git a/dag/gunbc/roadmap/roadmap_provider_events.dag b/dag/gunbc/roadmap/roadmap_provider_events.dag index aa3ecbc2d71..0bd82f6da62 100644 --- a/dag/gunbc/roadmap/roadmap_provider_events.dag +++ b/dag/gunbc/roadmap/roadmap_provider_events.dag @@ -24,6 +24,17 @@ import extdeps.languages.json.parse { json_text_parseable, parse_json_document, JsonDocumentParse, JsonDocumentParsed, JsonDocumentUnreadable, json_field_string, json_field_int, JsonFieldRead, FieldRead, FieldAbsent, FieldMalformed, } +import extdeps.llm.claude_code_stream_json { + ClaudeCodeStreamLine, + ClaudeCodeRateLimitEventLine, + ClaudeCodeResultLine, + ClaudeCodeSystemLine, + ClaudeCodeAssistantLine, + ClaudeCodeUserLine, + ClaudeCodeOtherLine, + ClaudeCodeLineUnreadable, + claude_code_stream_line, +} // THE HARNESS ARMS ARE THE READ SIDE OF gunbc.harness.harness_tool_kind. A harness tool use is // written with an item type derived from its HarnessTool, and codex_item_activity inverts that same @@ -160,7 +171,11 @@ data codex_provider_event_projection_filter: NonEmptyStr = join([ to_string(character_count_value(c: codex_provider_projection_message_budget)), "]),\"truncated\":((.item.text | length) > ", to_string(character_count_value(c: codex_provider_projection_message_budget)), - ")} else null end; (event_type) as $type | if $type == \"item.started\" or $type == \"item.completed\" then (if item_type == \"agent_message\" and (agent_report != null) then {\"type\":$type,\"item\":{\"type\":item_type,\"message\":agent_report}} else {\"type\":$type,\"item\":{\"type\":item_type}} end) elif $type == \"turn.failed\" then {\"type\":$type,\"error\":{\"message\":(.error | bounded_text)}} elif $type == \"turn.budget_exhausted\" then {\"type\":$type,\"steps_taken\":(.steps_taken | num)} elif $type == \"turn.placement_waiting\" then {\"type\":$type,\"wait\":(.wait | num),\"wait_cap\":(.wait_cap | num),\"reason\":(.reason | bounded_text)} elif $type == \"error\" then {\"type\":$type,\"message\":((.message // .error) | bounded_text)} elif $type == \"round.usage\" then ({\"type\":$type,\"step\":(.step | num),\"input_tokens\":(.input_tokens | num),\"output_tokens\":(.output_tokens | num),\"epoch_ms\":(.epoch_ms | num)} + (if (.files_changed | type) == \"number\" then {\"files_changed\":.files_changed} else {} end) + (if (.tool_calls | type) == \"number\" then {\"tool_calls\":.tool_calls} else {} end)) else {\"type\":$type} end", + ")} else null end; (event_type) as $type | if $type == \"item.started\" or $type == \"item.completed\" then (if item_type == \"agent_message\" and (agent_report != null) then {\"type\":$type,\"item\":{\"type\":item_type,\"message\":agent_report}} else {\"type\":$type,\"item\":{\"type\":item_type}} end) elif $type == \"turn.failed\" then {\"type\":$type,\"error\":{\"message\":(.error | bounded_text)}} elif $type == \"turn.budget_exhausted\" then {\"type\":$type,\"steps_taken\":(.steps_taken | num)} elif $type == \"turn.placement_waiting\" then {\"type\":$type,\"wait\":(.wait | num),\"wait_cap\":(.wait_cap | num),\"reason\":(.reason | bounded_text)} elif $type == \"error\" then {\"type\":$type,\"message\":((.message // .error) | bounded_text)} elif $type == \"round.usage\" then ({\"type\":$type,\"step\":(.step | num),\"input_tokens\":(.input_tokens | num),\"output_tokens\":(.output_tokens | num),\"epoch_ms\":(.epoch_ms | num)} + (if (.files_changed | type) == \"number\" then {\"files_changed\":.files_changed} else {} end) + (if (.tool_calls | type) == \"number\" then {\"tool_calls\":.tool_calls} else {} end)) elif $type == \"result\" then {\"type\":$type,\"subtype\":(if (.subtype | type) == \"string\" then .subtype[0:", + to_string(character_count_value(c: codex_provider_projection_type_budget)), + "] else \"unknown\" end),\"is_error\":(if (.is_error | type) == \"boolean\" then .is_error else true end)} elif $type == \"system\" then {\"type\":$type} elif $type == \"assistant\" or $type == \"user\" then {\"type\":$type,\"message\":{\"content\":(if (.message.content | type) == \"array\" then [.message.content[] | {\"type\":(if (.type | type) == \"string\" then .type[0:", + to_string(character_count_value(c: codex_provider_projection_type_budget)), + "] else \"unknown\" end)}] else [] end)}} elif $type == \"rate_limit_event\" then empty else {\"type\":$type} end", ], "") as NonEmptyStr fn codex_provider_event_projection_args(path: FilePath) -> List { @@ -172,6 +187,33 @@ fn codex_provider_event_projection_args(path: FilePath) -> List { ] } +fn claude_code_line_codex_kind(line: ClaudeCodeStreamLine) -> CodexProviderEventKind? { + match line { + ClaudeCodeSystemLine => Present { value: CodexThreadStarted } + ClaudeCodeAssistantLine { has_tool_use } => + if has_tool_use { + Present { value: CodexItemStarted { activity: CodexRunningCommand } } + } else { + Present { value: CodexItemCompleted { activity: CodexReporting } } + } + ClaudeCodeUserLine { has_tool_result } => + if has_tool_result { + Present { value: CodexItemCompleted { activity: CodexRunningCommand } } + } else { + none + } + ClaudeCodeResultLine { result } => + if result.is_error { + Present { value: CodexTurnFailed } + } else { + Present { value: CodexTurnCompleted } + } + ClaudeCodeRateLimitEventLine { info: _ } => none + ClaudeCodeOtherLine { event_type: _ } => none + ClaudeCodeLineUnreadable { gap: _ } => none + } +} + // The attempt-owned provider-events.jsonl remains the complete Codex protocol authority. Workflow // observation first runs a separately modeled jq capability over that file. `--raw-input` plus // `fromjson` proves exactly one JSON value per physical JSONL line before emitting only a bounded @@ -183,7 +225,7 @@ fn codex_provider_event_projection_args(path: FilePath) -> List { // visible as an observation refusal for retained attempts. A malformed raw event makes jq nonzero, // so no parsed prefix is accepted as a complete observation. -data codex_projection_filter_scaffold_note: String = "DECLARED SCAFFOLD (review 44058, DESIGN 6). codex_provider_event_projection_filter is a jq PROGRAM assembled by join — a foreign language built as a string, the same class the tmux pipe payload was in before it moved onto the bash backend (gunbc.dispatch_pipe_pane_emit). It is marked rather than fixed because the two available routes are not equivalent: minting a jq grammar to read backward would buy a modeled emitter for a filter that exists ONLY to bound a line before the substrate reads it, which is the purity trap DESIGN 6 names — the displaced cost is zero once the read itself is bounded. DISSOLVES ON: the observation reading the attempt JSONL directly under a substrate-side line bound instead of shelling to a projector — the witness-realization lane's typed file read (docs/plans/witness-realization-plan.md), at which point BOTH the filter string and the ProviderEventProjectionCapability jq dependency delete together. Until then the interpolated values are the two bounds above (both Int data rows), never caller input, and the classifier downstream is anchored and total, so a projector drift refuses rather than reclassifying." +data codex_projection_filter_scaffold_note: String = "DECLARED SCAFFOLD (review 44058, DESIGN 6). codex_provider_event_projection_filter is a jq PROGRAM assembled by join — a foreign language built as a string, the same class the tmux pipe payload was in before it moved onto the bash backend (gunbc.dispatch_pipe_pane_emit). It is marked rather than fixed because the two available routes are not equivalent: minting a jq grammar to read backward would buy a modeled emitter for a filter that exists ONLY to bound a line before the substrate reads it, which is the purity trap DESIGN 6 names — the displaced cost is zero once the read itself is bounded. DISSOLVES ON: the observation reading the attempt JSONL directly under a substrate-side line bound instead of shelling to a projector — the witness-realization lane's typed file read (docs/plans/witness-realization-plan.md), at which point BOTH the filter string and the ProviderEventProjectionCapability jq dependency delete together. Until then the interpolated values are the two bounds above (both Int data rows), never caller input, and the classifier downstream is anchored and total, so a projector drift refuses rather than reclassifying. Claude Code stream-json is bounded here (type, is_error, content-block types) and classified by claude_code_line_codex_kind from parse_codex_jsonl; the jq does not mint Codex event types for Claude lines." // Codex --json is the provider protocol authority: one JSON object per line with thread.started, // turn.started, item.*, turn.completed, turn.budget_exhausted, turn.failed, and error event types, @@ -467,7 +509,15 @@ fn codex_event_from_complete_line( kind: CodexError, } } } else { - none + match claude_code_line_codex_kind(line: claude_code_stream_line(line: raw)) { + Present { value: kind } => + Present { value: CodexProviderEvent { + line_number: line_number, + raw: raw, + kind: kind, + } } + Absent => none + } } } diff --git a/dag/test/claim/claude_code_dispatch_witness_test.dag b/dag/test/claim/claude_code_dispatch_witness_test.dag index e7b780b0886..5a59f9201ba 100644 --- a/dag/test/claim/claude_code_dispatch_witness_test.dag +++ b/dag/test/claim/claude_code_dispatch_witness_test.dag @@ -11,6 +11,9 @@ import extdeps.llm.claude_code_stream_json { ClaudeCodeRateLimitEventLine, ClaudeCodeResultLine, ClaudeCodeOtherLine, + ClaudeCodeSystemLine, + ClaudeCodeAssistantLine, + ClaudeCodeUserLine, ClaudeCodeLineUnreadable, StreamLineNotJson, StreamLineMemberMissing, @@ -169,6 +172,9 @@ test fn the_captured_rate_limit_line_decodes_both_windows_in_basis_points() -> B && any(info.windows, w => w.name as String == "seven_day" && basis_point_count(bp: w.utilization) == 4500) ClaudeCodeResultLine { result: _ } => false ClaudeCodeOtherLine { event_type: _ } => false + ClaudeCodeSystemLine => false + ClaudeCodeAssistantLine { has_tool_use: _ } => false + ClaudeCodeUserLine { has_tool_result: _ } => false ClaudeCodeLineUnreadable { gap: _ } => false } } @@ -184,6 +190,9 @@ test fn the_unauthorized_result_carries_is_error_and_numeric_status() -> Bool { } ClaudeCodeRateLimitEventLine { info: _ } => false ClaudeCodeOtherLine { event_type: _ } => false + ClaudeCodeSystemLine => false + ClaudeCodeAssistantLine { has_tool_use: _ } => false + ClaudeCodeUserLine { has_tool_result: _ } => false ClaudeCodeLineUnreadable { gap: _ } => false } } @@ -199,6 +208,9 @@ test fn a_string_utilization_refuses_naming_the_member() -> Bool { ClaudeCodeRateLimitEventLine { info: _ } => false ClaudeCodeResultLine { result: _ } => false ClaudeCodeOtherLine { event_type: _ } => false + ClaudeCodeSystemLine => false + ClaudeCodeAssistantLine { has_tool_use: _ } => false + ClaudeCodeUserLine { has_tool_result: _ } => false } } @@ -219,6 +231,9 @@ fn refuses_naming(line: String, member: String) -> Bool { ClaudeCodeRateLimitEventLine { info: _ } => false ClaudeCodeResultLine { result: _ } => false ClaudeCodeOtherLine { event_type: _ } => false + ClaudeCodeSystemLine => false + ClaudeCodeAssistantLine { has_tool_use: _ } => false + ClaudeCodeUserLine { has_tool_result: _ } => false } } diff --git a/dag/test/claim/extdeps_llm_claude_trust_witness_test.dag b/dag/test/claim/extdeps_llm_claude_trust_witness_test.dag index 017f1256bbb..2fd3438724c 100644 --- a/dag/test/claim/extdeps_llm_claude_trust_witness_test.dag +++ b/dag/test/claim/extdeps_llm_claude_trust_witness_test.dag @@ -6,6 +6,9 @@ import extdeps.llm.cli { claude_trust_store_path, claude_workspace_trust_seeded, shape_claude_invoke_argv, + shape_claude_print_argv, + ClaudePrintUnboundedTurns, + ClaudePrintMaxTurns, ProviderAccountDefaultModel, ReasoningHigh, } @@ -78,3 +81,33 @@ test fn w_argv_head_is_program_authority() -> Bool { } } +test fn w_print_argv_is_stream_json_headless() -> Bool { + let argv = shape_claude_print_argv( + append_system_prompt: "brief", + start_prompt: "go", + permission_args: ["--dangerously-skip-permissions"], + model: ProviderAccountDefaultModel, + effort: ReasoningHigh, + turn_bound: ClaudePrintMaxTurns { turns: 1 }, + ) + let wire = join(argv, separator: "\0") + match argv.first() { + Present { value: head } => head == claude_cli_program + Absent => false + } + && string_contains(s: wire, pattern: "\0-p\0--output-format\0stream-json\0--verbose") + && string_contains(s: wire, pattern: "\0--max-turns\01\0") + && string_contains(s: wire, pattern: "\0--effort\0high\0") + && string_contains(s: wire, pattern: "\0--append-system-prompt\0brief\0") + && string_contains(s: wire, pattern: "\0--dangerously-skip-permissions\0go") + && !string_contains(s: wire, pattern: "--session-id") + && !string_contains(s: join(shape_claude_print_argv( + append_system_prompt: "brief", + start_prompt: "go", + permission_args: [], + model: ProviderAccountDefaultModel, + effort: ReasoningHigh, + turn_bound: ClaudePrintUnboundedTurns, + ), separator: "\0"), pattern: "--max-turns") +} + diff --git a/dag/test/claim/roadmap/roadmap_dispatch_actuator_witness_test.dag b/dag/test/claim/roadmap/roadmap_dispatch_actuator_witness_test.dag index 32867dd4f52..befdb13f74c 100644 --- a/dag/test/claim/roadmap/roadmap_dispatch_actuator_witness_test.dag +++ b/dag/test/claim/roadmap/roadmap_dispatch_actuator_witness_test.dag @@ -64,6 +64,9 @@ import gunbc.roadmap_dispatch_actuator { dispatch_supervisor_unit_command, dispatch_attempt_audit_events_path_for_instance, dispatch_attempt_supervisor_events_path_for_instance, + dispatch_attempt_events_path_for_instance, + dispatch_attempt_worker_log_path_for_instance, + dispatch_attempt_unit_name, attempt_state_prepare_argv_for_instance, dispatch_reviewer_attempt_identity, dispatch_attempt_publication_dir_for_instance, @@ -110,6 +113,9 @@ import gunbc.roadmap_dispatch_actuator { HostExecArgv, SystemdUnitContainer, TmuxSessionContainer, + dispatch_session_container_for_provider, + DispatchSessionContainerReady, + DispatchSessionContainerUnavailable, DispatchSpawnReady, DispatchSpawnRefused, git_worktree_add_argv, @@ -304,13 +310,13 @@ test fn witness_claude_argv_uses_dangerous_skip_permissions() -> Bool { let argv = claude_spawn_argv( effort: ReasoningHigh, model: ProviderAccountDefaultModel, - session_uuid: "uuid-1", - node_id: roadmap_dispatch_actuator_nid(s: "node-1"), brief: "brief body", ) string_contains(s: join(argv, separator: "\0"), pattern: "--dangerously-skip-permissions") - && string_contains(s: join(argv, separator: "\0"), pattern: "skipDangerousModePermissionPrompt") - && string_contains(s: join(argv, separator: "\0"), pattern: "--session-id") + && string_contains(s: join(argv, separator: "\0"), pattern: "-p") + && string_contains(s: join(argv, separator: "\0"), pattern: "stream-json") + && string_contains(s: join(argv, separator: "\0"), pattern: "--effort") + && !string_contains(s: join(argv, separator: "\0"), pattern: "--session-id") && !string_contains(s: join(argv, separator: "\0"), pattern: "--resume") } @@ -328,8 +334,6 @@ test fn witness_tmux_spawn_argv_threads_claude_argv() -> Bool { let claude = claude_spawn_argv( effort: ReasoningMedium, model: ProviderAccountDefaultModel, - session_uuid: "uuid-2", - node_id: roadmap_dispatch_actuator_nid(s: "node-2"), brief: "brief", ) let cmd = tmux_spawn_argv( @@ -925,8 +929,6 @@ test fn witness_claude_invoke_permission_args_are_gunbc_policy() -> Bool { let argv = claude_spawn_argv( effort: ReasoningHigh, model: ProviderAccountDefaultModel, - session_uuid: "uuid-policy", - node_id: roadmap_dispatch_actuator_nid(s: "node-policy"), brief: "brief", ) string_contains(s: join(argv, separator: "\0"), pattern: "--dangerously-skip-permissions") @@ -1074,8 +1076,6 @@ test fn witness_pinned_model_reaches_claude_argv() -> Bool { let argv = claude_spawn_argv( effort: ReasoningHigh, model: ProviderModelPinned { model: witness_pinned_dispatch_model }, - session_uuid: "uuid-pin", - node_id: roadmap_dispatch_actuator_nid(s: "node-pin"), brief: "brief", ) string_contains( @@ -1088,8 +1088,6 @@ test fn witness_claude_account_default_passes_no_model_flag() -> Bool { let argv = claude_spawn_argv( effort: ReasoningHigh, model: ProviderAccountDefaultModel, - session_uuid: "uuid-default", - node_id: roadmap_dispatch_actuator_nid(s: "node-default"), brief: "brief", ) !string_contains(s: join(argv, separator: "\0"), pattern: "--model") @@ -1518,3 +1516,66 @@ test fn witness_supervisor_launch_writes_the_events_path_metering_reads() -> Boo ) } } + +test fn witness_headless_claude_uses_a_transient_unit_with_stream_json_on_the_events_path() -> Bool { + let instance = srv1_live_dashboard_instance() + let node_id = "claude-headless" + let attempt_key = "feedfacefeedface" + let events = dispatch_attempt_events_path_for_instance( + instance: instance, + node_id: roadmap_dispatch_actuator_nid(s: node_id), + attempt_key: attempt_key, + ) + let log = dispatch_attempt_worker_log_path_for_instance( + instance: instance, + node_id: roadmap_dispatch_actuator_nid(s: node_id), + attempt_key: attempt_key, + ) + match dispatch_session_container_for_provider( + provider: ClaudeCodeProvider, + instance: instance, + node_id: node_id, + attempt_key: attempt_key, + session_name: "unused", + events_path: events, + worker_log_path: log, + pane_working_directory: "/wt", + worktree_path: "/wt", + provider_inner: ["claude", "-p"], + ) { + DispatchSessionContainerUnavailable { step: _, detail: _ } => false + DispatchSessionContainerReady { container } => + match container { + TmuxSessionContainer { spawn: _, default_shell: _, remain_on_exit: _, event_pipe: _, provider_start: _ } => false + SystemdUnitContainer { plan } => + plan.unit == dispatch_attempt_unit_name(node_id: node_id, attempt_key: attempt_key) + && plan.working_directory == "/wt" + && plan.stdout_path == events + && plan.stderr_path == log + } + } +} + +// review 78389. The spawn fold used to emit the tmux tee/pipe payload before choosing a container, so +// a Claude (systemd stdout) refusal was named tmux-event-pipe-emit. Pipe emission now lives only on +// the tmux container arm; this spawn is the real producer, and a pipe-step refusal here is the defect. +test fn witness_headless_claude_spawn_does_not_refuse_as_tmux_event_pipe() -> Bool { + match dispatch_spawn_commands_for_resolved_instance( + instance: srv1_live_dashboard_instance(), + node: sized_derivable(id: "claude-headless-spawn", prs: [], title: "Headless Claude spawn"), + session_uuid: "00000000-0000-4000-8000-000000000005", + attempt_key: "feedfacefeedface", + provider: ClaudeCodeProvider, + effort: ReasoningMedium, + model: ProviderAccountDefaultModel, + origin: FreshFromBase, + resolution: fixture_ready_resolution(node_id: "claude-headless-spawn", project_ids: []), + ) { + DispatchSpawnRefused { step, detail: _ } => (step as String) != "tmux-event-pipe-emit" + DispatchSpawnReady { commands } => + match commands.session_container { + SystemdUnitContainer { plan: _ } => true + TmuxSessionContainer { spawn: _, default_shell: _, remain_on_exit: _, event_pipe: _, provider_start: _ } => false + } + } +} diff --git a/dag/test/claim/roadmap/roadmap_provider_events_witness_test.dag b/dag/test/claim/roadmap/roadmap_provider_events_witness_test.dag index a75a6e4ef63..c6c90766d05 100644 --- a/dag/test/claim/roadmap/roadmap_provider_events_witness_test.dag +++ b/dag/test/claim/roadmap/roadmap_provider_events_witness_test.dag @@ -8,6 +8,8 @@ import gunbc.harness.harness_tool_kind { } import std.measure { character_count, character_count_value } import gunbc.harness.harness_turn { harness_event_placement_waiting, harness_tool_round_events } +import extdeps.llm.claude_code_stream_json { claude_code_stream_line } +import std.optional { Present, Absent } import gunbc.roadmap_provider_events { CodexItemActivity, CodexRunningCommand, CodexEditingFiles, CodexUsingConnectedTool, CodexSearching, CodexUpdatingPlan, CodexReasoningItem, CodexReporting, HarnessReadingFile, HarnessWritingFile, @@ -23,6 +25,12 @@ import gunbc.roadmap_provider_events { ProviderEventsRefused, parse_codex_jsonl, parse_codex_jsonl_with_line_budget, + claude_code_line_codex_kind, + CodexThreadStarted, + CodexTurnCompleted, + CodexTurnFailed, + CodexItemStarted, + CodexRunningCommand, codex_provider_event_line_observation_budget, codex_provider_projection_type_budget, codex_provider_projection_detail_budget, @@ -436,6 +444,65 @@ test fn an_undeclared_tool_round_is_read_back_as_undeclared_not_as_a_command() - tool_use_round_tags(name: "delete_everything") == "started:undeclared,completed:undeclared" } +test fn claude_stream_json_projects_into_the_codex_event_kinds_the_belt_reads() -> Bool { + let system_ok = match claude_code_line_codex_kind(line: claude_code_stream_line(line: "{\"type\":\"system\",\"subtype\":\"init\"}")) { + Present { value: kind } => + match kind { + CodexThreadStarted => true + _ => false + } + Absent => false + } + let tool_ok = match claude_code_line_codex_kind(line: claude_code_stream_line(line: "{\"type\":\"assistant\",\"message\":{\"content\":[{\"type\":\"tool_use\",\"name\":\"Bash\"}]}}")) { + Present { value: kind } => + match kind { + CodexItemStarted { activity } => + match activity { + CodexRunningCommand => true + _ => false + } + _ => false + } + Absent => false + } + let result_ok = match claude_code_line_codex_kind(line: claude_code_stream_line(line: "{\"type\":\"result\",\"subtype\":\"success\",\"is_error\":false}")) { + Present { value: kind } => + match kind { + CodexTurnCompleted => true + _ => false + } + Absent => false + } + let fail_ok = match claude_code_line_codex_kind(line: claude_code_stream_line(line: "{\"type\":\"result\",\"subtype\":\"success\",\"is_error\":true,\"api_error_status\":401}")) { + Present { value: kind } => + match kind { + CodexTurnFailed => true + _ => false + } + Absent => false + } + let rate_skip = match claude_code_line_codex_kind(line: claude_code_stream_line(line: "{\"type\":\"rate_limit_event\",\"rate_limit_info\":{\"status\":\"allowed\",\"unifiedWindows\":{}}}")) { + Absent => true + Present { value: _ } => false + } + let filter = codex_provider_event_projection_filter as String + let parsed = parse_codex_jsonl( + text: "{\"type\":\"system\"}\n{\"type\":\"assistant\",\"message\":{\"content\":[{\"type\":\"tool_use\"}]}}\n{\"type\":\"result\",\"subtype\":\"success\",\"is_error\":false}\n", + ) + let inhabitance = match parsed { + CodexEventsRefused { events_before_refusal: _, line_number: _, raw: _, reason: _ } => false + CodexEventsParsed { events: _ } => + match provider_execution_state(parsed: parsed) { + ProviderCompleted { activity: _ } => true + _ => false + } + } + system_ok && tool_ok && result_ok && fail_ok && rate_skip && inhabitance + && string_contains(s: filter, pattern: "$type == \"result\"") + && string_contains(s: filter, pattern: "$type == \"system\"") + && string_contains(s: filter, pattern: "rate_limit_event") +} + test fn a_harness_tool_round_reaches_the_operator_as_its_own_activity() -> Bool { let read_line = concat(join(harness_tool_round_events( uses: [WireToolUseBlock { id: "toolu_1", name: harness_tool_wire_name(tool: ReadFileTool) as String, input: json_object([]) }],