Feed both writers' decision bodies through tiles, retiring WriterBuffer

.github/scripts/shared-state/test_check_shared_state.py

@@ -0,0 +1,267 @@
+#!/usr/bin/env python3
+"""Unit tests for the shared-state checker.
+
+Run first in CI so a checker bug cannot mask a real back channel — or, worse,
+manufacture a failure that sends someone editing correct code.
+"""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from check_shared_state import (  # noqa: E402
+    _code_view,
+    _inner_type,
+    _justified,
+    _normalize,
+    check_file,
+)
+
+FAKE = Path(__file__).resolve().parents[3] / "src" / "interpreter" / "fake.rs"
+
+
+def findings(src: str) -> list[str]:
+    return [f.what for f in check_file(FAKE, src)[0]]
+
+
+def exceptions(src: str) -> list[str]:
+    return [f.what for f in check_file(FAKE, src)[1]]
+
+
+class InnerType(unittest.TestCase):
+    def test_takes_balanced_brackets_not_the_first_close(self) -> None:
+        # The bug this replaced: a non-greedy regex stopped at the first `>>`,
+        # yielding `Option<Box<dyn TileOperator` — which matches no allowlist
+        # entry, so a legitimate cycle slot read as a violation.
+        text = "type S = Rc<RefCell<Option<Box<dyn TileOperator>>>>;"
+        start = text.index("Option")
+        self.assertEqual(_inner_type(text, start), "Option<Box<dyn TileOperator>>")
+
+    def test_unterminated_type_is_none(self) -> None:
+        text = "x: Rc<RefCell<HashMap<String,"
+        self.assertIsNone(_inner_type(text, text.index("HashMap")))
+
+
+class Normalize(unittest.TestCase):
+    def test_keeps_the_space_after_dyn(self) -> None:
+        self.assertEqual(_normalize("Box< dyn Consumer >"), "Box<dyn Consumer>")
+
+    def test_collapses_punctuation_spacing(self) -> None:
+        self.assertEqual(
+            _normalize("HashMap<(String,String) , RouteSender>"),
+            "HashMap<(String, String), RouteSender>",
+        )
+
+
+class Justification(unittest.TestCase):
+    def test_same_line(self) -> None:
+        self.assertEqual(findings("x: Rc<RefCell<Rows>>, // shared-state-ok: why"), [])
+
+    def test_preceding_line(self) -> None:
+        self.assertEqual(findings("// shared-state-ok: why\nx: Rc<RefCell<Rows>>,"), [])
+
+    def test_scans_up_past_doc_comments_and_attributes(self) -> None:
+        src = "// shared-state-ok: why\n/// docs\n#[allow(dead_code)]\nx: Rc<RefCell<Rows>>,"
+        self.assertEqual(findings(src), [])
+
+    def test_does_not_leak_across_a_code_line(self) -> None:
+        # An annotation two fields up must not silence an unrelated one below.
+        src = "// shared-state-ok: why\na: Rc<RefCell<Rows>>,\nb: Rc<RefCell<Other>>,"
+        self.assertEqual(findings(src), ["shared cell of `Other`"])
+
+    def test_bare_marker_without_a_reason_does_not_count(self) -> None:
+        self.assertEqual(
+            findings("// shared-state-ok:\nx: Rc<RefCell<Rows>>,"),
+            ["shared cell of `Rows`"],
+        )
+
+
+class Detection(unittest.TestCase):
+    def test_flags_the_retired_writer_buffer(self) -> None:
+        # The violation this checker exists for.
+        self.assertEqual(
+            findings("pub type BodyInputBuffer = Rc<RefCell<WriterBuffer>>;"),
+            ["shared cell of `WriterBuffer`"],
+        )
+
+    def test_allows_known_kinds(self) -> None:
+        src = (
+            "pub type SharedConsumer = Rc<RefCell<dyn Consumer>>;\n"
+            "shared: Rc<RefCell<FanOutShared>>,\n"
+            "source: Rc<RefCell<dyn DataSourceDomainExtentImpl>>,\n"
+        )
+        self.assertEqual(findings(src), [])
+
+    def test_flags_a_hand_rolled_cycle_slot(self) -> None:
+        # `CycleSlot` is the one definition of this cell (justified at its own
+        # declaration), so a second copy has to argue for itself rather than
+        # inherit the first one's reasoning.
+        self.assertEqual(
+            findings("type WriterSlot = Rc<RefCell<Option<Box<dyn TileOperator>>>>;"),
+            ["shared cell of `Option<Box<dyn TileOperator>>`"],
+        )
+
+    def test_flags_other_cell_kinds(self) -> None:
+        src = "a: Arc<Mutex<Rows>>,\nb: Rc<Cell<usize>>,\nc: Arc<RwLock<Rows>>,"
+        self.assertEqual(
+            findings(src),
+            [
+                "shared cell of `Rows`",
+                "shared cell of `usize`",
+                "shared cell of `Rows`",
+            ],
+        )
+
+    def test_flags_ambient_mutable_state(self) -> None:
+        self.assertEqual(
+            findings("static mut ROWS: usize = 0;"), ["ambient mutable state"]
+        )
+        self.assertEqual(
+            findings("thread_local! { static ROWS: usize = 0; }"),
+            ["ambient mutable state"],
+        )
+
+    def test_flags_a_static_holding_an_interior_mutable_cell(self) -> None:
+        # A `static` needs no `mut` to be ambient: the cell supplies the
+        # mutability and the name supplies the reach.
+        self.assertEqual(
+            findings("static ROWS: Mutex<Vec<Row>> = Mutex::new(Vec::new());"),
+            ["ambient mutable state"],
+        )
+        self.assertEqual(
+            findings("static ROWS: OnceLock<Mutex<Rows>> = OnceLock::new();"),
+            ["ambient mutable state"],
+        )
+
+    def test_flags_a_bare_cell_field(self) -> None:
+        # The owner supplies the sharing: `Arc<State>` over a `State` holding a
+        # `Mutex` is `Arc<Mutex<…>>` with the layers swapped.
+        self.assertEqual(findings("    pending: Mutex<Rows>,"), ["cell of `Rows`"])
+        self.assertEqual(findings("    pub used: RefCell<bool>,"), ["cell of `bool`"])
+
+    def test_a_bare_cell_takes_a_justification_like_any_other(self) -> None:
+        self.assertEqual(findings("// shared-state-ok: why\n    pending: Mutex<Rows>,"), [])
+
+    def test_a_local_binding_is_not_a_field(self) -> None:
+        self.assertEqual(findings("    let seen: RefCell<Rows> = RefCell::new(rows);"), [])
+
+    def test_ignores_commented_out_code(self) -> None:
+        self.assertEqual(findings("// x: Rc<RefCell<Rows>>,"), [])
+
+    def test_reads_through_a_type_wrapped_across_lines(self) -> None:
+        self.assertEqual(
+            findings("x: Rc<RefCell<HashMap<String,\n    Rows>>>,"),
+            ["shared cell of `HashMap<String, Rows>`"],
+        )
+
+    def test_flags_a_cell_whose_wrapper_rustfmt_split(self) -> None:
+        # The gap this closes: matching per line, `Rc<` and `RefCell<` land on
+        # different lines and the site is not seen *at all* — worse than the
+        # `<unterminated type>` report below, which at least fails the gate.
+        self.assertEqual(
+            findings("x: Rc<\n    RefCell<Rows>,\n>,"),
+            ["shared cell of `Rows`"],
+        )
+
+    def test_reports_an_unbalanced_type_rather_than_guessing(self) -> None:
+        self.assertEqual(
+            findings("x: Rc<RefCell<HashMap<String,"),
+            ["shared cell of `<unterminated type>`"],
+        )
+
+
+
+class Exceptions(unittest.TestCase):
+    def test_a_justified_site_is_listed_not_dropped(self) -> None:
+        # The suppression and the listing are the same fact seen twice: an excused
+        # site produces no finding, and is exactly what `EXPECTED_EXCEPTIONS`
+        # enumerates.
+        src = "// shared-state-ok: why\nx: Rc<RefCell<Rows>>,"
+        self.assertEqual(findings(src), [])
+        self.assertEqual(exceptions(src), ["shared cell of `Rows`"])
+
+    def test_an_allowlisted_kind_is_not_an_exception(self) -> None:
+        # A known-legitimate kind is not a hole in the invariant, so it does not
+        # need an entry.
+        src = "shared: Rc<RefCell<FanOutShared>>,"
+        self.assertEqual(findings(src), [])
+        self.assertEqual(exceptions(src), [])
+
+
+class TestCode(unittest.TestCase):
+    def test_skips_a_cfg_test_module(self) -> None:
+        src = (
+            "a: Rc<RefCell<Rows>>,\n"
+            "#[cfg(test)]\n"
+            "mod tests {\n"
+            "    struct Spy {\n"
+            "        log: Rc<RefCell<Vec<Guard>>>,\n"
+            "    }\n"
+            "}\n"
+        )
+        self.assertEqual(findings(src), ["shared cell of `Rows`"])
+
+    def test_skips_a_cfg_test_fn_without_swallowing_what_follows(self) -> None:
+        # `#[cfg(test)]` is not always the trailing `mod tests` — it also gates a
+        # single function mid-file, and the code after it is still production.
+        src = (
+            "impl E {\n"
+            "    #[cfg(test)]\n"
+            "    fn for_test() -> Self {\n"
+            "        let _: Rc<RefCell<Vec<Guard>>>;\n"
+            "    }\n"
+            "}\n"
+            "b: Rc<RefCell<Rows>>,\n"
+        )
+        self.assertEqual(findings(src), ["shared cell of `Rows`"])
+
+    def test_skips_a_cfg_test_use(self) -> None:
+        src = "#[cfg(test)]\nuse foo::Rc;\nb: Rc<RefCell<Rows>>,\n"
+        self.assertEqual(findings(src), ["shared cell of `Rows`"])
+
+    def test_does_not_skip_a_test_helpers_feature_gate(self) -> None:
+        # That configuration compiles into a real library build, so it is
+        # production code that tests also use.
+        src = '#[cfg(any(test, feature = "test-helpers"))]\npub fn h() {\n    let _: Rc<RefCell<Rows>>;\n}\n'
+        self.assertEqual(findings(src), ["shared cell of `Rows`"])
+
+    def test_does_not_skip_cfg_not_test(self) -> None:
+        src = "#[cfg(not(test))]\nb: Rc<RefCell<Rows>>,\n"
+        self.assertEqual(findings(src), ["shared cell of `Rows`"])
+
+
+class CodeView(unittest.TestCase):
+    def test_blanks_a_brace_inside_a_string(self) -> None:
+        # The brace matching that skips a `#[cfg(test)]` item would otherwise
+        # desync on it and blank an arbitrary amount of the file.
+        src = (
+            "#[cfg(test)]\n"
+            'fn t() { let s = "}"; let _: Rc<RefCell<Spy>>; }\n'
+            "b: Rc<RefCell<Rows>>,\n"
+        )
+        self.assertEqual(findings(src), ["shared cell of `Rows`"])
+
+    def test_blanks_a_raw_string(self) -> None:
+        self.assertEqual(findings('let s = r#"x: Rc<RefCell<Rows>>,"#;'), [])
+
+    def test_blanks_a_block_comment(self) -> None:
+        self.assertEqual(findings("/* x: Rc<RefCell<Rows>>, */"), [])
+
+    def test_a_lifetime_is_not_a_char_literal(self) -> None:
+        # Treating `'a` as an unterminated char literal would blank forward to
+        # the next quote and hide everything between.
+        src = "struct S<'a> { r: &'a u8 }\nb: Rc<RefCell<Rows>>,\n"
+        self.assertEqual(findings(src), ["shared cell of `Rows`"])
+
+    def test_preserves_line_numbers(self) -> None:
+        src = '// a comment\n/* block */\nx: Rc<RefCell<Rows>>,'
+        self.assertEqual(_code_view(src).count("\n"), src.count("\n"))
+        self.assertEqual(check_file(FAKE, src)[0][0].line, 3)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=0)

