refactor(ding): split the composer into per-harness adapters

Pure restructure: no behaviour change, no test renamed, no assertion altered.

`ding.rs` had grown to hold transport, delivery state, and both harnesses'
screen grammars in one file, with routing expressed as an ordered preference
between them. The preceding commit replaced that preference with a positional
rule — classify the lowest composer, because scrollback is above the live one
by construction. This gives that rule a structural home.

  src/ding/
    mod.rs          transport, pending notices, DingConfig, run_ding, serve
    composer.rs     ComposerState, the router, strip_ansi, soft-wrap candidates
    harness/
      mod.rs        trait Harness + Located + the registry
      claude.rs     locate / classify / idle footer / blocked
      codex.rs      locate / classify / idle footer / blocked

The router is now `max_by_key(row)` over every registered harness, so the
positional rule is one comparison rather than a special case per harness pair,
and a third harness cannot reintroduce an ordering. Claude is registered last
so an exact row tie still resolves to Claude, as the explicit comparison did.

Active-turn and modal detection moves to the harnesses, since those shapes are
harness-specific TUI chrome rather than a shared contract. The shared predicate
was already Claude-flavoured — `Create a plan?`, `Our systems are thinking a
bit more`, and `Retry with a faster model` were applied to Codex panes too. It
is duplicated verbatim into both adapters rather than narrowed, so this commit
changes nothing; narrowing the Codex side is separate work. Only the
choice-menu heuristic, which is genuinely cross-harness, stays in `composer`.

INVARIANTS.md names tests by file path, so its seventeen `src/ding.rs::`
references are repointed to `src/ding/mod.rs::`. Every test stayed in that
module: the screen fixtures are shared between the composer assertions and the
transport ones, so splitting them would have meant duplicating fixtures or
exporting them across modules, and neither is worth it for a move.

Verified rather than assumed:
  - `cargo test --lib -- --list` is byte-identical before and after (142 tests).
  - All 27 `src/<file>.rs::<test>` references in INVARIANTS.md resolve, checked
    both by grepping the named path for `fn <name>` and by running each one
    with `cargo test <full::path> -- --exact` (27 resolved, 27 passed).
  - The two real captures behind the preceding commits classify identically.

agent-session-id: e5217740-eef6-48eb-a794-3e5e11939e4d
agent-tool: Claude Code
agent-tool-version: 2.1.220
agent-model: claude-opus-5
agent-runtime-profile: /nix/store/i8y8b542cyqi385ywcjw5fvsq24f75v4-coding-agent-runtime-profile/share/coding-agents/profile.json
agent-skills-manifest: /nix/store/i81qxhzlrzcxrrdwpp6i8hagka2gby8y-agent-skills-corpus/share/agent-skills/manifest.json
tooling-profile: dotfiles@unknown-dirty

Author: schickling-assistant

INVARIANTS.md

@@ -10,10 +10,10 @@ materialization, messaging, DING, or presence must preserve them.
 | **Clean exec teardown** | Killing an exec task reaps its whole process group. | `tests/exec_backend.rs::exec_kill_reaps_the_whole_process_group_not_just_the_leader` |
 | **Bounded restart diagnostics** | Relaunching an exec task preserves the just-finished log as one prior generation while bounding retained diagnostics to current plus prior. Final retirement removes the PID and both logs. | `tests/exec_backend.rs::exec_restart_reap_keeps_bounded_diagnostics_and_final_remove_cleans_them`; `tests/run.rs::up_once_finally_removes_dead_retired_tasks_without_restarting_them` |
 | **Exactly-once-safe native bus** | Messages use stable `<unix-ms>-<rand6>.md` files. An archive filename is a durable receipt that shadows and cleans restored inbox replicas and makes repeated archive cleanup idempotent. | `src/message.rs::filename_grammar`; `src/message.rs::archive_receipt_suppresses_and_idempotently_cleans_a_restored_inbox_copy`; `tests/message.rs` |
