diff --git a/INVARIANTS.md b/INVARIANTS.md index bc5c6fe7..cf668a0b 100644 --- a/INVARIANTS.md +++ b/INVARIANTS.md @@ -24,10 +24,10 @@ materialization, messaging, DING, or presence must preserve them. | **Agent-declared presence** | Refresh preserves non-DND declared status and advances the version 1 heartbeat. A missing status starts as `available`. Legacy DND migrates without renewing its hold. Version 1 DND is not refreshed. Stale, malformed, or implausibly future heartbeats read as `unknown`. The outer Codex, Claude, and pi session wrappers own a five-minute heartbeat while their provider remains alive. | `src/status.rs::refresh_preserves_value_and_changes_heartbeat_bytes`; `src/status.rs::refresh_upgrades_legacy_dnd_without_renewing_the_hold`; `src/status.rs::refresh_missing_writes_available_default`; `src/status.rs::version_1_staleness_and_future_skew_are_bounded`; `src/status.rs::malformed_versioned_record_is_unknown_without_mtime_fallback`; `src/claude_session.rs::idle_provider_refreshes_presence_without_mcp_input`; `src/pi_session.rs::idle_pi_provider_refreshes_presence_without_channel_input`; `src/codex_app_server.rs::inbox_fallback_does_not_write_a_fifteen_second_presence_heartbeat` | | **Scoped delivery-input wakeups** | Native delivery pumps watch only their inputs: the agent's `resources/inbox` subtree and its `status` file. Runtime records written beside them by the pump's own process group — presence temp siblings, the `harness-state` and `harness-context` records with their locks and staged siblings, stream state — never wake delivery, so a producer that writes on every turn boundary cannot pump its own delivery loop. | `src/watch.rs::delivery_watcher_ignores_runtime_records_but_wakes_on_inbox_and_status` | | **Observed harness state discipline** | The `harness-state` record is written only by the owning session's driver processes — wrapper, channel, or hooks — serialized by a cross-process lock and coalesced against the on-disk record, atomically and byte-distinct on every write that lands, with freshness from its embedded timestamp and never file mtime; restating an unchanged state touches the record only when the refresh cadence is due. `unknown` is derived, never written: staleness, future skew, malformation, an unsupported schema, and a provably dead pty session each read as `unknown` with a distinct reason; an indeterminate liveness probe downgrades nothing; a missing record is no observation rather than `unknown`; no absence derives a definite state. A writer that loses sight of its harness stops heartbeating instead of refreshing a state it cannot see, and a predecessor session's record is never re-stamped. A reaped provider yields a terminal `ended` record carrying its real exit — written before the stop path's SIGKILL escalation and rewritten from the escalation cover when a grace-window reap observes the real status, proven against the real wrapper binaries of both stop implementations — and never a live state. Faults are their own axis, in their own record version: `st2.harness-state.v3` carries a tagged condition whose fault names a closed category, an open provider-namespaced code, a recovery class, and its own SEMANTIC observation time, beside a tagged ask axis that speaks only about actual human prompts — a fault is not an ask, and where both hold remediation is primary while the ask stays visible. Versions 1 and 2 project their condition as explicitly absent: never `clear`, and no fault is inferred from their legacy words. Every contradiction is rejected with its own reason word and reported as typed indeterminacy carrying the age of the evidence, while an unrecognized category leaves a fault untyped and still routed by its recovery, and an unrecognized recovery is treated like one needing a human. The heartbeat moves transport freshness only: it never moves a fault's observation time or its deadline, attention is derived at read time, and an automatic recovery past its own deadline becomes an untyped unknown-recovery fault that pages until an explicit paired clear, a terminal record, a new claim, or a new incarnation replaces it. One pure function folds those axes and the sibling driver diagnostic into exactly three closed disposition axes that the roster, the catalog graph, and Doctor all read rather than re-derive, where ended and record-level indeterminate never page and no delivery path reads the result. A conversation reference is identity and capability only — tagged, with an explicit history-mutability claim, its evidence, and a positive finite verification bound — and carries no conversation content. Reader-first holds for the new version too: this build reads, strictly validates, and projects version 3 while one writer-selection point keeps the writer on version 2, and the version-independent ownership envelope makes a version 2 claim refuse a version 3 record rather than overwrite a meaning it cannot read, while still carrying an undecodable record's ownership sequence forward. Strictness is scoped to what a contradiction damages, because it must not destroy evidence: a recovery deadline is refused only on a recovery this version recognizes as non-automatic, an unrecognized recovery word keeps its fault paging, an inverted deadline is refused whatever the recovery says, and a badly stated conversation reference degrades only that axis to `unavailable` carrying st2's own rejection word while the activity, condition, and ask axes stand. | `src/harness_state.rs::unknown_state_is_derived_and_cannot_be_written`; `src/harness_state.rs::malformed_record_is_unknown_without_mtime_fallback`; `src/harness_state.rs::staleness_and_future_skew_derive_unknown_with_distinct_reasons`; `src/harness_state.rs::a_dead_session_reads_unknown_even_while_fresh_but_ended_survives`; `src/harness_state.rs::every_landed_write_is_byte_distinct_and_fresh_restatements_do_not_write`; `src/harness_state.rs::a_chatty_producer_restating_its_state_causes_zero_writes`; `src/harness_state.rs::concurrent_writers_defer_to_the_on_disk_record_not_their_cache`; `src/harness_state.rs::a_predecessor_sessions_record_is_never_heartbeat_eligible`; `src/harness_state.rs::missing_record_reads_as_none_not_unknown`; `src/codex_app_server.rs::pump_publishes_observations_and_stops_heartbeating_on_evidence_loss`; `src/claude_session.rs::a_provider_killed_mid_turn_reads_ended_rather_than_active`; `src/claude_session.rs::a_clean_provider_exit_writes_the_terminal_record`; `tests/harness_state_teardown.rs::stop_escalation_writes_the_terminal_record_before_sigkill`; `tests/harness_state_teardown.rs::opencode_stop_escalation_writes_the_cover_record_before_sigkill`; `tests/harness_state_teardown.rs::opencode_graceful_stop_records_the_real_reaped_exit`; `src/harness_state.rs::golden_fixtures_cover_every_shape_a_consumer_must_handle`; `src/harness_state.rs::strict_version_three_edges_are_rejected_with_distinct_reasons`; `src/harness_state.rs::the_conversation_bridge_requires_a_complete_finite_link`; `src/harness_state.rs::the_heartbeat_moves_transport_freshness_only_and_never_the_semantic_fault_clock`; `src/harness_state.rs::a_driver_diagnostic_failure_contributes_without_touching_raw_axes_or_paging_the_dead`; `src/harness_state.rs::a_version_three_record_is_read_but_never_written_over_by_this_builds_writer`; `src/harness_state.rs::an_undecodable_records_envelope_still_carries_its_ownership_forward` | -| **Harness context discipline** | The numeric `harness-context` record is a sibling of `harness-state` and shares none of its ownership machinery: `incarnation` is provenance and never a fence, and a straggler's write lands rather than being refused. A reading is written only when it enters a different bucket of `usedPercent`, when a compaction edge arrives, when proven Claude account-window exhaustion changes, or when the record is older than the heartbeat, so a chatty producer cannot inflate the write rate; Codex account-window occupancy does not classify availability without its separate credit metadata. Every landed write is byte-distinct, and freshness comes from the record's own `observedAtMs`, never file mtime. A withheld value is carried as `null` and never fabricated from zero, the previous reading, or a division st2 could have done itself, and a reading above the window is carried unclamped. There is no `unknown` on this axis: past the horizon the reading is returned marked `stale` with its age, surviving every derivation the categorical record makes, while an unparseable record, a foreign schema, an uninterpretable harness, and an untrustworthy clock each read as absent. The relaunch claim removes the record, and Doctor's high-reading and stale-record lines are advisory in both directions and never change its exit status. Where a producer ships, its numerator is the harness's own and is pinned to the build it was measured on: pi's `tokens` is the last assistant message's `totalTokens` and omp's is its prompt figure, so neither harness's arithmetic can be published under the other's tag, and a harness bump that changes a number's meaning without changing its shape fails a fixture rather than shipping. A harness that positively reports it does not know its own occupancy is forwarded as withheld in the same write as the compaction edge that emptied it, a compaction edge whose harness names no reason yields `unknown` rather than an invented word, and a durable count the producer cannot read narrows to st2 counting edges rather than losing them. | `src/harness_context.rs::a_reading_inside_the_written_bucket_does_not_write_and_a_crossing_does`; `src/harness_context.rs::claude_rate_limit_exhaustion_and_reset_crossings_land_inside_one_usage_bucket`; `src/harness_context.rs::codex_account_window_exhaustion_does_not_prove_the_runtime_is_blocked`; `src/harness_context.rs::a_withheld_percent_has_no_bucket_and_only_a_compaction_or_heartbeat_writes`; `src/harness_context.rs::a_record_older_than_the_heartbeat_is_rewritten_and_every_landed_write_is_byte_distinct`; `src/harness_context.rs::a_compaction_always_lands_with_its_trigger_and_may_carry_a_durable_count`; `src/harness_context.rs::withheld_values_are_null_and_are_never_fabricated`; `src/harness_context.rs::a_reading_above_the_window_is_carried_unclamped`; `src/harness_context.rs::a_stale_reading_is_returned_with_its_age_rather_than_derived_away`; `src/harness_context.rs::freshness_comes_from_the_record_bytes_and_never_from_file_mtime`; `src/harness_context.rs::additive_fields_decode_but_foreign_schema_and_harness_read_as_nothing`; `src/harness_context.rs::a_record_beyond_the_future_skew_bound_reads_as_nothing`; `src/harness_context.rs::a_straggler_lands_and_is_visible_as_provenance_rather_than_being_refused`; `src/harness_state.rs::the_relaunch_claim_removes_the_harness_context_record`; `src/agents.rs::context_is_a_fourth_axis_that_survives_an_indeterminate_observed_state`; `tests/doctor.rs::harness_context_doctor_lines_are_advisory_and_never_change_the_exit_status`; `src/pi_channel.rs::the_pi_0_84_2_fixture_pins_total_tokens_as_the_numerator`; `src/pi_channel.rs::the_omp_18_0_9_fixture_pins_prompt_input_as_the_numerator`; `src/pi_channel.rs::a_pi_compaction_withholds_the_reading_it_emptied_in_the_same_write`; `src/pi_channel.rs::an_omp_compaction_yields_unknown_because_the_event_names_no_reason`; `src/pi_channel.rs::an_unreadable_durable_count_degrades_to_counting_edges_not_to_losing_them`; `src/pi_channel.rs::context_frames_decode_conservatively_or_not_at_all`; `src/pi_channel.rs::the_measured_pi_release_is_the_one_the_extension_gate_pins`; `src/omp_session.rs::the_measured_context_builds_are_admitted_by_this_gate` | +| **Harness context discipline** | The numeric `harness-context` record is a sibling of `harness-state` and shares none of its ownership machinery: `incarnation` is provenance and never a fence, and a straggler's write lands rather than being refused. A reading is written only when it enters a different bucket of `usedPercent`, when a compaction edge arrives, when proven Claude account-window exhaustion changes, or when the record is older than the heartbeat, so a chatty producer cannot inflate the write rate; Codex account-window occupancy does not classify availability without its separate credit metadata. Every landed write is byte-distinct, and freshness comes from the record's own `observedAtMs`, never file mtime. A withheld value is carried as `null` and never fabricated from zero, the previous reading, or a division st2 could have done itself, and a reading above the window is carried unclamped. There is no `unknown` on this axis: past the horizon the reading is returned marked `stale` with its age, surviving every derivation the categorical record makes, while an unparseable record, a foreign schema, an uninterpretable harness, and an untrustworthy clock each read as absent. The relaunch claim removes the record, and Doctor's high-reading and stale-record lines are advisory in both directions and never change its exit status. Where a producer ships, its numerator is the harness's own and is pinned to the build it was measured on: pi's `tokens` is the last assistant message's `totalTokens` and omp's is its prompt figure, so neither harness's arithmetic can be published under the other's tag, and a harness bump that changes a number's meaning without changing its shape fails a fixture rather than shipping. A harness that positively reports it does not know its own occupancy is forwarded as withheld in the same write as the compaction edge that emptied it, a compaction edge whose harness names no reason yields `unknown` rather than an invented word, and a durable count the producer cannot read narrows to st2 counting edges rather than losing them. | `src/harness_context.rs::a_reading_inside_the_written_bucket_does_not_write_and_a_crossing_does`; `src/harness_context.rs::claude_rate_limit_exhaustion_and_reset_crossings_land_inside_one_usage_bucket`; `src/harness_context.rs::codex_account_window_exhaustion_does_not_prove_the_runtime_is_blocked`; `src/harness_context.rs::a_withheld_percent_has_no_bucket_and_only_a_compaction_or_heartbeat_writes`; `src/harness_context.rs::a_record_older_than_the_heartbeat_is_rewritten_and_every_landed_write_is_byte_distinct`; `src/harness_context.rs::a_compaction_always_lands_with_its_trigger_and_may_carry_a_durable_count`; `src/harness_context.rs::withheld_values_are_null_and_are_never_fabricated`; `src/harness_context.rs::a_reading_above_the_window_is_carried_unclamped`; `src/harness_context.rs::a_stale_reading_is_returned_with_its_age_rather_than_derived_away`; `src/harness_context.rs::freshness_comes_from_the_record_bytes_and_never_from_file_mtime`; `src/harness_context.rs::additive_fields_decode_but_foreign_schema_and_harness_read_as_nothing`; `src/harness_context.rs::a_record_beyond_the_future_skew_bound_reads_as_nothing`; `src/harness_context.rs::a_straggler_lands_and_is_visible_as_provenance_rather_than_being_refused`; `src/harness_state.rs::the_relaunch_claim_removes_the_harness_context_record`; `src/agents.rs::context_is_a_fourth_axis_that_survives_an_indeterminate_observed_state`; `tests/doctor.rs::harness_context_doctor_lines_are_advisory_and_never_change_the_exit_status`; `src/pi_channel.rs::the_pinned_pi_fixture_pins_total_tokens_as_the_numerator`; `src/pi_channel.rs::the_omp_18_0_9_fixture_pins_prompt_input_as_the_numerator`; `src/pi_channel.rs::a_pi_compaction_withholds_the_reading_it_emptied_in_the_same_write`; `src/pi_channel.rs::an_omp_compaction_yields_unknown_because_the_event_names_no_reason`; `src/pi_channel.rs::an_unreadable_durable_count_degrades_to_counting_edges_not_to_losing_them`; `src/pi_channel.rs::context_frames_decode_conservatively_or_not_at_all`; `src/pi_channel.rs::the_measured_pi_release_is_the_one_the_extension_gate_pins`; `src/omp_session.rs::the_measured_context_builds_are_admitted_by_this_gate` | | **Replicated-path discipline** | st2 pins the exact driver-record names it expects the replication transport's include list to carry — `harness-state` and `harness-context` — and both readers derive their paths from that list, so a rename cannot silently stop replication in a repository that does not own the list. A harness-context write stages in the catalog control plane, outside the complete `agents` identity namespace, after validating canonical ancestry and same-filesystem atomic rename. Current-catalog identity walkers overlook only an exact legacy `.harness-context.tmp--` regular non-symlink host child and leave it untouched; directories, symlinks, special files, generic dotfiles, near misses, and prepared topology remain strict. `harness-state` still stages beside itself, pending its own change; the shared write helper takes the staging directory as an argument precisely because the two answer this differently. | `src/harness_context.rs::the_replicated_driver_record_names_are_pinned`; `src/harness_context.rs::writes_stage_in_catalog_control_and_clean_up_after_success_or_failure`; `src/catalog_transaction.rs::current_projection_ignores_and_preserves_an_exact_legacy_staging_file`; `src/catalog_transaction.rs::current_projection_rejects_legacy_shape_type_confusion_and_near_misses`; `src/message.rs::transition_addressability_ignores_and_preserves_exact_legacy_staging_files`; `src/message.rs::transition_addressability_rejects_legacy_type_confusion_and_near_misses`; `tests/catalog_apply.rs::snapshot_ignores_and_preserves_an_exact_legacy_harness_context_staging_file` | | **Status-line slot chaining** | Claude's `statusLine` is a single slot whose winning declaration replaces the others outright, and `.claude/settings.local.json` — the file st2 materializes — wins over every other settings file. So st2's entry there is a tee: it records the reading and then invokes the operator's own renderer, resolved from `$ST_CLAUDE_STATUSLINE_RENDERER` and then `~/.claude/statusline-renderer.json`, first hit wins and never both. Where no renderer resolves, and wherever a resolved renderer fails, the tee writes NOTHING to stdout and puts its diagnostic on stderr: the payload is machine JSON, so echoing it into the slot is strictly worse for the operator than a blank row. Recording is unaffected by which arm runs. | `tests/claude_statusline.rs::the_tee_records_the_reading_and_hands_the_same_payload_to_the_env_renderer`; `tests/claude_statusline.rs::the_operator_file_supplies_the_renderer_when_no_variable_does`; `tests/claude_statusline.rs::the_variable_wins_over_the_file_and_the_file_is_never_also_run`; `tests/claude_statusline.rs::with_no_renderer_the_tee_renders_nothing_rather_than_the_raw_payload`; `tests/claude_statusline.rs::a_recording_failure_still_renders_the_status_line`; `tests/claude_statusline.rs::a_renderer_that_exits_non_zero_leaves_stdout_empty`; `tests/claude_statusline.rs::a_renderer_file_that_is_not_executable_leaves_stdout_empty`; `tests/claude_statusline.rs::without_st2_on_path_the_script_drains_stdin_and_renders_nothing`; `tests/claude_statusline.rs::the_rendered_registration_names_the_chaining_tee_and_carries_the_whole_slot` | -| **Version-pinned producer arithmetic** | Every harness-context producer publishes the number its own harness means by that number, and each one is pinned to the exact build the meaning was measured on: Claude's integer percent over `total_input_tokens` and `context_window_size` (2.1.250), Codex's window with the 12,000 baseline subtracted from both sides (codex-cli 0.151.0), pi's last assistant `totalTokens` and omp's prompt-only `input` (0.84.2 and 18.0.9), and OpenCode's last non-summary assistant total over the providers window (1.18.25). The failure this bounds is a harness bump that changes what a key means without changing its shape, which no type gate and no round-trip assertion can see, so each fixture decodes a verbatim capture, asserts the version literally, and asserts the wrong numerator the same capture would yield. | `src/claude_session.rs::a_mid_session_statusline_payload_yields_claudes_own_triple`; `src/claude_session.rs::a_pre_turn_statusline_payload_withholds_rather_than_reporting_zero`; `src/codex_app_server.rs::codex_context_recomputes_the_captured_reading_and_pins_its_verified_version`; `src/pi_channel.rs::the_pi_0_84_2_fixture_pins_total_tokens_as_the_numerator`; `src/pi_channel.rs::the_omp_18_0_9_fixture_pins_prompt_input_as_the_numerator`; `src/opencode_session.rs::captured_opencode_turns_publish_the_assistant_total_over_the_providers_window` | +| **Version-pinned producer arithmetic** | Every harness-context producer publishes the number its own harness means by that number, and each one is pinned to the exact build the meaning was measured on: Claude's integer percent over `total_input_tokens` and `context_window_size` (2.1.250), Codex's window with the 12,000 baseline subtracted from both sides (codex-cli 0.151.0), pi's last assistant `totalTokens` and omp's prompt-only `input` (0.84.4 and 18.0.9), and OpenCode's last non-summary assistant total over the providers window (1.18.25). The failure this bounds is a harness bump that changes what a key means without changing its shape, which no type gate and no round-trip assertion can see, so each fixture decodes a verbatim capture, asserts the version literally, and asserts the wrong numerator the same capture would yield. | `src/claude_session.rs::a_mid_session_statusline_payload_yields_claudes_own_triple`; `src/claude_session.rs::a_pre_turn_statusline_payload_withholds_rather_than_reporting_zero`; `src/codex_app_server.rs::codex_context_recomputes_the_captured_reading_and_pins_its_verified_version`; `src/pi_channel.rs::the_pinned_pi_fixture_pins_total_tokens_as_the_numerator`; `src/pi_channel.rs::the_omp_18_0_9_fixture_pins_prompt_input_as_the_numerator`; `src/opencode_session.rs::captured_opencode_turns_publish_the_assistant_total_over_the_providers_window` | | **Native-driver diagnostics are bounded and recoverable** | One additive-tolerant `st2.driver-diagnostic.v1` snapshot exposes the earliest failing native-driver boundary through closed driver/stage/reason/source/support vocabulary and origin-timed evidence age. Absent, malformed, foreign, unknown, and wrongly paired records never read healthy. Stage recovery clears only its own failure and removes the record after final recovery without changing prompt submission, retry, read-back, or archive semantics. A rejected provider credential is published on its own `providerAuth` stage from the harness's own typed turn result — never from provider prose, and never from an auth-class flag a capacity, policy, or retry class also sets — outranking the delivery and read-back symptoms it causes while the four producer gates outrank it; only a turn that reached its ordinary end clears it, and no credential knowledge or remedy text enters st2. Roster and Doctor share the typed reader/repair policy — Doctor advising on absence only for a driver that publishes a boundary result on every launch; telemetry labels contain only bounded driver/stage/reason/source/support/outcome values, never versions, identities, prompts, messages, or paths. | `src/driver_diagnostic.rs::every_stage_reason_and_source_has_bounded_wire_vocabulary`; `src/driver_diagnostic.rs::additive_fields_decode_but_malformed_foreign_and_unknown_records_are_indeterminate`; `src/driver_diagnostic.rs::recovery_clears_only_its_stage_and_reveals_the_next_failure`; `src/driver_diagnostic.rs::a_credential_rejection_is_evidence_only_from_a_typed_turn_result`; `src/driver_diagnostic.rs::a_rejected_credential_outranks_its_symptoms_but_not_the_producer_gates`; `src/claude_session.rs::stop_failure_classifies_only_the_credential_class_as_provider_auth`; `src/claude_session.rs::a_rejected_claude_credential_stands_until_a_turn_reaches_its_ordinary_end`; `src/codex_app_server.rs::a_rejected_codex_credential_reads_provider_auth_while_a_quota_failure_does_not`; `src/codex_app_server.rs::protocol_schema_gate_requires_the_distinct_credential_and_quota_error_words`; `src/opencode_session.rs::delivery_and_read_back_boundaries_publish_and_clear_diagnostics_without_changing_retry`; `src/metrics.rs::driver_diagnostic_metric_attributes_are_exactly_the_bounded_axes`; `tests/doctor.rs::native_driver_diagnostic_roster_and_doctor_agree_and_recovery_clears`; `src/pi_channel.rs::only_omps_own_credential_class_becomes_provider_auth`; `src/pi_channel.rs::a_rejected_omp_credential_stands_until_a_turn_reaches_its_ordinary_end`; `src/pi_channel.rs::the_pi_channel_publishes_no_credential_verdict`; `tests/doctor.rs::a_rejected_provider_credential_advises_while_absence_stays_silent` | | **Retirement health** | A retired declaration is healthy only after every declared task ID is absent. Any live or dead declared task record reports incomplete retirement; retired declarations do not require presence. Live declarations retain their existing task and presence checks. | `tests/doctor.rs::retired_declaration_is_healthy_when_tasks_and_presence_are_absent`; `tests/doctor.rs::retired_declaration_is_unhealthy_while_a_declared_task_is_alive`; `tests/doctor.rs::retired_declaration_is_unhealthy_while_a_dead_task_record_remains` | | **Suspension health** | A suspended declaration is healthy when no declared task is live and every retained dead record is explicitly keep-pinned. It requires no presence, but this weaker result never proves retirement. Resume preserves ordinary keep and adopt-only policy. | `tests/doctor.rs::suspended_declaration_is_healthy_when_tasks_are_absent_without_presence`; `tests/doctor.rs::suspended_declaration_distinguishes_live_dead_keep_and_dead_nonkeep`; `tests/reconcile.rs::resuming_uses_ordinary_reconcile_and_does_not_override_keep` | diff --git a/docs/vrs/08-harness-context/spec.md b/docs/vrs/08-harness-context/spec.md index 6b9d4d40..764f687e 100644 --- a/docs/vrs/08-harness-context/spec.md +++ b/docs/vrs/08-harness-context/spec.md @@ -415,7 +415,7 @@ number (HC-R02); where st2 computes it, the row says so. | --- | --- | --- | --- | --- | --- | | claude | 2.1.250 | `statusLine` command stdin JSON | `context_window.total_input_tokens` = `input + cache_creation + cache_read` of the last response | `context_window.context_window_size` | `context_window.used_percentage` — Claude's own integer, clamped 0..100 | | codex | codex-cli 0.151.0 | app-server `thread/tokenUsage/updated` | `tokenUsage.last.totalTokens` | `tokenUsage.modelContextWindow` | st2 computes with the baseline rule below; equals `100 −` Codex's displayed "% context left" | -| pi | 0.84.2 | injected extension `ctx.getContextUsage()` | `.tokens` = last assistant `totalTokens` (input + output + cacheRead + cacheWrite) | `.contextWindow` | `.percent` (float) | +| pi | 0.84.4 | injected extension `ctx.getContextUsage()` | `.tokens` = last assistant `totalTokens` (input + output + cacheRead + cacheWrite) | `.contextWindow` | `.percent` (float) | | omp | 18.0.9 (and 18.0.3) | injected extension `ctx.getContextUsage()` | `.tokens` = last assistant **`input`** only | `.contextWindow` | `.percent` (float) | | opencode | 1.18.25 | SSE `message.updated` joined with `GET /config/providers` | last **non-summary** assistant `tokens.total` | `providers[].models[].limit.context` | st2 computes `usedTokens / windowTokens`; the server displays none | @@ -727,7 +727,7 @@ turn the measurement into a semantic-version promise. Every later Codex release still requires its own source comparison and live delivery proof before the exact launch gate moves. -### pi (0.84.2) +### pi (0.84.4) One call answers everything, present on the `ctx` of every lifecycle event: `getContextUsage()` returns `{ tokens, contextWindow, percent }`. @@ -1126,7 +1126,7 @@ each only once a real test proves it (per `CLAUDE.md`): baseline-free `last.totalTokens / window` reads 36, the truth is 33). **Shipped for pi and omp**: - `src/pi_channel.rs::the_pi_0_84_2_fixture_pins_total_tokens_as_the_numerator`, + `src/pi_channel.rs::the_pinned_pi_fixture_pins_total_tokens_as_the_numerator`, `src/pi_channel.rs::the_omp_18_0_9_fixture_pins_prompt_input_as_the_numerator`, `src/pi_channel.rs::a_pi_compaction_withholds_the_reading_it_emptied_in_the_same_write`, `src/pi_channel.rs::an_omp_compaction_yields_unknown_because_the_event_names_no_reason`, diff --git a/flake.nix b/flake.nix index 7a30cdaf..5801a1f2 100644 --- a/flake.nix +++ b/flake.nix @@ -492,10 +492,10 @@ # without integrity hashes, which `fetchNpmDeps` cannot express. checks.pi-extension-types = let - piVersion = "0.84.2"; + piVersion = "0.84.4"; piTarball = pkgs.fetchurl { url = "https://registry.npmjs.org/@earendil-works/pi-coding-agent/-/pi-coding-agent-${piVersion}.tgz"; - hash = "sha256-lbiZzXsaDB8BdMe/M6tCdDXjVTp9H0dWZhqpx/Gmj/o="; + hash = "sha256-W852bRnDzroY8/uq2RxEnJ+dc5gfnjQA7O+TIAbwaWg="; }; nodeTypesTarball = pkgs.fetchurl { url = "https://registry.npmjs.org/@types/node/-/node-26.2.0.tgz"; diff --git a/hooks/omp-channel.ts b/hooks/omp-channel.ts index 601d65ff..b51727d2 100644 --- a/hooks/omp-channel.ts +++ b/hooks/omp-channel.ts @@ -137,7 +137,7 @@ const terminalProviderError = (event: AgentEndFrame): ProviderError | undefined * build whose telemetry surface moved still loads and still delivers mail, and a widened cast alone * would make that tolerance absolute and silent. Erased at runtime. * - * Note what this can and cannot prove for omp. It pins the SHAPE against pi 0.84.2's typings, which + * Note what this can and cannot prove for omp. It pins the SHAPE against pi 0.84.4's typings, which * is all this asset compiles against — omp ships no typings of its own. It cannot prove omp's * `tokens` still means prompt-only input, because that is a meaning and not a shape; the * version-pinned fixture in `src/pi_channel.rs` is what bounds that (HC-R13, HC-T03). diff --git a/hooks/pi-channel.ts b/hooks/pi-channel.ts index f95b7696..385f199f 100644 --- a/hooks/pi-channel.ts +++ b/hooks/pi-channel.ts @@ -20,7 +20,19 @@ import type { SessionShutdownEvent, } from "@earendil-works/pi-coding-agent"; -const PROTOCOL = 1; +// The wire versions this asset speaks, highest first in preference. Version 1 is the frame set +// this file shipped with; version 2 adds the condition frame and nothing else. Negotiation is +// `max(SUPPORTED ∩ hello.protocols)`, with the hello's scalar `protocol` as the only offer when an +// older control plane sends no list — so this asset keeps working against both, and a control +// plane that offers nothing this asset speaks is refused rather than guessed at. +const SUPPORTED = [1, 2] as const; +const CONDITION_PROTOCOL = 2; + +// The only two faults pi can prove, provider-namespaced so a code cannot collide with another +// harness's. Named here because the same strings appear on a raise and on its paired clear, and a +// clear whose code drifted from its raise silently stops clearing anything. +const ASSISTANT_ERROR = "pi/assistantError"; +const COMPACT_FAILED = "pi/session_compact_failed"; const BIN = "ST2_PI_CHANNEL_BIN"; const CATALOG = "ST2_PI_CHANNEL_CATALOG"; @@ -36,6 +48,7 @@ const HELLO_TIMEOUT_MS = 5000; type Frame = { type?: string; protocol?: number; + protocols?: unknown; sessionContext?: string; content?: string; deliverAs?: "steer" | "followUp"; @@ -58,6 +71,13 @@ type Stash = { session?: string; seq?: string; child?: childProcess.ChildProcess; + /** + * The wire version agreed with the channel this stash currently holds, unset until its hello + * arrives. It lives beside `child` because it describes that channel and nothing else: a + * session replacement spawns a new channel and must re-negotiate rather than inherit, or a + * downgraded control plane would keep receiving frames its wire cannot carry. + */ + protocol?: number; /** * The last assistant message's `usage.cost.total`. * @@ -169,7 +189,11 @@ export default function (pi: ExtensionAPI) { const closeChild = (child: childProcess.ChildProcess | undefined) => { if (!child) return; - if (state.child === child) state.child = undefined; + if (state.child === child) { + state.child = undefined; + // The agreement belonged to that channel. A successor re-negotiates from its own hello. + state.protocol = undefined; + } // The channel treats EOF on its stdin as the session boundary, so ending this pipe is what // reaps it. That happens on its own whenever pi exits, however it exits. if (!child.stdin?.destroyed) child.stdin?.end(); @@ -211,6 +235,9 @@ export default function (pi: ExtensionAPI) { { stdio: ["pipe", "pipe", "inherit"], env: channelEnv }, ); state.child = child; + // A fresh channel has agreed nothing yet, and until its hello lands this asset sends only + // the frames every version carries. + state.protocol = undefined; return new Promise((resolve) => { let settled = false; @@ -221,16 +248,20 @@ export default function (pi: ExtensionAPI) { }; const timer = setTimeout(() => settle(""), HELLO_TIMEOUT_MS); timer.unref?.(); + // Retiring the channel retires its agreement with it. + const retire = () => { + if (state.child !== child) return; + state.child = undefined; + state.protocol = undefined; + }; child.on("error", () => { - if (state.child === child) state.child = undefined; + retire(); settle(""); }); // An observability pipe must never take pi down: a channel that closed its stdin mid-write // surfaces EPIPE on the stream, which without a listener is an uncaught exception in the // host process. Retire the channel instead — frames simply stop, fail-open. - child.stdin.on("error", () => { - if (state.child === child) state.child = undefined; - }); + child.stdin.on("error", retire); child.on("exit", () => settle("")); const send = (frame: Record) => { @@ -246,17 +277,34 @@ export default function (pi: ExtensionAPI) { return; } if (frame.type === "hello") { - // A newer control plane may speak a wire this asset was not written against. Refusing is - // the honest outcome: presence still decays, so the agent reads as unreachable rather - // than silently never receiving mail. - if (frame.protocol !== PROTOCOL) { + // Negotiate rather than compare. The hello's scalar `protocol` is a FLOOR that never + // moves — an already-published asset compares it for equality — so a control plane that + // speaks a newer wire advertises the set in `protocols` and this asset picks the highest + // it also speaks. An empty intersection is refused, which is the honest outcome: + // presence still decays, so the agent reads as unreachable rather than silently never + // receiving mail. + const offered = Array.isArray(frame.protocols) + ? frame.protocols.filter((value): value is number => typeof value === "number") + : typeof frame.protocol === "number" + ? [frame.protocol] + : []; + const agreed = offered + .filter((value) => (SUPPORTED as readonly number[]).includes(value)) + .reduce( + (best, value) => (best === undefined || value > best ? value : best), + undefined, + ); + if (agreed === undefined) { closeChild(child); ctx.ui?.notify?.( - `st2: pi channel protocol ${frame.protocol} is not understood by this extension (expected ${PROTOCOL}); reinstall st2's hook set`, + `st2: pi channel offers protocol ${offered.join(", ") || "none"}, which this extension does not speak (it speaks ${SUPPORTED.join(", ")}); reinstall st2's hook set`, ); settle(""); return; } + // Recorded against the channel this hello came from, never "whatever is current": a + // predecessor draining its queued hello must not re-version the successor's channel. + if (state.child === child) state.protocol = agreed; clearTimeout(timer); settle(typeof frame.sessionContext === "string" ? frame.sessionContext : ""); return; @@ -302,7 +350,7 @@ export default function (pi: ExtensionAPI) { // starts exactly at that boundary — an `agent_end` emit would blip a spurious idle before it. // `agent_settled` is the first point pi is provably idle. The frame is observational — st2 // decides what becomes of it — and a closed channel drops it silently, matching the fail-open - // rule this file already follows. pi 0.84.2 exposes no typed waiting-on-a-human event, so no + // rule this file already follows. pi 0.84.4 exposes no typed waiting-on-a-human event, so no // frame here ever claims one. const sendFrame = (frame: Record) => { const child = state.child; @@ -311,6 +359,71 @@ export default function (pi: ExtensionAPI) { }; const sendState = (word: "active" | "idle") => sendFrame({ type: "state", state: word }); + /** + * The condition axis, extension side (protocol 2 only). + * + * A condition frame states the fault axis and NOTHING ELSE: pi's fault evidence (`agent_end`) + * carries no activity claim, and folding it onto a state frame would fabricate one and refresh + * a stale activity from an event that observed none. The converse holds too — a state frame + * never touches the condition axis — so a standing fault survives every activity edge and the + * record settles as `idle` beside it: activity honest, wedged seat visible. That pairing is the + * whole fix for the measured false idle, where `agent_settled` fires from a `finally` after a + * failed turn and published a clean idle for a wedged seat. + * + * Gated on the negotiated version, not on the code being installed: an st2 that speaks only + * protocol 1 has nowhere to put this frame, and sending it anyway would put an unreadable line + * on a wire that is otherwise exactly the one it shipped with. + */ + const sendCondition = (op: Record) => { + if ((state.protocol ?? 0) < CONDITION_PROTOCOL) return; + sendFrame({ type: "condition", ...op }); + }; + + /** + * pi's own verdict on the run that just ended, read off the LAST assistant message in + * `agent_end`'s `messages` array — measured against the published tarball's own declarations + * (`AgentEndEvent { type: "agent_end"; messages: AgentMessage[] }`, 0.84.2 + * dist/core/extensions/types.d.ts:542-544 and 0.84.4 :555-558). There is no `message` singular + * on this event; reading one returned `undefined` for every real end, which is silently the + * worst possible outcome — no raise, no clear, and the false idle unfixed. + * + * The classification compares pi's typed `stopReason` against its own closed vocabulary + * (`pending | stop | length | toolUse | error | aborted | deferred`, pi-ai + * 0.84.4 dist/types.d.ts:287, 0.84.2 :277) and never reads prose. `errorMessage` rides along + * as DIAGNOSTIC detail only: nothing branches on it and no category is inferred from it. + * + * Three answers, because two would have to lie. `failed` is an error-ended run. `completed` is + * an ordinary end — `stop` or `length`, the two words that prove the provider answered and the + * run finished — and it is the only positive success edge. `undefined` covers everything else: + * no assistant message at all, an `aborted` run (a person interrupted it, which proves neither + * health nor fault), and `pending`/`deferred`/`toolUse`/any future word. Those emit NO frame: + * a raise would invent a failure and a clear would silence a fault nobody saw resolve. + */ + const runOutcome = ( + messages: readonly unknown[], + ): { state: "failed" | "completed"; detail?: string } | undefined => { + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index]; + if (!message || typeof message !== "object") continue; + if (!("role" in message) || message.role !== "assistant") continue; + if (!("stopReason" in message) || typeof message.stopReason !== "string") return undefined; + if (message.stopReason === "error") { + const detail = + "errorMessage" in message && + typeof message.errorMessage === "string" && + message.errorMessage.trim() !== "" + ? message.errorMessage + : undefined; + return { state: "failed", detail }; + } + if (message.stopReason === "stop" || message.stopReason === "length") { + return { state: "completed" }; + } + return undefined; + } + return undefined; + }; + // Harness context, extension side (HC-R02, HC-R03, HC-R11, HC-R12). // // `ctx.getContextUsage()` answers the whole fill triple in one call and rides the ctx of every @@ -417,31 +530,107 @@ export default function (pi: ExtensionAPI) { sendContext(ctx); }); - const onContextEvent = (name: "message_end" | "turn_end" | "agent_end") => + // A helper for the events pi types loosely; the cast is on the REGISTRATION function, not on an + // event payload, and every payload read below narrows. + const onEvent = ( + name: string, + handler: (event: unknown, ctx: ExtensionContext) => void | Promise, + ) => (pi.on as unknown as ( event: string, handler: (event: unknown, ctx: ExtensionContext) => void | Promise, - ) => void)(name, async (event, ctx) => { + ) => void)(name, handler); + + for (const name of ["message_end", "turn_end"]) { + onEvent(name, async (event, ctx) => { captureCost(event); sendContext(ctx); }); - onContextEvent("message_end"); - onContextEvent("turn_end"); - onContextEvent("agent_end"); + } + + // `agent_end` is where a pi run's OUTCOME becomes visible, and until now this handler read only + // the numbers off it. That outcome is the entire cause of the measured false idle: + // `_emitAgentSettled` runs from the `finally` of `_runAgentPrompt` and fires unconditionally + // after a failed run, so a wedged seat published a clean `idle` with no reason at all. + // Classifying here — before `agent_settled`, which is the measured order — is what makes the + // fault land first and the idle land honestly beside it. + // + // Registered through pi's TYPED overload, deliberately: `event.messages` is then checked + // against the pinned tarball's own `AgentEndEvent`, which is exactly the check the untyped + // registration cast defeated while this handler read a `message` field that does not exist. + pi.on("agent_end", async (event, ctx) => { + captureCost(event); + sendContext(ctx); + const outcome = runOutcome(event.messages ?? []); + if (!outcome) return; + if (outcome.state === "failed") { + // `harness`, not `authentication`/`quota`/`provider`: the typed evidence is "a pi run + // failed", nothing more. pi ships no error-classification field (omp's `errorId` bitfield + // is exactly what pi lacks), so any narrower category would be inferred from prose. The + // prose rides `detail`, diagnostic-only. `unknown` recovery because pi says nothing about + // who clears this — never optimistic, so it pages. + sendCondition({ + op: "raise", + category: "harness", + code: ASSISTANT_ERROR, + recovery: "unknown", + ...(outcome.detail === undefined ? {} : { detail: outcome.detail }), + }); + return; + } + // pi's ONLY positive success edge: a run that reached its ordinary end proves the provider + // accepted the credential and the work ran. `agent_settled` is not this edge — it fires from + // a `finally` after a failure too — which is why the unkeyed clear hangs here alone. + sendCondition({ op: "clearAll", proof: "turnCompleted" }); + }); // pi's `session_compact` carries `reason ∈ manual | threshold | overflow` — the only v1 producer // that names its trigger at all. Measured in the handler itself: `getContextUsage()` already // reports `{tokens: null, percent: null}` there, and `getEntries()` already counts the new // entry, so this one frame carries both the honest withheld reading and the durable count. - (pi.on as unknown as ( - event: string, - handler: (event: unknown, ctx: ExtensionContext) => void | Promise, - ) => void)("session_compact", async (event, ctx) => { - const reason = (event as { reason?: unknown })?.reason; + onEvent("session_compact", async (event, ctx) => { + const reason = event && typeof event === "object" && "reason" in event ? event.reason : null; sendContext(ctx, { trigger: typeof reason === "string" ? reason : null, count: durableCompactions(ctx), }); + // A compaction that succeeded is the paired clear for its own failure, keyed on the EXACT + // category and code. A category-only clear would also wipe any other `context` fault, and + // there is no fallback to an unkeyed clear: on a healthy seat this ordinarily matches nothing + // and st2 logs that at debug. + sendCondition({ op: "clear", category: "context", code: COMPACT_FAILED }); + }); + + // pi-only, and typed: `session_compact_failed` is the failure sibling of the event above, so + // this is a real signal rather than an inference from prose. `context`, because the harness has + // no usable window left and could not reclaim any; `human`, because nothing in pi retries it. + // No paired clear is derived from anything else — only a later successful compaction retires + // it. + // + // Two version facts ride on this handler. The event exists from pi 0.84.3 (zero occurrences in + // the 0.84.2 tarball, declared in 0.84.4 dist/core/extensions/types.d.ts:464-476), and + // registering a name a running build never emits is harmless — pi's `on` is a plain map insert + // (dist/core/extensions/loader.js:209-213) — so on an older build this is simply inert rather + // than a fault this seat cannot report. And the event's own `aborted` flag is load-bearing: a + // cancelled `/compact` is a person changing their mind, not a harness that ran out of context, + // so only `aborted === false` is a fault. The flag is read POSITIVELY — a build that does not + // carry it states nothing, and an unreadable flag must not become a raise. + onEvent("session_compact_failed", async (event) => { + if (!event || typeof event !== "object") return; + if (!("aborted" in event) || event.aborted !== false) return; + const detail = + "errorMessage" in event && + typeof event.errorMessage === "string" && + event.errorMessage.trim() !== "" + ? event.errorMessage + : undefined; + sendCondition({ + op: "raise", + category: "context", + code: COMPACT_FAILED, + recovery: "human", + ...(detail === undefined ? {} : { detail }), + }); }); pi.on("session_start", async (_event, ctx) => { diff --git a/hooks/typecheck/smoke.mjs b/hooks/typecheck/smoke.mjs index 80533534..8d429e83 100644 --- a/hooks/typecheck/smoke.mjs +++ b/hooks/typecheck/smoke.mjs @@ -13,8 +13,13 @@ import path from "node:path"; // producer that silently writes nothing is indistinguishable from the pre-producer state, where // every declaration's context reads null. That is the failure this file has to be able to see. // -// It speaks the two things the extension needs: a protocol-1 hello so `open()` settles without -// waiting out its timeout, and an append of every frame line to a file this smoke reads back. +// It speaks the two things the extension needs: a hello it can negotiate against — the offer is +// selected by `ST2_SMOKE_HELLO`, so one recorder can stand in for this build's control plane, for +// one that never learned protocol 2, and for one that speaks nothing this asset does — and an +// append of every frame line to the file named by `ST2_SMOKE_FRAMES`. Both ride the environment +// rather than being baked in, because the channel is spawned fresh on every `session_start` and +// inherits this process's environment at that moment, which is what lets one loaded extension be +// driven against several control planes. const dir = fs.mkdtempSync(path.join(os.tmpdir(), "st2-pi-smoke-")); const framesPath = path.join(dir, "frames.jsonl"); const recorder = path.join(dir, "recorder"); @@ -22,14 +27,24 @@ fs.writeFileSync( recorder, `#!${process.execPath} import fs from "node:fs"; -process.stdout.write(JSON.stringify({ type: "hello", protocol: 1, sessionContext: "" }) + "\\n"); +const offers = { + // What this build's hello carries: the floor a published asset compares for strict equality, + // beside the set a newer asset negotiates over. + negotiated: { protocol: 1, protocols: [1, 2] }, + // A control plane that never learned protocol 2 — the bytes this channel shipped with. + legacy: { protocol: 1 }, + // A wire this asset does not speak at all. + foreign: { protocol: 9, protocols: [9] }, +}; +const offer = offers[process.env.ST2_SMOKE_HELLO ?? "negotiated"]; +process.stdout.write(JSON.stringify({ type: "hello", ...offer, sessionContext: "" }) + "\\n"); process.stdin.setEncoding("utf8"); -process.stdin.on("data", (chunk) => fs.appendFileSync(${JSON.stringify(framesPath)}, chunk)); +process.stdin.on("data", (chunk) => fs.appendFileSync(process.env.ST2_SMOKE_FRAMES, chunk)); `, { mode: 0o755 }, ); -const readFrames = () => - (fs.existsSync(framesPath) ? fs.readFileSync(framesPath, "utf8") : "") +const readFrames = (file = framesPath) => + (fs.existsSync(file) ? fs.readFileSync(file, "utf8") : "") .split("\n") .filter((line) => line.trim()) .map((line) => JSON.parse(line)); @@ -40,6 +55,9 @@ process.env.ST2_PI_CHANNEL_IDENTITY = "smoke.worker"; process.env.ST2_PI_CHANNEL_RUNTIME_ID = "smoke.worker"; process.env.ST2_PI_CHANNEL_SESSION = "smoke-session"; process.env.ST2_PI_CHANNEL_SEQ = "1"; +// Read by the recorder, not by the extension: the channel inherits this environment when the +// extension spawns it, so a phase below can redirect the sink and the hello together. +process.env.ST2_SMOKE_FRAMES = framesPath; const mod = await import("./smoke-out/pi-channel.mjs"); assert.strictEqual(typeof mod.default, "function", "extension exports its entry point"); @@ -53,8 +71,15 @@ for (const name of ["session_start", "session_shutdown", "agent_start", "agent_s assert.ok(handlers.has(name), `extension registers ${name}`); } -// The harness-context producer must be registered on the events it reads, or it observes nothing. -for (const name of ["message_end", "turn_end", "agent_end", "session_compact"]) { +// The harness-context producer must be registered on the events it reads, or it observes nothing; +// `session_compact_failed` is the fault axis's only pi-specific source. +for (const name of [ + "message_end", + "turn_end", + "agent_end", + "session_compact", + "session_compact_failed", +]) { assert.ok(handlers.has(name), `extension registers ${name}`); } @@ -109,6 +134,44 @@ const messageEvent = { usage: { input: 23300, output: 25, totalTokens: 23425, cost: { total: 0.070305 } }, }, }; +// The `agent_end` payloads pi ACTUALLY emits: `{type:"agent_end", messages: AgentMessage[]}` +// (0.84.2 dist/core/extensions/types.d.ts:542-544, 0.84.4 :555-558), with `stopReason` drawn from +// pi's own closed vocabulary `pending | stop | length | toolUse | error | aborted | deferred` +// (pi-ai 0.84.4 dist/types.d.ts:287, 0.84.2 :277). These fixtures are the reason this smoke can +// see the defect class it exists for: a producer reading a `message` singular, or comparing +// against a stop word pi never emits, classifies nothing at all — no raise, no clear, false idle. +// +// Each run's tail is the LAST assistant message, so every fixture carries a tool result after a +// tool-use assistant message: a producer scanning forwards, or reading `messages[0]`, gets the +// wrong verdict here rather than passing by luck. +const toolLeg = [ + { role: "assistant", stopReason: "toolUse", usage: { totalTokens: 100 } }, + { role: "toolResult", toolName: "bash", content: [] }, +]; +const failedRun = { + type: "agent_end", + messages: [ + ...toolLeg, + { role: "assistant", stopReason: "error", errorMessage: "401 Unauthorized from fakelab" }, + ], +}; +const completedRun = { + type: "agent_end", + messages: [...toolLeg, { role: "assistant", stopReason: "stop" }], +}; +// A person pressed escape. Neither health nor fault, so neither a raise nor a clear. +const abortedRun = { + type: "agent_end", + messages: [...toolLeg, { role: "assistant", stopReason: "aborted" }], +}; +// A run whose last assistant message never resolved: also no verdict. +const pendingRun = { + type: "agent_end", + messages: [{ role: "assistant", stopReason: "pending" }], +}; +// How many times the loop below drives one full turn sequence — the exact count of raises, clears, +// and clearAlls the negotiated phase must produce. +const drives = 3; for (const ctx of [bareCtx, fullCtx, throwingCtx]) { // Two session starts in a row: the second exercises the predecessor close-and-await path — the @@ -118,7 +181,24 @@ for (const ctx of [bareCtx, fullCtx, throwingCtx]) { await handlers.get("agent_start")({}, ctx); await handlers.get("message_end")(messageEvent, ctx); await handlers.get("turn_end")(messageEvent, ctx); - await handlers.get("agent_end")({}, ctx); + // Every `agent_end` shape pi produces. Only the failed and the completed run say anything + // about the condition axis; a run with no messages at all, an aborted one, and one whose tail + // never resolved each emit nothing. + await handlers.get("agent_end")({ type: "agent_end", messages: [] }, ctx); + await handlers.get("agent_end")(failedRun, ctx); + await handlers.get("agent_end")(completedRun, ctx); + await handlers.get("agent_end")(abortedRun, ctx); + await handlers.get("agent_end")(pendingRun, ctx); + // A real compaction failure, and a cancelled one: `aborted` is what separates a harness out of + // context from a person changing their mind, and only the former is a fault. + await handlers.get("session_compact_failed")( + { reason: "overflow", aborted: false, errorMessage: "summarizer returned no content", willRetry: false, fromExtension: false }, + ctx, + ); + await handlers.get("session_compact_failed")( + { reason: "manual", aborted: true, willRetry: false, fromExtension: false }, + ctx, + ); // pi withholds tokens and percent here for real; the producer must forward that, not fill it in. await handlers.get("session_compact")( { reason: "overflow", willRetry: false }, @@ -183,6 +263,182 @@ assert.strictEqual(durable.reading.usedTokens, null, "the withheld reading rides assert.strictEqual(durable.reading.usedPercent, null); assert.strictEqual(durable.reading.windowTokens, 4000, "pi still knows its denominator"); +// The condition axis, on the wire. Everything below is the only place these emitters actually +// execute, so nothing else couples the words this asset writes to the words +// `src/pi_channel.rs::condition_frame` decodes. +const conditions = frames.filter((frame) => frame.type === "condition"); +// Told apart from a mapping defect deliberately. Every condition assertion below is an exact +// count, and there are two ways to reach zero: the classification is broken, or the recorder's +// hello never arrived inside the asset's 5 s window on a loaded machine and no channel ever +// negotiated protocol 2. The second is this harness's problem, not the asset's, and it must not +// masquerade as the first. +assert.ok( + conditions.length > 0 || frames.length === 0, + "frames arrived but no condition frame did: either the classification is broken or no channel " + + "negotiated protocol 2 (a hello that timed out leaves every condition frame gated off)", +); +// A condition frame states the fault axis and NOTHING else. An activity claim here would +// fabricate one from an event that observed none, and it is what made a wedged seat publish a +// clean idle. +for (const condition of conditions) { + assert.ok(!("state" in condition), "a condition frame never restates activity"); + assert.ok(!("ask" in condition), "a condition frame never claims an ask"); + assert.notStrictEqual(condition.op, "ended", "no frame from this asset can end a session"); +} + +const raises = conditions.filter((frame) => frame.op === "raise"); +const assistantErrors = raises.filter((frame) => frame.code === "pi/assistantError"); +assert.strictEqual( + assistantErrors.length, + drives, + "an error-ended assistant tail raises exactly one fault", +); +for (const raise of assistantErrors) { + // The teeth: prose that says `401` still yields `harness`. pi ships no error-classification + // field, so `authentication` here would be inferred from the very string carried as detail. + assert.strictEqual(raise.category, "harness", "pi's untyped turn failure is a harness fault"); + assert.strictEqual(raise.recovery, "unknown", "pi says nothing about who clears it"); + assert.strictEqual( + raise.detail, + "401 Unauthorized from fakelab", + "the prose is diagnostic only", + ); +} + +const clearAlls = conditions.filter((frame) => frame.op === "clearAll"); +assert.strictEqual(clearAlls.length, drives, "a clean agent_end is pi's only unkeyed clear"); +for (const clear of clearAlls) { + assert.strictEqual(clear.proof, "turnCompleted", "an unkeyed clear names the progress it saw"); +} + +const compactFailures = raises.filter((frame) => frame.code === "pi/session_compact_failed"); +assert.strictEqual( + compactFailures.length, + drives, + "only the non-aborted compaction failure raises; a cancelled /compact is not a fault", +); +for (const raise of compactFailures) { + assert.strictEqual(raise.category, "context"); + assert.strictEqual(raise.recovery, "human", "nothing in pi retries a failed compaction"); + assert.strictEqual( + raise.detail, + "summarizer returned no content", + "the typed event's own errorMessage rides as detail", + ); +} + +const paired = conditions.filter((frame) => frame.op === "clear"); +assert.strictEqual(paired.length, drives, "a successful compaction clears its own failure once"); +for (const clear of paired) { + assert.strictEqual(clear.category, "context"); + // Keyed on the FULL code, never the category alone: a category-only clear would also wipe any + // other context fault standing on the seat. + assert.strictEqual(clear.code, "pi/session_compact_failed"); +} + +// Four condition frames per drive and not a fifth. Everything else driven above is deliberately +// silent: an `agent_end` with no messages, an aborted run, a run whose tail is still pending, and +// a cancelled compaction each emit nothing in either direction. +assert.strictEqual( + conditions.length, + 4 * drives, + "only a failed run, a completed run, a real compaction failure, and a compaction success speak", +); + +// And the idle edge is untouched: `agent_settled` still sends its plain state frame, which is the +// honest activity beside a standing fault rather than a suppressed one. +const idles = frames.filter((frame) => frame.type === "state" && frame.state === "idle"); +assert.ok(idles.length >= drives, "agent_settled still publishes a plain idle"); +for (const idle of idles) { + assert.deepStrictEqual( + Object.keys(idle).sort(), + ["state", "type"], + "pi's state frame carries no ask and no condition", + ); +} + +// The whole point of raising from `agent_end`: it is measured to fire BEFORE `agent_settled`, so +// the fault lands first and the idle that follows carries it forward instead of publishing a +// clean yield for a wedged seat. +const firstRaise = frames.findIndex( + (frame) => frame.type === "condition" && frame.code === "pi/assistantError", +); +const idleAfterRaise = frames.findIndex( + (frame, index) => index > firstRaise && frame.type === "state" && frame.state === "idle", +); +assert.ok(firstRaise >= 0, "the error-tailed turn must raise"); +assert.ok( + idleAfterRaise > firstRaise, + "the standing fault must precede the idle it is published beside", +); + +// The state vocabulary stays closed to what pi's own turn boundaries prove. `ended` is the outer +// session wrapper's word — it alone sees the provider die — and no frame from this asset may +// claim it on any axis. +for (const frame of frames.filter((frame) => frame.type === "state")) { + assert.ok( + frame.state === "active" || frame.state === "idle", + `a state frame states only active or idle: ${JSON.stringify(frame)}`, + ); +} +assert.strictEqual( + frames.filter((frame) => frame.state === "ended" || "exit" in frame).length, + 0, + "the channel's asset never writes a terminal record", +); + +// Phase 2: a control plane that never learned protocol 2. Every frame this asset sends must be +// one that wire can carry — the condition frames are gated on the negotiated version, not on the +// code being installed. +const legacyFrames = path.join(dir, "frames-legacy.jsonl"); +process.env.ST2_SMOKE_HELLO = "legacy"; +process.env.ST2_SMOKE_FRAMES = legacyFrames; +await handlers.get("session_start")({}, fullCtx); +await handlers.get("agent_start")({}, fullCtx); +await handlers.get("agent_end")(failedRun, fullCtx); +await handlers.get("agent_end")(completedRun, fullCtx); +await handlers.get("session_compact_failed")( + { reason: "overflow", aborted: false, willRetry: false, fromExtension: false }, + fullCtx, +); +await handlers.get("agent_settled")({}, fullCtx); +await new Promise((resolve) => setTimeout(resolve, 500)); +const legacy = readFrames(legacyFrames); +assert.ok(legacy.length > 0, "a protocol-1 control plane still receives the frames it understands"); +assert.strictEqual( + legacy.filter((frame) => frame.type === "condition").length, + 0, + "no condition frame may reach a wire that cannot carry one", +); +assert.ok( + legacy.some((frame) => frame.type === "state" && frame.state === "idle"), + "protocol 1 keeps its exact existing state frames", +); +assert.ok( + legacy.some((frame) => frame.type === "context"), + "protocol 1 keeps its exact existing context frames", +); + +// Phase 3: a control plane offering nothing this asset speaks. Refusing is the honest outcome — +// presence decays and the seat reads as unreachable — and it must be a NOTIFIED refusal, not a +// silent one. +const notices = []; +const foreignCtx = Object.create(fullCtx, { + ui: { value: { notify: (message) => notices.push(message) } }, +}); +const foreignFrames = path.join(dir, "frames-foreign.jsonl"); +process.env.ST2_SMOKE_HELLO = "foreign"; +process.env.ST2_SMOKE_FRAMES = foreignFrames; +await handlers.get("session_start")({}, foreignCtx); +await handlers.get("agent_start")({}, foreignCtx); +await handlers.get("agent_settled")({}, foreignCtx); +await new Promise((resolve) => setTimeout(resolve, 500)); +assert.strictEqual(readFrames(foreignFrames).length, 0, "a refused channel receives no frames"); +assert.ok( + notices.some((notice) => notice.includes("protocol")), + "a refused negotiation tells the operator why", +); + fs.rmSync(dir, { recursive: true, force: true }); console.log("pi extension smoke: ok"); process.exit(0); diff --git a/src/pi_channel.rs b/src/pi_channel.rs index d6a02539..37e240f1 100644 --- a/src/pi_channel.rs +++ b/src/pi_channel.rs @@ -35,10 +35,19 @@ inbox by reading, acting on, replying when useful, and archiving each handled me resuming or starting work, set your status to busy; set available only when yielding or ready for \ new work."; -/// The wire version the shipped extension is written against. A mismatch is the extension's to -/// refuse: st2 never guesses what an older asset understands. +/// The wire version the hello ADVERTISES as its floor, and the only one an already-published +/// asset understands. It stays literally 1 forever: the pinned v1 asset compares this field for +/// strict equality and closes the channel on a mismatch, so advertising a newer number here would +/// make every hook set in the field refuse the channel and silently stop all delivery. pub const PROTOCOL: u32 = 1; +/// Every wire version this build speaks, offered beside the floor so a NEWER asset can negotiate +/// up without the floor ever moving. The asset selects `max(supported ∩ protocols)` and refuses +/// only on an empty intersection; an older asset ignores the list and keeps reading `protocol`. +/// Version 2 adds the condition frame — nothing else — so an asset that stays on 1 keeps its +/// exact existing behaviour. +pub const PROTOCOLS: [u32; 2] = [1, 2]; + /// Last-resort durable state when compaction begins before the agent authored a checkpoint. /// /// The stable text deliberately carries no extension-owned path or clock. Rust owns both the @@ -88,12 +97,28 @@ pub struct ChannelKind { pub runtime_id_env: &'static str, pub session_env: &'static str, pub seq_env: &'static str, + /// What this kind's frames state on the version 3 ask axis when the wire claims no pending + /// ask. `HumanAsk::Unknown` is a POSITIVE statement — "this harness exposes no ask surface at + /// all", which is pi's case on every frame forever — and it is stronger than it looks: a kind + /// carrying it never derives a `pending` ask from any frame, because a channel that cannot see + /// the axis cannot vouch for a foreign asset's claim about it. `None` is the ordinary answer + /// for a kind that does see the axis (omp) and reports nothing waiting. + pub default_ask: harness_state::HumanAsk, + /// The conversation bridge this kind can prove from its OWN typed evidence, stated once here + /// because it is a property of the harness rather than of a frame. pi's is + /// `Unsupported`: none of its extension events and no pinned ctx surface carries a + /// conversation identity, and st2's own runtime ID and session token belong to st2's + /// namespace, so publishing them as a link would be a fabricated cross-namespace claim. + /// `None` leaves the axis to whatever the record already holds. + pub conversation: Option, } const PI_KIND: ChannelKind = ChannelKind { label: "pi", harness: harness_context::Harness::Pi, diagnostic_driver: None, + default_ask: harness_state::HumanAsk::Unknown, + conversation: Some(harness_state::ConversationState::Unsupported), runtime_id_env: crate::pi_session::CHANNEL_RUNTIME_ID, session_env: crate::pi_session::CHANNEL_SESSION, seq_env: crate::pi_session::CHANNEL_SEQ, @@ -103,6 +128,8 @@ const OMP_KIND: ChannelKind = ChannelKind { label: "omp", harness: harness_context::Harness::Omp, diagnostic_driver: Some(driver_diagnostic::Driver::Omp), + default_ask: harness_state::HumanAsk::None, + conversation: None, runtime_id_env: crate::omp_session::CHANNEL_RUNTIME_ID, session_env: crate::omp_session::CHANNEL_SESSION, seq_env: crate::omp_session::CHANNEL_SEQ, @@ -118,6 +145,22 @@ pub fn run_omp(catalog_root: &Path, identity: &str) -> Result<()> { run_for(catalog_root, identity, &OMP_KIND) } +/// The channel's opening frame. +/// +/// `protocol` is the FLOOR and `protocols` the offer. Both are load-bearing in opposite +/// directions: the pinned asset compares `protocol` for strict equality and closes the channel on +/// a mismatch — so raising it would make every already-published hook set refuse this channel and +/// silently stop all delivery — while a newer asset ignores it and negotiates over `protocols`. +fn hello(identity: &str, session_context: &str) -> Value { + json!({ + "type": "hello", + "protocol": PROTOCOL, + "protocols": PROTOCOLS, + "identity": identity, + "sessionContext": session_context, + }) +} + fn run_for(catalog_root: &Path, identity: &str, kind: &ChannelKind) -> Result<()> { let agent_dir = message::resolve_agent_dir(catalog_root, identity, &crate::run::detect_host())? .with_context(|| format!("{} channel agent '{identity}' is not declared", kind.label))?; @@ -134,15 +177,10 @@ fn run_for(catalog_root: &Path, identity: &str, kind: &ChannelKind) -> Result<() } }); let mut stdout = io::BufWriter::new(io::stdout().lock()); - write_json( - &mut stdout, - &json!({ - "type": "hello", - "protocol": PROTOCOL, - "identity": identity, - "sessionContext": session_context, - }), - )?; + // Named so the version negotiation is provable without a live channel: the floor and the + // offered set are the one thing in this file an already-published asset compares against, and + // getting them wrong stops all delivery silently. + write_json(&mut stdout, &hello(identity, &session_context))?; stdout.flush()?; // The channel owns the live half of observed harness state: it is the one process that sees // the harness's own turn events, and its stdio connection to the extension is the evidence @@ -236,6 +274,12 @@ fn channel_loop( let mut delivered = HashSet::new(); let label = kind.label; let mut next_heartbeat = Instant::now() + heartbeat_every; + // A raise the record had no observation of this session's to attach to. It is held rather + // than dropped and rather than published beside a fabricated activity: the seat is faulted, + // and the first genuine activity edge is what makes the fault recordable. Unreachable in + // practice — the extension seeds a state frame from `ctx.isIdle()` at open time — which is + // exactly why the fallback must not be a guess. + let mut deferred_fault: Option = None; loop { match input.recv_timeout(poll) { Ok(line) => { @@ -251,16 +295,66 @@ fn channel_loop( // The typed turn result, decoded once: it feeds two independent records and the // credential edge must not depend on the categorical write landing. let turn = frame.as_ref().and_then(turn_result); + // The categorical axis. Which SURFACE states it is decided by the version this + // writer emits, and nothing else: while the wire cannot carry a condition the + // legacy call is made verbatim, so activating version 3 is the only thing that + // changes these bytes. if let Some(observation) = frame .as_ref() .and_then(state_observation) .or_else(|| turn.as_ref().and_then(turn_observation)) + { // A queued live frame must never overwrite the wrapper's terminal record: // the channel and the wrapper are separate processes, so the flock alone // serializes but does not order their writes. - && let Err(error) = writer.observe_unless_ended(observation) + let recorded = if writer.writes_condition_axis() { + publish_frame(writer, kind_frame(kind, observation), label) + } else { + writer.observe_unless_ended(observation).map(|_landed| ()) + }; + if let Err(error) = recorded { + tracing::warn!( + "st2 {label} channel: recording observed state failed: {error}" + ); + } + // An activity edge is the one thing that can retire a deferred raise's + // reason for waiting: the record now holds an observation of this session's + // for the fault to attach to. + if let Some(fault) = deferred_fault.take() { + deferred_fault = apply_condition(writer, ConditionEdge::Raise(fault), label); + } + } + // The condition axis, on its own frame and deliberately without an activity + // claim: pi's fault evidence carries none, and restating activity here would + // refresh a stale one from a frame that observed nothing about it. An activity + // edge never clears a condition, and a condition edge never restates activity. + if let Some(edge) = frame + .as_ref() + .and_then(|frame| condition_frame(frame, message::now_ms())) { - tracing::warn!("st2 {label} channel: recording observed state failed: {error}"); + if writer.writes_condition_axis() { + // A clear the harness sent while a raise is still held retires that + // raise. Without this the held fault outlives the very edge that + // resolved it and the next activity edge republishes a condition the + // harness has already retracted. + if deferred_fault + .as_ref() + .is_some_and(|held| retires_deferred(&edge, held)) + { + deferred_fault = None; + } + if let Some(fault) = apply_condition(writer, edge, label) { + deferred_fault = Some(fault); + } + } else { + // Version 2 has nowhere to carry it and this record has exactly one + // source of truth, so the edge is dropped rather than cached where the + // session's sibling writers could not see it. + tracing::debug!( + "st2 {label} channel: condition frame dropped: this build's records \ + carry no condition axis" + ); + } } // The credential axis is a third record, independent of the numbers and of the // categorical state: a rejection stands until a turn reaches its ordinary end, @@ -357,6 +451,292 @@ fn state_observation(frame: &Value) -> Option { Some(observation) } +/// Project one decoded observation into the version 3 tuple this channel can vouch for. +/// +/// The condition axis is `Unchanged` on EVERY activity edge, and that is the whole ordering rule +/// of this adapter: a turn starting or a seat settling has learned nothing about whether the +/// provider is faulted, so a standing fault survives both. It is retired only by its own paired +/// clear, a positive success edge, a terminal record, or a new incarnation — never by activity. +/// The record therefore settles as `idle` beside a standing fault: activity honest, wedged seat +/// visible. +fn kind_frame(kind: &ChannelKind, observation: harness_state::Observation) -> harness_state::Frame { + let mut frame = harness_state::Frame::new( + observation.state, + observation.input_buffer, + harness_state::ConditionReport::Unchanged, + tagged_ask(kind, observation.blocked_on, observation.ask), + ); + if let Some(conversation) = kind.conversation.clone() { + frame = frame.with_conversation(conversation); + } + if let Some(reason) = observation.reason { + frame = frame.with_reason(reason); + } + if let Some(exit) = observation.exit { + frame = frame.with_exit(exit); + } + frame +} + +/// The tagged ask this kind can vouch for, from the legacy pair its own asset speaks. +/// +/// A kind whose harness exposes no ask surface states `Unknown` on every frame and derives a +/// `pending` ask from nothing: not from its own frames, which cannot carry one, and not from a +/// foreign asset's `blockedOn`/`ask` either — a channel that cannot see the axis cannot vouch for +/// somebody else's claim about it, and a phantom `pending` row injects a question into the ask +/// queue that nobody can answer. +fn tagged_ask( + kind: &ChannelKind, + blocked_on: harness_state::BlockedOn, + ask: harness_state::Ask, +) -> harness_state::HumanAsk { + if kind.default_ask == harness_state::HumanAsk::Unknown { + return harness_state::HumanAsk::Unknown; + } + match blocked_on { + harness_state::BlockedOn::Human => { + harness_state::HumanAsk::Pending(match ask { + harness_state::Ask::Permission => harness_state::AskKind::Permission, + harness_state::Ask::Question => harness_state::AskKind::Question, + harness_state::Ask::Review => harness_state::AskKind::Review, + // Blocked on a human without a nameable kind: the ask is real and its kind + // unstated, which is not the same as no ask at all. + harness_state::Ask::None | harness_state::Ask::Unknown => { + harness_state::AskKind::Unknown + } + }) + } + harness_state::BlockedOn::None => kind.default_ask, + harness_state::BlockedOn::Unknown => harness_state::HumanAsk::Unknown, + } +} + +/// What one `type: "condition"` frame asks of the condition axis. Three operations and no fourth: +/// there is no word here that ends a session, because a condition frame is never evidence that a +/// process exited — the outer session wrapper alone writes the terminal record. +#[derive(Debug, Clone, PartialEq, Eq)] +enum ConditionEdge { + Raise(harness_state::FaultReport), + ClearPaired(harness_state::FaultKey), + ClearAll(harness_state::ProgressProof), +} + +/// Decode a condition frame, protocol 2. +/// +/// Same discipline as [`state_observation`], for the same reason: an unrecognized `op`, a category +/// outside the closed set, a recovery word this build does not know, a `code` that is not +/// `provider/code`, and a `clearAll` whose proof word is missing or unknown each drop the frame +/// silently rather than guessing. A newer asset must not be able to make this channel record +/// something it cannot vouch for — and of the two failure directions, dropping an edge leaves the +/// record as it was while guessing one publishes a claim nobody made. +/// +/// The observation instant is stamped HERE, by Rust: the asset and this process share no clock, +/// so a semantic timestamp off the wire would be another harness's idea of now. +fn condition_frame(frame: &Value, observed_at_ms: u64) -> Option { + if frame.get("type").and_then(Value::as_str) != Some("condition") { + return None; + } + let op = frame.get("op").and_then(Value::as_str)?; + // The unkeyed clear names no fault, so it must name the progress it witnessed. There is + // deliberately no default: an unproven blanket clear is exactly how a fault that still holds + // stops paging. + if op == "clearAll" { + return Some(ConditionEdge::ClearAll( + match frame.get("proof").and_then(Value::as_str)? { + "turnCompleted" => harness_state::ProgressProof::TurnCompleted, + "requestSucceeded" => harness_state::ProgressProof::RequestSucceeded, + _ => return None, + }, + )); + } + let category = fault_category(frame.get("category").and_then(Value::as_str)?)?; + // Absent is a real answer (a fault, or a clear, carrying no code at all); present-but-not- + // namespaced is a malformed frame, because a bare word is how two providers' codes collide. + let code = match frame.get("code") { + None | Some(Value::Null) => None, + Some(Value::String(code)) + if code + .split_once('/') + .is_some_and(|(provider, rest)| !provider.is_empty() && !rest.is_empty()) => + { + Some(code.clone()) + } + Some(_) => return None, + }; + match op { + "raise" => { + let recovery = fault_recovery(frame.get("recovery").and_then(Value::as_str)?)?; + let mut fault = + harness_state::FaultReport::new(category, recovery, observed_at_ms); + if let Some(code) = code { + fault = fault.with_code(code); + } + // Diagnostic only, and the only place provider prose is allowed to ride: no consumer + // branches on it, and nothing above ever reclassifies the category from it. + if let Some(detail) = frame + .get("detail") + .and_then(Value::as_str) + .filter(|detail| !detail.is_empty()) + { + fault = fault.with_detail(detail); + } + Some(ConditionEdge::Raise(fault)) + } + // A paired clear names the EXACT `(category, code)` of the fault it clears — never the + // category alone, which is how one provider signal silences a different live fault in the + // same category. + "clear" => { + let mut key = harness_state::FaultKey::new(category); + if let Some(code) = code { + key = key.with_code(code); + } + Some(ConditionEdge::ClearPaired(key)) + } + _ => None, + } +} + +/// The closed category vocabulary, spelled out rather than inferred: a word outside it drops the +/// frame here instead of becoming an untyped fault, because a producer in this build has no +/// excuse for a category it cannot name. +fn fault_category(word: &str) -> Option { + Some(match word { + "authentication" => harness_state::FaultCategory::Authentication, + "account" => harness_state::FaultCategory::Account, + "quota" => harness_state::FaultCategory::Quota, + "rateLimit" => harness_state::FaultCategory::RateLimit, + "provider" => harness_state::FaultCategory::Provider, + "context" => harness_state::FaultCategory::Context, + "configuration" => harness_state::FaultCategory::Configuration, + "policy" => harness_state::FaultCategory::Policy, + "harness" => harness_state::FaultCategory::Harness, + _ => return None, + }) +} + +/// The closed recovery vocabulary. `unknown` is a word an asset may STATE — "the harness said +/// nothing about who clears this" — and it is never optimistic: it pages exactly like `human`. +fn fault_recovery(word: &str) -> Option { + Some(match word { + "automatic" => harness_state::Recovery::Automatic, + "human" => harness_state::Recovery::Human, + "terminal" => harness_state::Recovery::Terminal, + "unknown" => harness_state::Recovery::Unknown, + _ => return None, + }) +} + +/// Publish one activity edge, stating the condition axis once when the record has none. +/// +/// A version 3 record's condition axis is not writable as `absent` and silence is not a value +/// either, so the FIRST activity-only frame of an incarnation is refused with +/// [`harness_state::Refusal::Unstated`] — and pi's first frame is exactly that, the +/// `session_start` seed. Without the restatement a healthy pi seat would publish no record at +/// all, which reads as `unknown` for every consumer. +fn publish_frame( + writer: &mut harness_state::Writer, + frame: harness_state::Frame, + label: &str, +) -> Result<()> { + let outcome = writer.publish_unless_ended(frame.clone())?; + if let Some(stated) = restate_condition(&frame, &outcome) { + report_outcome( + label, + "observed state", + &writer.publish_unless_ended(stated)?, + ); + return Ok(()); + } + report_outcome(label, "observed state", &outcome); + Ok(()) +} + +/// The one frame this channel restates, and the one condition it may state to do it. +/// +/// `Unstated` is refused ONLY when this session's record holds no stated condition — a standing +/// fault of this session's makes the axis stated, so this can never overwrite one. That refusal +/// is therefore its own evidence: nothing this producer observed is faulted, which is exactly +/// what `clear` says. Every other outcome is left alone; in particular a mismatch, a terminal +/// record, or a later session's claim is never retried, because restating an activity edge +/// against those would be arguing with the record rather than observing the harness. +fn restate_condition( + frame: &harness_state::Frame, + outcome: &harness_state::WriteOutcome, +) -> Option { + if !matches!(outcome.refusal(), Some(harness_state::Refusal::Unstated)) { + return None; + } + let mut stated = frame.clone(); + stated.condition = harness_state::ConditionReport::Clear; + Some(stated) +} + +/// Whether one condition edge retires a raise that is still being held. +/// +/// A held raise is a fault the record has not accepted yet, so the ordinary clear paths cannot +/// reach it: it lives in this process. An unkeyed clear retires whatever stands, held included — +/// its proof is a positive observation that the harness made progress. A paired clear retires it +/// only on the EXACT key, the same rule the writer applies to a landed fault, so one signal +/// cannot silence a different held fault. A raise never retires another: replacement is +/// [`apply_condition`]'s business. +fn retires_deferred(edge: &ConditionEdge, held: &harness_state::FaultReport) -> bool { + match edge { + ConditionEdge::ClearAll(_) => true, + ConditionEdge::ClearPaired(key) => *key == held.key(), + ConditionEdge::Raise(_) => false, + } +} + +/// Apply one condition edge, returning a raise that found no observation of this session's to +/// attach to so the caller can hold it until the next activity edge. +/// +/// A condition operation never states activity: the writer carries the rest of the tuple — +/// including the pty session that vouches for it — across from this session's own record. +fn apply_condition( + writer: &mut harness_state::Writer, + edge: ConditionEdge, + label: &str, +) -> Option { + match edge { + ConditionEdge::Raise(fault) => match writer.raise_fault(fault.clone()) { + Ok(harness_state::WriteOutcome::Refused(harness_state::Refusal::Unobserved)) => { + tracing::debug!( + "st2 {label} channel: holding a raise until this session states an \ + observation for it to attach to" + ); + return Some(fault); + } + Ok(outcome) => report_outcome(label, "fault", &outcome), + Err(error) => tracing::warn!("st2 {label} channel: raising a fault failed: {error}"), + }, + ConditionEdge::ClearPaired(key) => match writer.clear_fault(key) { + Ok(outcome) => report_outcome(label, "paired clear", &outcome), + Err(error) => tracing::warn!("st2 {label} channel: clearing a fault failed: {error}"), + }, + ConditionEdge::ClearAll(proof) => match writer.clear_all(proof) { + Ok(outcome) => report_outcome(label, "clear", &outcome), + Err(error) => tracing::warn!("st2 {label} channel: clearing conditions failed: {error}"), + }, + } + None +} + +/// Log what a typed write outcome means. Landing and coalescing are both success and say nothing. +/// A paired clear that matched no standing fault is the ORDINARY outcome of a successful +/// compaction on a healthy seat, so it is debug and never a warning — and it never falls back to +/// an unkeyed clear, which would silence whatever else stands. +fn report_outcome(label: &str, what: &str, outcome: &harness_state::WriteOutcome) { + match outcome.refusal() { + None => {} + Some(harness_state::Refusal::ConditionMismatch { current }) => tracing::debug!( + "st2 {label} channel: {what} matched no standing fault (standing: {current:?})" + ), + Some(refusal) => { + tracing::warn!("st2 {label} channel: {what} was refused: {refusal:?}"); + } + } +} + /// The observed-state reason a rejected provider credential publishes, shared verbatim with the /// OpenCode, Codex, and Claude producers: one word for one class, whatever named it. const PROVIDER_AUTH_REASON: &str = "providerAuth"; @@ -875,7 +1255,9 @@ mod tests { } /// The wrapper's terminal record is the incarnation's last word: a live frame the extension - /// queued before dying must not resurrect the session after the wrapper reaped it. + /// queued before dying must not resurrect the session after the wrapper reaped it. That holds + /// for every axis — a queued condition edge is refused by the same terminal fence, and the + /// record's bytes are unchanged by the whole queue. #[test] fn a_queued_live_frame_never_overwrites_the_wrappers_terminal_record() { let tmp = tempfile::tempdir().unwrap(); @@ -895,8 +1277,13 @@ mod tests { let terminal = std::fs::read(&record).unwrap(); let (tx, rx) = mpsc::channel(); - tx.send(Ok(r#"{"type":"state","state":"idle"}"#.to_string())) - .unwrap(); + for frame in [ + r#"{"type":"state","state":"idle"}"#, + r#"{"type":"condition","op":"raise","category":"harness","code":"pi/assistantError","recovery":"unknown","detail":"401 Unauthorized"}"#, + r#"{"type":"condition","op":"clearAll","proof":"turnCompleted"}"#, + ] { + tx.send(Ok(frame.to_string())).unwrap(); + } drop(tx); let mut out = Vec::new(); channel_loop( @@ -934,7 +1321,9 @@ mod tests { .expect("a record must have been written") } - /// HC-R13, pinned to pi 0.84.2. The payload is verbatim from the credential-free pi lab: one + /// HC-R13, pinned to pi 0.84.4. The payload is verbatim from the credential-free pi lab + /// (captured on 0.84.2, and carried forward because `getContextUsage()` and + /// `calculateContextTokens` are byte-identical in the 0.84.4 tarball): one /// `message_end` for an assistant message, with `getContextUsage()` and the message's own /// `usage` side by side. /// @@ -946,11 +1335,11 @@ mod tests { /// pi reports a float that runs far above 100 when a turn overruns the window, and a producer /// or reader that clamped it would hide exactly the saturation this record exists to show. #[test] - fn the_pi_0_84_2_fixture_pins_total_tokens_as_the_numerator() { + fn the_pinned_pi_fixture_pins_total_tokens_as_the_numerator() { const MEASURED: &str = crate::pi_session::MEASURED_CONTEXT_VERSION; assert_eq!( - MEASURED, "0.84.2", - "the fixture below was captured on this build" + MEASURED, "0.84.4", + "the fixture below is pinned to the build the flake type-checks against" ); // Verbatim `event.message.usage` from the same event, carried so the assertion below can // name the number this producer must NOT publish. @@ -1538,5 +1927,459 @@ mod tests { driver_diagnostic::read(&driver_diagnostic::path(agent_dir)), driver_diagnostic::Observed::Absent ); + // And the fault axis draws the same line. pi's error-ended turn arrives as prose that + // says `401`, and the category it decodes to is `harness` — the honest reading of "a pi + // turn failed" — never `authentication`, which would be inferred from that very string. + let raise = condition_frame( + &json!({"type":"condition","op":"raise","category":"harness", + "code":"pi/assistantError","recovery":"unknown", + "detail":"401 Unauthorized: invalid x-api-key"}), + 1_000, + ) + .unwrap(); + let ConditionEdge::Raise(fault) = &raise else { + panic!("an error tail raises a fault: {raise:?}"); + }; + assert_eq!(fault.category, harness_state::FaultCategory::Harness); + assert_ne!( + fault.category, + harness_state::FaultCategory::Authentication, + "pi has no error-classification field, so a credential verdict here would be prose" + ); + assert_eq!( + fault.detail.as_deref(), + Some("401 Unauthorized: invalid x-api-key"), + "the prose is carried, diagnostically, and nothing branches on it" + ); + } + + /// The measured false idle, and its fix. `agent_end` raises before `agent_settled` reports the + /// idle, and an activity edge carries the condition axis forward `Unchanged`, so the record + /// settles as idle BESIDE a standing fault instead of laundering a wedged seat into a clean + /// yield. + #[test] + fn a_failed_pi_turn_publishes_idle_beside_a_standing_fault() { + let raise = condition_frame( + &json!({"type":"condition","op":"raise","category":"harness", + "code":"pi/assistantError","recovery":"unknown","detail":"provider said no"}), + 4_200, + ) + .unwrap(); + let ConditionEdge::Raise(fault) = &raise else { + panic!("an error tail raises a fault: {raise:?}"); + }; + assert_eq!(fault.category, harness_state::FaultCategory::Harness); + assert_eq!(fault.code.as_deref(), Some("pi/assistantError")); + assert_eq!(fault.recovery, harness_state::Recovery::Unknown); + assert_ne!( + fault.recovery, + harness_state::Recovery::Automatic, + "an unsayable recovery must page, never wait for a deadline nobody set" + ); + assert_eq!( + fault.next_observation_due_ms, None, + "pi declares no deadline, and st2 never synthesizes one" + ); + // The semantic clock is Rust's, stamped when the frame was read: the asset and this + // process share no clock. + assert_eq!(fault.observed_at_ms, 4_200); + + // The idle that follows. It states activity and nothing else. + let settled = kind_frame( + &PI_KIND, + state_observation(&json!({"type":"state","state":"idle"})).unwrap(), + ); + assert_eq!(settled.state, harness_state::Activity::Idle); + assert_eq!( + settled.condition, + harness_state::ConditionReport::Unchanged, + "the idle must carry the standing fault forward, not clear it" + ); + assert_eq!(settled.ask, harness_state::HumanAsk::Unknown); + assert_eq!( + settled.conversation, + Some(harness_state::ConversationState::Unsupported) + ); + assert_eq!( + settled.reason, None, + "the prose belongs to the fault's detail, not to the activity axis" + ); + } + + /// A version 3 record's condition axis is not writable as `absent`, so the first activity-only + /// frame of an incarnation — pi's `session_start` seed — is refused as `Unstated`. That + /// refusal is its own evidence (a standing fault of this session's would have STATED the + /// axis), so the frame is restated once as `clear` and nothing else about it moves. Every + /// other outcome is left exactly alone: a retry there would argue with the record instead of + /// observing the harness. + #[test] + fn a_fresh_seats_first_frame_states_the_condition_axis_once() { + let seed = kind_frame( + &PI_KIND, + state_observation(&json!({"type":"state","state":"idle"})).unwrap(), + ); + assert_eq!(seed.condition, harness_state::ConditionReport::Unchanged); + + let restated = restate_condition( + &seed, + &harness_state::WriteOutcome::Refused(harness_state::Refusal::Unstated), + ) + .expect("an unstated axis must be stated once"); + assert_eq!(restated.condition, harness_state::ConditionReport::Clear); + assert_eq!(restated.state, seed.state, "the activity axis is unchanged"); + assert_eq!(restated.ask, seed.ask); + assert_eq!(restated.conversation, seed.conversation); + assert_eq!(restated.exit, None, "restating never terminates a record"); + + for outcome in [ + harness_state::WriteOutcome::Landed, + harness_state::WriteOutcome::Coalesced, + harness_state::WriteOutcome::Refused(harness_state::Refusal::Terminal), + harness_state::WriteOutcome::Refused(harness_state::Refusal::Unobserved), + harness_state::WriteOutcome::Refused(harness_state::Refusal::Unfenced), + harness_state::WriteOutcome::Refused(harness_state::Refusal::Superseded { + on_disk_seq: 4, + ours: 3, + }), + harness_state::WriteOutcome::Refused(harness_state::Refusal::ConditionMismatch { + current: None, + }), + ] { + assert_eq!( + restate_condition(&seed, &outcome), + None, + "outcome: {outcome:?}" + ); + } + } + + /// An activity edge has learned nothing about the provider, so it may not clear a condition — + /// in either direction, and however many of them arrive. + #[test] + fn an_activity_edge_never_clears_a_standing_pi_fault() { + for word in ["active", "idle"] { + let frame = json!({"type":"state","state":word}); + let observation = state_observation(&frame).unwrap(); + assert_eq!( + kind_frame(&PI_KIND, observation).condition, + harness_state::ConditionReport::Unchanged, + "state:{word} must not state the condition axis" + ); + assert_eq!( + condition_frame(&frame, 1_000), + None, + "a state frame is not a condition edge" + ); + } + // Nor is the numeric axis: a compaction reading says nothing about a fault either. + assert_eq!( + condition_frame( + &json!({"type":"context","reading":{"usedTokens":10,"usedPercent":1.0}}), + 1_000 + ), + None + ); + } + + /// pi's one positive success edge. An unkeyed clear must name the progress it witnessed, so a + /// `clearAll` whose proof word is missing or unrecognized is dropped rather than believed: + /// that frame is the only operation that can silence a fault nobody saw resolve. + #[test] + fn a_clean_agent_end_is_pis_only_clear_all() { + assert_eq!( + condition_frame(&json!({"type":"condition","op":"clearAll","proof":"turnCompleted"}), 1), + Some(ConditionEdge::ClearAll( + harness_state::ProgressProof::TurnCompleted + )) + ); + for frame in [ + json!({"type":"condition","op":"clearAll"}), + json!({"type":"condition","op":"clearAll","proof":"probably"}), + json!({"type":"condition","op":"clearAll","proof":"assumed"}), + json!({"type":"condition","op":"clearAll","proof":null}), + json!({"type":"condition","op":"settled","proof":"turnCompleted"}), + ] { + assert_eq!(condition_frame(&frame, 1), None, "frame: {frame}"); + } + } + + /// CX-1, pinned for pi: a paired clear names the EXACT `(category, code)` pair. The compaction + /// success edge clears its own failure and nothing else — a category-only key, or one carrying + /// a sibling's code, is a different key and the writer refuses it as a mismatch. + #[test] + fn a_compaction_failure_is_cleared_only_by_its_own_key() { + let raised = condition_frame( + &json!({"type":"condition","op":"raise","category":"context", + "code":"pi/session_compact_failed","recovery":"human"}), + 9_000, + ) + .unwrap(); + let ConditionEdge::Raise(fault) = &raised else { + panic!("a compaction failure raises a fault: {raised:?}"); + }; + assert_eq!(fault.category, harness_state::FaultCategory::Context); + assert_eq!(fault.recovery, harness_state::Recovery::Human); + assert_eq!(fault.detail, None, "the typed event needs no prose"); + + let cleared = condition_frame( + &json!({"type":"condition","op":"clear","category":"context", + "code":"pi/session_compact_failed"}), + 9_100, + ) + .unwrap(); + assert_eq!(cleared, ConditionEdge::ClearPaired(fault.key())); + let ConditionEdge::ClearPaired(key) = cleared else { + panic!("a compaction success is a paired clear"); + }; + // The two keys a lazier producer would have sent instead. Neither matches the fault, so + // neither can silence it. + assert_ne!( + key, + harness_state::FaultKey::new(harness_state::FaultCategory::Context), + "a category-only clear would wipe any other context fault" + ); + assert_ne!( + key, + harness_state::FaultKey::new(harness_state::FaultCategory::Context) + .with_code("pi/session_compact") + ); + // A code that is not provider-namespaced is a malformed frame, not a bare-word key: two + // providers' codes collide exactly there. + assert_eq!( + condition_frame( + &json!({"type":"condition","op":"clear","category":"context", + "code":"session_compact_failed"}), + 9_100 + ), + None + ); + // And a category this build cannot name drops the frame rather than becoming untyped. + assert_eq!( + condition_frame( + &json!({"type":"condition","op":"raise","category":"vibes", + "recovery":"human"}), + 9_100 + ), + None + ); + // As does a recovery word it cannot name: the urgency axis is what routes the fault. + assert_eq!( + condition_frame( + &json!({"type":"condition","op":"raise","category":"context", + "recovery":"eventually"}), + 9_100 + ), + None + ); + } + + /// A raise the record could not accept yet lives in this process, so the ordinary clear paths + /// cannot reach it. An unkeyed clear and its own paired key must retire it anyway, or the held + /// fault outlives the very edge that resolved it and the next activity edge republishes a + /// condition the harness already retracted. + #[test] + fn a_clear_retires_a_raise_that_has_not_landed_yet() { + let held = harness_state::FaultReport::new( + harness_state::FaultCategory::Context, + harness_state::Recovery::Human, + 5_000, + ) + .with_code("pi/session_compact_failed"); + + assert!(retires_deferred( + &ConditionEdge::ClearAll(harness_state::ProgressProof::TurnCompleted), + &held + )); + assert!(retires_deferred( + &ConditionEdge::ClearPaired(held.key()), + &held + )); + // A different key is a different fault: it may not silence this one, held or standing. + assert!(!retires_deferred( + &ConditionEdge::ClearPaired(harness_state::FaultKey::new( + harness_state::FaultCategory::Context + )), + &held + )); + assert!(!retires_deferred( + &ConditionEdge::ClearPaired( + harness_state::FaultKey::new(harness_state::FaultCategory::Harness) + .with_code("pi/assistantError") + ), + &held + )); + // Replacing one raise with another is the writer's business, not a retirement. + assert!(!retires_deferred( + &ConditionEdge::Raise( + harness_state::FaultReport::new( + harness_state::FaultCategory::Harness, + harness_state::Recovery::Unknown, + 6_000, + ) + .with_code("pi/assistantError") + ), + &held + )); + } + + /// pi has no ask surface at all, so every pi frame states `unknown` — positively, forever — + /// and derives a pending ask from nothing, not even from a foreign asset's claim of one. A + /// fabricated `pending` injects a question into the ask queue that nobody can answer. + #[test] + fn pi_frames_never_claim_an_ask() { + assert_eq!(PI_KIND.default_ask, harness_state::HumanAsk::Unknown); + assert_eq!(OMP_KIND.default_ask, harness_state::HumanAsk::None); + + let foreign = json!({"type":"state","state":"active","blockedOn":"human", + "ask":"permission"}); + let observation = state_observation(&foreign).unwrap(); + assert_eq!( + kind_frame(&PI_KIND, observation.clone()).ask, + harness_state::HumanAsk::Unknown, + "a channel that cannot see the axis cannot vouch for somebody else's claim about it" + ); + assert_eq!( + kind_frame(&PI_KIND, observation.clone()).ask.pending(), + None + ); + // The same frame on a kind that DOES see the axis reads it, which is what makes the pi + // answer a statement about pi rather than a decoder that lost the field. + assert_eq!( + kind_frame(&OMP_KIND, observation).ask, + harness_state::HumanAsk::Pending(harness_state::AskKind::Permission) + ); + assert_eq!( + kind_frame( + &PI_KIND, + state_observation(&json!({"type":"state","state":"idle"})).unwrap() + ) + .ask, + harness_state::HumanAsk::Unknown + ); + } + + /// The channel writes no terminal record on any axis. `src/pi_session.rs` — the outer wrapper, + /// the only process that sees the provider die — owns that word, so neither an `ended` state + /// word nor a fault claiming terminal recovery may produce one. + #[test] + fn no_pi_frame_can_write_ended() { + let tmp = tempfile::tempdir().unwrap(); + let agent_dir = tmp.path(); + std::fs::create_dir_all(message::inbox_dir(agent_dir)).unwrap(); + let mut writer = + harness_state::Writer::new(agent_dir, "h.worker", "pi", Some("h.worker".into())); + let (tx, rx) = mpsc::channel(); + for frame in [ + r#"{"type":"state","state":"active"}"#, + r#"{"type":"state","state":"ended","exit":"signal 9"}"#, + r#"{"type":"condition","op":"raise","category":"harness","code":"pi/assistantError","recovery":"terminal"}"#, + ] { + tx.send(Ok(frame.to_string())).unwrap(); + } + drop(tx); + channel_loop( + &rx, + &mut Vec::new(), + &message::inbox_dir(agent_dir), + agent_dir, + &mut writer, + None, + "h.worker", + &PI_KIND, + Duration::from_millis(1), + Duration::from_secs(60), + ) + .unwrap(); + + let raw: Value = serde_json::from_slice( + &std::fs::read(harness_state::harness_state_path(agent_dir)).unwrap(), + ) + .unwrap(); + assert_eq!(raw["state"], "active", "no channel frame may terminate a session"); + assert!(raw["exit"].is_null()); + // The decoder itself has no terminal operation: `terminal` is a recovery word — nothing + // clears this fault for this incarnation — and never a session outcome. + let edge = condition_frame( + &json!({"type":"condition","op":"raise","category":"harness", + "code":"pi/assistantError","recovery":"terminal"}), + 1_000, + ) + .unwrap(); + assert!(matches!(edge, ConditionEdge::Raise(_))); + } + + /// Protocol 1 acceptance, which is what keeps every already-published hook set delivering + /// mail: the pinned asset compares `protocol` for strict equality and closes the channel on a + /// mismatch, so the floor stays 1 and the newer wire is offered beside it. + #[test] + fn protocol_one_assets_still_accept_the_hello() { + let hello = hello("h.worker", "restored"); + assert_eq!(hello["protocol"], json!(1)); + assert_eq!(PROTOCOL, 1); + assert_eq!(hello["protocols"], json!([1, 2])); + assert_eq!(PROTOCOLS, [1, 2]); + assert!( + PROTOCOLS.contains(&PROTOCOL), + "the floor must be one of the offered versions" + ); + assert_eq!(hello["identity"], json!("h.worker")); + assert_eq!(hello["sessionContext"], json!("restored")); + } + + /// While this build's writer emits version 2 the condition axis has nowhere to live, and this + /// record has exactly one source of truth — so a condition edge is dropped, not cached, and + /// the legacy bytes are exactly the ones this channel already wrote. + #[test] + fn condition_frames_change_nothing_while_the_record_carries_no_condition_axis() { + let tmp = tempfile::tempdir().unwrap(); + let agent_dir = tmp.path(); + std::fs::create_dir_all(message::inbox_dir(agent_dir)).unwrap(); + let record = harness_state::harness_state_path(agent_dir); + let run = |frames: &[&str], writer: &mut harness_state::Writer| { + let (tx, rx) = mpsc::channel(); + for frame in frames { + tx.send(Ok((*frame).to_string())).unwrap(); + } + drop(tx); + channel_loop( + &rx, + &mut Vec::new(), + &message::inbox_dir(agent_dir), + agent_dir, + writer, + None, + "h.worker", + &PI_KIND, + Duration::from_millis(1), + Duration::from_secs(60), + ) + .unwrap(); + }; + let mut writer = + harness_state::Writer::new(agent_dir, "h.worker", "pi", Some("h.worker".into())); + assert!( + !writer.writes_condition_axis(), + "this build's writer emits version 2; the branch under test is the legacy one" + ); + run(&[r#"{"type":"state","state":"idle"}"#], &mut writer); + let legacy = std::fs::read(&record).unwrap(); + + run( + &[ + r#"{"type":"condition","op":"raise","category":"harness","code":"pi/assistantError","recovery":"unknown","detail":"provider said no"}"#, + r#"{"type":"condition","op":"clear","category":"context","code":"pi/session_compact_failed"}"#, + r#"{"type":"condition","op":"clearAll","proof":"turnCompleted"}"#, + ], + &mut writer, + ); + assert_eq!( + std::fs::read(&record).unwrap(), + legacy, + "a condition edge may not touch the version 2 record" + ); + let raw: Value = serde_json::from_slice(&legacy).unwrap(); + assert_eq!(raw["state"], "idle"); + assert!(raw["condition"].is_null()); + assert!(raw["blockedOn"] == json!("none") || raw["blockedOn"].is_null()); } } diff --git a/src/pi_session.rs b/src/pi_session.rs index bb59c25c..2fe10850 100644 --- a/src/pi_session.rs +++ b/src/pi_session.rs @@ -32,7 +32,13 @@ const EXTENSION: &str = "pi-channel.ts"; /// type-checks and runtime-smokes the shipped asset against exactly this tarball. That makes the /// flake pin the gate for this constant, and `pi_channel`'s fixture asserts the two agree: a pi /// bump that changes what `getContextUsage().tokens` means must move both together or fail. -pub const MEASURED_CONTEXT_VERSION: &str = "0.84.2"; +/// +/// Moved 0.84.2 -> 0.84.4 with the flake pin, and the arithmetic was re-verified rather than +/// assumed: `getContextUsage()` (`dist/core/agent-session.js`) and `calculateContextTokens` +/// (`dist/core/compaction/compaction.js:86-88`, `usage.totalTokens || input + output + cacheRead +/// + cacheWrite`) are byte-identical between the two published tarballs, so the recorded fixture +/// still means what it says under the newer pin. +pub const MEASURED_CONTEXT_VERSION: &str = "0.84.4"; /// The exact st2 executable the pi extension must spawn for its channel. pub const CHANNEL_BIN: &str = "ST2_PI_CHANNEL_BIN";