.github/workflows/ci.yml

@@ -51,6 +51,13 @@ jobs:
       - name: Doc references
         run: ./ci.sh doc_refs
 
+      # Gate the interpreter's no-back-channel invariant (operators exchange
+      # tiles through `get`, never through shared mutable state). Python-only
+      # like the doc-ref check, so it runs here rather than behind the toolchain
+      # setup, and ahead of the paths filter below so it cannot be skipped.
+      - name: Shared state
+        run: ./ci.sh shared_state
+
       # 1. Detect if changes are only in ignored paths
       # Skip CI for changes that don't affect code
       - uses: dorny/paths-filter@de90cc6fb38fc0963ad72b210f1f284cd68cea36

ci.sh

@@ -69,6 +69,15 @@ ci_doc_refs() {
   python3 .github/scripts/doc-refs/check_doc_refs.py || return 1
 }
 
+# Gate the interpreter's no-back-channel invariant: operators exchange tiles
+# through `get`, never through shared mutable state (`src/interpreter/CLAUDE.md`).
+# Same ordering rule as `ci_doc_refs`: checker tests first, and `|| return 1` per
+# command because `ci_all` disables errexit for this function's extent.
+ci_shared_state() {
+  python3 .github/scripts/shared-state/test_check_shared_state.py || return 1
+  python3 .github/scripts/shared-state/check_shared_state.py || return 1
+}
+
 # Fast inner-loop gate for local iteration: format, lint (debug, lib+bins only),
 # and test. Deliberately skips the phases whose cost is compile-bound and rarely
 # relevant mid-iteration — the *release* clippy pass (~2x the debug one; only