-| **Fail-closed observed native DING** | Each unread message becomes one normalized `[DING]` frame. A maintained Codex or Claude composer must be positively empty before a bracketed paste, then show the exact notice in two immediately adjacent inspections before a separate bare Return. Human, modal, active, changed, timed-out, and unknown states never receive Return. Once paste starts, inspect-only staged ownership prevents duplicate paste across command failures, archive races, and restart adoption. Startup backlog otherwise becomes one generic recovery DING; new arrivals remain FIFO; `busy` delivers immediately; only fresh `dnd` defers. | `src/ding.rs::poke_text_normalizes_and_bounds_untrusted_fields`; `src/ding.rs::malicious_controls_cannot_escape_the_single_paste_frame`; `src/ding.rs::pty_stage_and_submit_are_separate_exact_sequences`; `src/ding.rs::maintained_composer_classifiers_require_exact_idle_state`; `src/ding.rs::paste_then_two_exact_observations_precede_return`; `src/ding.rs::changed_modal_ambiguous_and_bounded_timeout_never_return`; `src/ding.rs::final_observation_change_and_staged_retry_are_fail_closed`; `src/ding.rs::staged_ownership_survives_archive_and_never_repastes`; `src/ding.rs::pty_commands_have_a_real_outer_timeout`; `src/ding.rs::session_watch_has_startup_grace_debounce_and_live_reset`; `src/ding.rs::new_arrivals_is_fifo_and_archive_receipts_prevent_reding`; `src/ding.rs::pending_delivery_ignores_busy_but_respects_fresh_dnd_archive_and_retry`; `src/ding.rs::startup_recovery_notice_retries_in_memory`; `src/ding.rs::startup_backlog_gets_one_generic_recovery_then_new_arrivals_poke` |
-| **Mutation-only filesystem wakeups** | Supervisor and DING filesystem watchers ignore read/open access events and wake early only for create, modify, rename, or remove events. Their own catalog and inbox reads therefore cannot bypass the bounded timer cadence or form a Linux inotify CPU loop. | `src/watch.rs::only_mutations_wake_watch_loops`; `src/watch.rs::linux_reads_are_silent_but_real_mutations_wake`; `src/ding.rs::idle_ding_does_not_spin_on_its_own_inbox_reads`; `src/run.rs::idle_supervisor_does_not_spin_on_its_own_catalog_reads` |
-| **Bounded DING PTY probe churn** | An unsafe or active composer retains its FIFO notice but deferred delivery retries use a bounded backoff, so each inbox poll cannot spawn another short-lived PTY probe. | `src/ding.rs::deferred_delivery_backoff_bounds_short_lived_pty_attempts` |
-| **Agent-declared presence discipline** | The shipped bus contract requires agents to declare `busy` before executing work, use `available` only while yielding or ready, and reserve `dnd` for an explicit hold. Both native harnesses materialize that contract. Busy remains observable but does not suppress DING; fresh `dnd` is the only delivery gate. | `tests/compile_agent.rs::compile_agent_generates_claude_then_materializes_verbatim_persona`; `tests/compile_agent.rs::compile_agent_generates_codex_then_materializes_composed_agents_md`; `src/ding.rs::pending_delivery_ignores_busy_but_respects_fresh_dnd_archive_and_retry` |
+| **Fail-closed observed native DING** | Each unread message becomes one normalized `[DING]` frame. A maintained Codex or Claude composer must be positively empty before a bracketed paste, then show the exact notice in two immediately adjacent inspections before a separate bare Return. Human, modal, active, changed, timed-out, and unknown states never receive Return. Once paste starts, inspect-only staged ownership prevents duplicate paste across command failures, archive races, and restart adoption. Startup backlog otherwise becomes one generic recovery DING; new arrivals remain FIFO; `busy` delivers immediately; only fresh `dnd` defers. | `src/ding/mod.rs::poke_text_normalizes_and_bounds_untrusted_fields`; `src/ding/mod.rs::malicious_controls_cannot_escape_the_single_paste_frame`; `src/ding/mod.rs::pty_stage_and_submit_are_separate_exact_sequences`; `src/ding/mod.rs::maintained_composer_classifiers_require_exact_idle_state`; `src/ding/mod.rs::paste_then_two_exact_observations_precede_return`; `src/ding/mod.rs::changed_modal_ambiguous_and_bounded_timeout_never_return`; `src/ding/mod.rs::final_observation_change_and_staged_retry_are_fail_closed`; `src/ding/mod.rs::staged_ownership_survives_archive_and_never_repastes`; `src/ding/mod.rs::pty_commands_have_a_real_outer_timeout`; `src/ding/mod.rs::session_watch_has_startup_grace_debounce_and_live_reset`; `src/ding/mod.rs::new_arrivals_is_fifo_and_archive_receipts_prevent_reding`; `src/ding/mod.rs::pending_delivery_ignores_busy_but_respects_fresh_dnd_archive_and_retry`; `src/ding/mod.rs::startup_recovery_notice_retries_in_memory`; `src/ding/mod.rs::startup_backlog_gets_one_generic_recovery_then_new_arrivals_poke` |
+| **Mutation-only filesystem wakeups** | Supervisor and DING filesystem watchers ignore read/open access events and wake early only for create, modify, rename, or remove events. Their own catalog and inbox reads therefore cannot bypass the bounded timer cadence or form a Linux inotify CPU loop. | `src/watch.rs::only_mutations_wake_watch_loops`; `src/watch.rs::linux_reads_are_silent_but_real_mutations_wake`; `src/ding/mod.rs::idle_ding_does_not_spin_on_its_own_inbox_reads`; `src/run.rs::idle_supervisor_does_not_spin_on_its_own_catalog_reads` |
+| **Bounded DING PTY probe churn** | An unsafe or active composer retains its FIFO notice but deferred delivery retries use a bounded backoff, so each inbox poll cannot spawn another short-lived PTY probe. | `src/ding/mod.rs::deferred_delivery_backoff_bounds_short_lived_pty_attempts` |
+| **Agent-declared presence discipline** | The shipped bus contract requires agents to declare `busy` before executing work, use `available` only while yielding or ready, and reserve `dnd` for an explicit hold. Both native harnesses materialize that contract. Busy remains observable but does not suppress DING; fresh `dnd` is the only delivery gate. | `tests/compile_agent.rs::compile_agent_generates_claude_then_materializes_verbatim_persona`; `tests/compile_agent.rs::compile_agent_generates_codex_then_materializes_composed_agents_md`; `src/ding/mod.rs::pending_delivery_ignores_busy_but_respects_fresh_dnd_archive_and_retry` |
 | **Stable roster JSON** | `st2 agents --json [--enrich]` preserves field names, order, null handling, presence, explicit retirement state, activity, and inbox counts. Human output marks retired declarations without changing active rows. | `src/agents.rs::agents_json_has_stable_wire_shape`; `tests/status_agents.rs::roster_json_and_human_output_distinguish_retirement_from_presence` |
 | **Agent-declared presence** | Refresh preserves non-DND declared status and only advances liveness; a missing status starts as `available`, while `dnd` is never refreshed and an unrefreshed declaration ages to `unknown`. | `src/status.rs::refresh_preserves_value_and_bumps_mtime`; `src/status.rs::refresh_leaves_dnd_to_age_out`; `src/status.rs::refresh_missing_writes_available_default`; `src/status.rs::stale_mtime_reads_as_unknown_regardless_of_contents` |
 | **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` |

src/ding/composer.rs

@@ -0,0 +1,159 @@
+//! What a peeked screen proves about one exact notice, and the routing that decides which
+//! composer on the screen is the live one.
+
+use super::harness::{self, Screen};
+
+/// What the current bottom composer proves about one exact normalized notice.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum ComposerState {
+    /// A maintained harness is positively idle and contains only its known placeholder.
+    EmptySafe,
+    /// The exact notice is the complete composer and the harness is positively idle.
+    ExactSafe,
+    /// The exact notice is present, but a modal, active turn, or non-idle footer blocks Return.
+    ExactBlocked,
+    /// A maintained composer contains different text (including a human draft).
+    Changed,
+    /// No maintained, unambiguous composer state was proven.
+    Ambiguous,
+}
+
+
+/// Enumerate the two logical strings possible at each renderer-shaped soft-wrap row: the TUI either
+/// discarded one inter-word space or split a token. Current 80-column Codex/Claude composers wrap
+/// long DING rows at 70+ content cells and indent continuations by exactly two cells. Short or
+/// unfamiliar multiline input remains literal and cannot equal a normalized single-line DING.
+pub(super) fn logical_soft_wrap_candidates(input: &str, minimum_first_content_chars: usize) -> Vec<String> {
+    let rows: Vec<&str> = input.lines().collect();
+    let Some(first) = rows.first() else {
+        return vec![String::new()];
+    };
+    if rows.len() == 1 {
+        return vec![(*first).to_string()];
+    }
+    let mut candidates = vec![(*first).to_string()];
+    let mut previous = *first;
+    for (index, row) in rows[1..].iter().enumerate() {
+        let required_previous_width = minimum_first_content_chars + usize::from(index > 0) * 2;
+        if previous.chars().count() < required_previous_width
+            || !row.starts_with("  ")
+            || row.trim().is_empty()
+        {
+            return vec![input.to_string()];
+        }
+        let continuation = row.strip_prefix("  ").expect("prefix checked").trim_end();
+        let mut next = Vec::with_capacity(candidates.len().saturating_mul(2).min(32));
+        for candidate in candidates {
+            if next.len() >= 32 {
+                return vec![input.to_string()];
+            }
+            next.push(format!("{candidate}{continuation}"));
+            next.push(format!("{candidate} {continuation}"));
+        }
+        candidates = next;
+        previous = row;
+    }
+    candidates
+}
+
+
+pub(super) fn looks_like_choice_menu(plain: &str) -> bool {
+    let mut first = false;
+    let mut later = false;
+    for line in plain.lines().map(str::trim_start) {
+        first |= line.starts_with("› 1.") || line.starts_with("> 1.");
+        later |= line.starts_with("2.") || line.starts_with("3.");
+    }
+    first && later
+}
+
+
+/// Strip the CSI/OSC sequences emitted by `pty peek` while preserving rendered text. Bounded
+/// cursor-forward sequences represent visible spaces in current Codex and Claude panes.
+pub(super) fn strip_ansi(input: &str) -> String {
+    let bytes = input.as_bytes();
+    let mut out = String::with_capacity(input.len());
+    let mut index = 0;
+    while index < bytes.len() {
+        if bytes[index] != 0x1b {
+            let ch = input[index..].chars().next().expect("valid UTF-8 boundary");
+            out.push(ch);
+            index += ch.len_utf8();
+            continue;
+        }
+        index += 1;
+        if index >= bytes.len() {
+            break;
+        }
+        match bytes[index] {
+            b'[' => {
+                index += 1;
+                let params_start = index;
+                let mut final_byte = None;
+                while index < bytes.len() {
+                    let byte = bytes[index];
+                    index += 1;
+                    if (0x40..=0x7e).contains(&byte) {
+                        final_byte = Some(byte);
+                        break;
+                    }
+                }
+                if final_byte == Some(b'C') {
+                    let params = &bytes[params_start..index.saturating_sub(1)];
+                    let width = if params.is_empty() {
+                        Some(1)
+                    } else if params.iter().all(u8::is_ascii_digit) {
+                        std::str::from_utf8(params)
+                            .ok()
+                            .and_then(|value| value.parse::<usize>().ok())
+                            .map(|value| value.max(1))
+                    } else {
+                        None
+                    };
+                    if let Some(width) = width.filter(|width| *width <= 512) {
+                        for _ in 0..width {
+                            out.push(' ');
+                        }
+                    }
+                }
+            }
+            b']' => {
+                index += 1;
+                while index < bytes.len() {
+                    if bytes[index] == 0x07 {
+                        index += 1;
+                        break;
+                    }
+                    if bytes[index] == 0x1b && bytes.get(index + 1) == Some(&b'\\') {
+                        index += 2;
+                        break;
+                    }
+                    index += 1;
+                }
+            }
+            _ => index += 1,
+        }
+    }
+    out
+}
+
+
+/// Locate every maintained composer and classify the LOWEST one on screen.
+///
+/// A pane has exactly one live composer and it sits at the bottom of the viewport, so anything
+/// composer-shaped above it is scrollback — typically a pasted or logged screen from the other
+/// harness. Preferring one harness unconditionally lets that transcript decide, which is a wrong
+/// *positive*: the paste and the Return go to the pane's real composer whichever text was read, so
+/// it can type into, or submit, a human's live draft. Scrollback is above the live composer by
+/// construction, so picking the lowest needs no per-pair special case.
+pub(super) fn classify_composer(screen: &str, expected: &str) -> ComposerState {
+    let plain = strip_ansi(screen);
+    let screen = Screen { raw: screen, plain: &plain };
+    harness::all()
+        .into_iter()
+        .filter_map(|harness| harness.locate(&screen).map(|located| (located.row, harness)))
+        .max_by_key(|(row, _)| *row)
+        .map(|(_, harness)| harness.classify(&screen, expected))
+        // No maintained composer is locatable, so nothing is proven either way.
+        .unwrap_or(ComposerState::Ambiguous)
+}