@@ -101,6 +110,9 @@ ci_all() {
   ci_doc_refs || failed=1
   # shellcheck disable=SC2310
   # intentional: || captures failure without exiting
+  ci_shared_state || failed=1
+  # shellcheck disable=SC2310
+  # intentional: || captures failure without exiting
   ci_fmt || failed=1
   # shellcheck disable=SC2310
   # intentional: || captures failure without exiting

src/ccl/design/mutability.md

@@ -789,16 +789,19 @@ causal matcher (`letrec::check_letrec_causal`).
 | Letrec pattern | Engine |
 |---|---|
 | a binding referenced only via `get_prev_seq(𝑏, …)`, over a finite/stream induction domain | `InductionStore` — the position-driven changelog loop engine (read densely via `StoreDenseRead`) |
-| commit-record bindings + `begin_<site>` oracles + `Txn` histories read via `get_prev_txn` | the commit operator (`CommitOperator`, `TransactWriter`, cyclic `FanOut`, `StoreValueStream`) |
+| commit-record bindings + `begin_<site>` oracles + `Txn` histories read via `get_prev_txn` | the commit operator (`CommitOperator`, `TransactDriver`, `TransactWriter`, cyclic `FanOut`, `StoreValueStream`) |
 | a `Txn` history read out of a read-only block (any reading loop — a live request stream, a finite loop, or a standalone singleton) | the as-of read (`AsOf`), latching the store's value as of the reading transaction's position, indexed by the outer reading loop |
 | a non-causal cycle, or a causal shape loop planning does not know | compile error (no silent fallback) |
 
 ### The runtime engines
 
-- **`InductionStore` (+ `StoreDenseRead`)** — the induction loop. It drives the loop *sequentially
-  inside the producer* — folding each position's prev-accumulator from its own commit engine, so
-  there is no cyclic `FanOut` — and writes a `Tile::Store` changelog (`init` at position 0; a
-  `commit: false` position carries the prior value forward). `StoreDenseRead` then folds that
+- **`InductionDriver` + `InductionStore` (+ `StoreDenseRead`)** — the induction loop, as a cycle
+  through a `FanOut::new_cyclic`. The store consumes the body's decisions and writes a
+  `Tile::Store` changelog (`init` at position 0; a `commit: false` position carries the prior
+  value forward); the driver reads that changelog back to produce the body's `(prev…, item)`
+  input, taking the next position from the decided frontier and the prev-accumulator from the
+  value at it. The accumulator therefore crosses between them as a tile, like every other
+  operator-to-operator value, at one position per pull. `StoreDenseRead` then folds the
   changelog over the loop domain to the dense `𝐷 ⇀ 𝑉` stream (serving both a scalar-final
   `ExtractFinal` and a co-iterated `fan_in`). A single always-commit or commit-gated writer over a
   finite *or* async domain.
@@ -1090,7 +1093,7 @@ Both a **finite** loop and an **async** (streaming) source drive an induction ac
 model treats a finite domain as a stream that terminates (§Liveness) — and every induction accumulator
 uses *one* realization: the changelog `InductionStore`. Plain, conditional, and feed-carrying loops
 over finite or async extents all route through it. The
-drive reads its source by absolute domain position (async domains arrive unordered), reclaims the
+driver reads its source by absolute domain position (async domains arrive unordered), reclaims the
 consumed prefix as it advances, and carries reply feeds as `__fire`-gated taps — see *Induction
 stores as a changelog* in `../../interpreter/design-operators.md`.
 

src/interpreter/CLAUDE.md

@@ -24,12 +24,15 @@ Concretely:
   reads the prev-acc directly from a `Rc<RefCell<...>>` instead of
   through `get`") is a violation, even if it works on the test cases at
   hand.  Cyclic graphs go through `FanOut::new_cyclic` and the
-  re-entrancy machinery on `FanOutShared` — see
-  [`Recurse`]'s docs for the contract — but the data on the wire is
-  still a [`Tile`].
-- Constructor-time wiring (e.g., `Recurse::recursive_input_setter`)
-  passes `TileOperator` handles, not raw values.  The operator graph is
-  static; values flow through it at `get` time.
+  re-entrancy machinery on `FanOutShared` — see [`FanOutReentrancy`]'s
+  docs for the contract — but the data on the wire is still a [`Tile`].
+  A cyclic pull is served the fan's cached snapshot rather than
+  re-entering the inner producer, which is why such a cycle advances one
+  step per outer pull.
+- Constructor-time wiring (a [`CycleSlot`], filled through its
+  `setter` once the rest of the cycle exists) passes `TileOperator`
+  handles, not raw values.  The operator graph is static; values flow
+  through it at `get` time.
 - Side effects (I/O, sinks, notifications) live at the boundary —
   `compile_program` wires a `SinkConsumer` to the final operator, and
   `Scheduler::check_for_notifications` drives them.  Operators