Serialize the audit state owner's writes so concurrent invocations cannot corrupt the record

.changeset/issue-1040-serialize-audit-state-writes.md

@@ -0,0 +1,7 @@
+---
+bump: patch
+type: Fixed
+---
+
+- **Serialize the create-issue audit state owner's writes so concurrent invocations cannot corrupt the record.** `scripts/issue-audit-state.py` now wraps every mutating subcommand in an exclusive-create sentinel critical section (a `.lock` file beside the state document, created with `os.O_CREAT | os.O_EXCL`), so two concurrent invocations for the same slug produce a document reflecting one of them entirely and then the other, never a mixture. `save_state` obtains a unique per-writer temporary path from `tempfile.mkstemp` (retaining the `.json.tmp` suffix) and retries `os.replace` over `PermissionError`, so two writers never share and truncate one temporary file. Read-only subcommands (`query-*`, `emit-body`, `check-claim-staleness`) acquire no sentinel and stay unserialized; an abandoned sentinel is recovered by age. A new fail-closed transitive call-graph check in `lib/test/check-audit-lifecycle-contracts.py` proves `save_state` is unreachable from every read-only-classified subcommand. The mechanism is standard-library only (no `fcntl`/`msvcrt`), adds no state-document field, and produces no new mutation-exit class — every section failure is a `could not persist state to …` condition the shipped routing already carries. The decision channel (`next_call=` / `query-*`) remains unserialized, and its non-authoritative-under-concurrency residual is stated in `docs/DEVFLOW_SYSTEM_OVERVIEW.md` §11. (#1045)
+- **#1004 migration note.** This change adds two test-only environment variables in the frozen `DEVFLOW_` namespace — `DEVFLOW_IAS_ACQUIRE_WINDOW_S` and `DEVFLOW_IAS_STALE_AFTER_S` — which the Tier-3 env-var rename (#1004) must migrate alongside the existing `DEVFLOW_*` members. (#1045)

lib/test/check-audit-lifecycle-contracts.py

@@ -43,6 +43,8 @@
 from __future__ import annotations
 
 import argparse
+import ast
+import builtins
 import contextlib
 import importlib.util
 import inspect
@@ -174,6 +176,213 @@ def _subparser_of(parser, name):
     return None
 
 
+# The builtins a call may name and still be provably unable to reach a module-level
+# save_state — every builtin function, type, and exception (`len`, `print`, `sorted`,
+# `SystemExit`, `ValueError`, …). `getattr` is safe in its BARE-name value form
+# `getattr(x, y)`; the dangerous form `getattr(x, y)()` is a Call-as-callee, caught by the
+# indirect-dispatch arm below regardless.
+_SAFE_BUILTIN_CALLS = frozenset(dir(builtins))
+
+
+def _module_level_names(tree):
+    """(functions_by_name, safe_leaf_names) for the module. `functions_by_name` are the
+    function defs the walk FOLLOWS — every module-level def PLUS every nested/closure helper
+    they define, indexed by name, so a bare-name call to a helper the source statically
+    carries is followed (proving whether it reaches save_state) rather than flagged
+    unresolvable. `safe_leaf_names` are names a call may reference and still reach no
+    followable function by name — imported names and module-level class names (a class
+    instantiation cannot BE the `save_state` function, e.g. `StateError(...)`). A name collision across scopes
+    over-approximates by keeping one body; that only ever follows MORE, which is the
+    fail-closed direction for an unreachability proof.
+    """
+    functions = {}
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            functions[node.name] = node
+    safe = set(_SAFE_BUILTIN_CALLS)
+    for node in tree.body:
+        if isinstance(node, ast.ClassDef):
+            safe.add(node.name)
+        elif isinstance(node, ast.Import):
+            for alias in node.names:
+                safe.add((alias.asname or alias.name).split(".")[0])
+        elif isinstance(node, ast.ImportFrom):
+            for alias in node.names:
+                safe.add(alias.asname or alias.name)
+    return functions, safe
+
+
+def check_readonly_complement(module, registered, report):
+    """Every read-only-classified subcommand's handler must be UNABLE to reach `save_state`
+    through the transitive call graph of module-level functions — not merely have it absent
+    from the handler's own source text (issue #1040). A source-text-only check would pass a
+    handler that reaches save_state one hop away through a helper, the fail-open shape this
+    exists to prevent.
+
+    The walk's closed set is module-level functions; its complement is named and handled,
+    never left silent. A call target the walk cannot resolve to a module-level function — a
+    bare-name alias/closure/nested helper, or an indirect dispatch through a table or
+    getattr — makes the check FAIL CLOSED for that subcommand, reporting the unresolvable
+    call site rather than reporting clean. An attribute/method call reaches no module-level
+    function by name (module-level functions are called as bare names), so it cannot be
+    `save_state` and is safe. A read-only-classified handler is proved safe only when every
+    call on its transitive path resolved.
+    """
+    # Analyze the PASSED module's own source (not IAS directly), so a fixture module drives
+    # this check exactly like check_emitting_complement — the positive controls in the test
+    # suite load a crafted module and expect a Refusal.
+    try:
+        source = inspect.getsource(module)
+    except (OSError, TypeError) as exc:
+        raise Refusal(f"readonly-complement: could not read the module source of "
+                      f"{getattr(module, '__name__', module)!r}: {exc}") from exc
+    try:
+        tree = ast.parse(source)
+    except SyntaxError as exc:
+        raise Refusal(f"readonly-complement: could not parse the module source: "
+                      f"{exc}") from exc
+    functions, safe = _module_level_names(tree)
+
+    parser = module.build_parser()
+    name_to_handler = {}
+    for action in parser._actions:  # noqa: SLF001
+        if isinstance(action, argparse._SubParsersAction):  # noqa: SLF001
+            for name, sub in action.choices.items():
+                fn = sub._defaults.get("func")  # noqa: SLF001
+                name_to_handler[name] = getattr(fn, "__name__", None)
+
+    predicate = module._is_read_only  # noqa: SLF001
+    readonly = sorted(n for n in registered if predicate(n))
+    if not readonly:
+        raise Refusal("readonly-complement: the read-only predicate selected NO registered "
+                      "subcommand, so the critical section would wrap every command and this "
+                      "complement check would be vacuous")
+
+    # Module-level constants (name -> value node), so a read-only handler that dispatches
+    # over a module-level TABLE of producers (query-boundary's `for _, produce in
+    # _BOUNDARY_PRODUCERS: produce(...)`) is RESOLVED through the constant rather than
+    # failed closed: name-reachability propagates through the constant's value into the
+    # producer lambdas and the module functions they name.
+    constants = {}
+    for node in tree.body:
+        if isinstance(node, ast.Assign):
+            for tgt in node.targets:
+                if isinstance(tgt, ast.Name):
+                    constants[tgt.id] = node.value
+
+    def _targets_name(target, name):
+        """True iff an assignment/for target binds `name` (a Name, or a Tuple/List of
+        them)."""
+        if isinstance(target, ast.Name):
+            return target.id == name
+        if isinstance(target, (ast.Tuple, ast.List)):
+            return any(_targets_name(el, name) for el in target.elts)
+        return False
+
+    def _local_call_resolvable(node, name):
+        """A bare-name call to a LOCAL `name` is resolvable only when `name` is bound within
+        `node` from a source that references a module-level function or constant — then the
+        call target is reached through name-propagation over that source (query-boundary's
+        `for _, produce in _BOUNDARY_PRODUCERS` binds `produce` from a module constant). A
+        name bound only from a getattr()/subscript/opaque source, or not bound at all (a
+        parameter or callback), is unresolvable and fails the subcommand closed.
+        """
+        for sub in ast.walk(node):
+            srcs = []
+            if isinstance(sub, ast.For) and _targets_name(sub.target, name):
+                srcs.append(sub.iter)
+            elif isinstance(sub, ast.Assign) and any(
+                    _targets_name(t, name) for t in sub.targets):
+                srcs.append(sub.value)
+            elif isinstance(sub, ast.comprehension) and _targets_name(sub.target, name):
+                srcs.append(sub.iter)
+            for src in srcs:
+                for ref in ast.walk(src):
+                    if isinstance(ref, ast.Name) and (
+                            ref.id in functions or ref.id in constants):
+                        return True
+        return False
+
+    def _analyze_node(node):
+        """(module-level names referenced in Load context, has-computed-callee,
+        [unresolvable local-call names]) in node's whole subtree — nested defs and lambdas
+        included, so a helper or a producer lambda is not a blind spot. A computed callee
+        (getattr(...)() / table[key]()) and a bare-name call to an unresolvable local
+        (a getattr-aliased callable) are the two shapes the name walk cannot resolve.
+        """
+        names = set()
+        has_computed = False
+        unresolvable = []
+        for sub in ast.walk(node):
+            if isinstance(sub, ast.Name) and isinstance(sub.ctx, ast.Load):
+                if sub.id in functions or sub.id in constants:
+                    names.add(sub.id)
+            elif isinstance(sub, ast.Call):
+                fn = sub.func
+                if isinstance(fn, ast.Name):
+                    nm = fn.id
+                    if (nm != "save_state" and nm not in functions
+                            and nm not in constants and nm not in safe
+                            and not _local_call_resolvable(node, nm)):
+                        unresolvable.append(nm)
+                elif not isinstance(fn, ast.Attribute):
+                    has_computed = True
+        return names, has_computed, unresolvable
+
+    def analyze(subcmd, handler):
+        # Transitive name-reachability over module-level functions AND constants. A read-only
+        # handler is proved safe only when `save_state` is unreachable by name AND no
+        # reachable function contains a computed (unresolvable) callee. This is sound by
+        # name — a handler that reaches save_state through any bare-name path (a call, a
+        # callback passed by name, a nested def, or a producer table) names it and is
+        # caught; the one residue the name walk cannot see, a getattr('save_state')()
+        # string dispatch, is caught by the computed-callee arm.
+        seen = set()
+        queue = [handler]
+        while queue:
+            nm = queue.pop()
+            if nm in seen:
+                continue
+            seen.add(nm)
+            node = functions.get(nm)
+            is_func = node is not None
+            if node is None:
+                node = constants.get(nm)
+            if node is None:
+                continue
+            names, has_computed, unresolvable = _analyze_node(node)
+            if is_func and has_computed:
+                raise Refusal(
+                    f"readonly-complement: read-only subcommand {subcmd!r} reaches an "
+                    f"indirect dispatch (a computed callee) in {nm!r}; the walk cannot "
+                    "prove that path does not reach save_state — fail closed")
+            if is_func and unresolvable:
+                raise Refusal(
+                    f"readonly-complement: read-only subcommand {subcmd!r} makes an "
+                    f"unresolvable call {unresolvable[0]!r}() in {nm!r} (a local bound from "
+                    "no module-level function or constant — a getattr alias, a closure, or a "
+                    "callback); a read-only handler is proved safe only when every call "
+                    "resolves")
+            if "save_state" in names:
+                raise Refusal(
+                    f"readonly-complement: read-only subcommand {subcmd!r} reaches "
+                    f"save_state (through {nm}); the read-only predicate has misclassified "
+                    "a mutating subcommand as read-only")
+            queue.extend(names - seen)
+
+    for name in readonly:
+        handler = name_to_handler.get(name)
+        if handler is None or handler not in functions:
+            raise Refusal(
+                f"readonly-complement: read-only subcommand {name!r} maps to handler "
+                f"{handler!r}, which is not a module-level function — the walk cannot begin, "
+                "so the classification is unproven")
+        analyze(name, handler)
+
+    report.append(f"readonly-complement: all {len(readonly)} read-only subcommands proved "
+                  "unable to reach save_state through the module-level call graph")
+
+
 def check_readbacks(module, registered, report):
     """The docstring's read-back enumeration vs. the dispatched `_MULTILINE_READBACKS`."""
     doc = module.__doc__ or ""
@@ -435,6 +644,7 @@ def main():
         module = _load_module()
         registered = module.registered_subcommands()
         check_readbacks(module, registered, report)
+        check_readonly_complement(module, registered, report)
         check_emitting_complement(module, registered, report)
         check_round_defaulted(module, registered, report)
         check_next_action_routing_totality(module, report)

lib/test/modules/coverage-map.json

@@ -678,6 +678,10 @@
       "note": "",
       "owner": "unmodularized"
     },
+    "1040": {
+      "note": "",
+      "owner": "issue-audit-state"
+    },
     "1032": {
       "note": "",
       "owner": "review-trigger-helpers"

lib/test/modules/issue-audit-state.sh

@@ -2196,3 +2196,63 @@ if [ -d "$WP_SB" ]; then
     "0" "$(sed -n 1p "$WP_SB/.wp-embed-rc" 2>/dev/null)"
   rm -rf "$WP_SB"
 fi
+
+# ── issue #1040: write-serialization sentinel at the process boundary ──────────────
+# The shell-level assertions on the section's process boundary — the exit code and stderr
+# breadcrumb the orchestrator routes on — driven with sub-second DEVFLOW_IAS_* overrides so
+# the section's timing behavior is exercised in milliseconds rather than at the shipped
+# 30s/45s bounds. Contention is created deterministically (the test plants the sentinel
+# itself), never by racing two writers.
+SL_SB="$(git_sandbox '#1040 cli_stale_break_exit_and_breadcrumb')"
+(
+  cd "$SL_SB" || exit 1
+  mkdir -p .prflow/tmp
+  # (1) stale break: plant a sentinel, age it past a sub-second stale_after_s, then run a
+  #     real mutation. It must break the stale sentinel, proceed, and exit 0.
+  printf '4242' > .prflow/tmp/issue-audit-state-s.json.lock
+  sleep 0.2
+  DEVFLOW_IAS_STALE_AFTER_S=0.05 DEVFLOW_IAS_ACQUIRE_WINDOW_S=0.5 \
+    python3 "$IAS" init s > .sl-out 2> .sl-err
+  printf '%s' "$?" > .sl-rc
+  # (2) contention refusal: a FRESH sentinel with INVERTED bounds (window < stale) is never
+  #     broken, so acquisition exhausts the window → exit non-zero, no state persisted, and
+  #     the could-not-persist breadcrumb (the routing class the skill already carries).
+  printf '9999' > .prflow/tmp/issue-audit-state-s3.json.lock
+  DEVFLOW_IAS_ACQUIRE_WINDOW_S=0.2 DEVFLOW_IAS_STALE_AFTER_S=30 \
+    python3 "$IAS" init s3 > .sl3-out 2> .sl3-err
+  printf '%s' "$?" > .sl3-rc
+  # the refused mutation left no state file for s3
+  [ -f .prflow/tmp/issue-audit-state-s3.json ] && printf 'yes' > .sl3-state || printf 'no' > .sl3-state
+) || true
+assert_eq "#1040 cli_stale_break_exit_and_breadcrumb: the mutation breaks the stale sentinel and exits 0" \
+  "0" "$(cat "$SL_SB/.sl-rc" 2>/dev/null)"
+assert_eq "#1040 cli_stale_break_exit_and_breadcrumb: stderr carries the stale-break breadcrumb" \
+  "1" "$(grep -c 'broke a stale audit-state sentinel' "$SL_SB/.sl-err" 2>/dev/null)"
+assert_eq "#1040 cli_stale_break_exit_and_breadcrumb: the breadcrumb names the recorded pid" \
+  "1" "$(grep -c '4242' "$SL_SB/.sl-err" 2>/dev/null)"
+assert_eq "#1040 cli_contention_refusal: inverted bounds over a fresh sentinel exit non-zero (1)" \
+  "1" "$(cat "$SL_SB/.sl3-rc" 2>/dev/null)"
+assert_eq "#1040 cli_contention_refusal: the refusal names could-not-persist-state" \
+  "1" "$(grep -c 'could not persist state' "$SL_SB/.sl3-err" 2>/dev/null)"
+assert_eq "#1040 cli_contention_refusal: the refused mutation persisted no state file" \
+  "no" "$(cat "$SL_SB/.sl3-state" 2>/dev/null)"
+rm -rf "$SL_SB"
+
+# readers_are_not_serialized (process boundary): a read-only subcommand acquires no
+# sentinel, so a query-* invoked WHILE a sentinel is held for the same slug still exits 0
+# (the AC: "a query-* invocation issued while a sentinel is held exits 0 with its decided
+# answer line"). Plant a fresh sentinel and confirm query-nonce (read-only) is unaffected.
+RO_SB="$(git_sandbox '#1040 readers_are_not_serialized_while_held')"
+(
+  cd "$RO_SB" || exit 1
+  python3 "$IAS" init s >/dev/null 2>&1
+  mkdir -p .prflow/tmp
+  printf '4242' > .prflow/tmp/issue-audit-state-s.json.lock
+  python3 "$IAS" query-nonce s > .ro-out 2> .ro-err
+  printf '%s' "$?" > .ro-rc
+) || true
+assert_eq "#1040 readers_are_not_serialized: query-nonce exits 0 while a sentinel is held" \
+  "0" "$(cat "$RO_SB/.ro-rc" 2>/dev/null)"
+assert_eq "#1040 readers_are_not_serialized: the held sentinel is left untouched by the reader" \
+  "1" "$( [ -f "$RO_SB/.prflow/tmp/issue-audit-state-s.json.lock" ] && echo 1 || echo 0 )"
+rm -rf "$RO_SB"

lib/test/run.sh

@@ -41471,7 +41471,7 @@ fi
 # and this full-suite call share the same lower-bound contract; test_module_runner.py
 # parses this operand and rejects any coupling drift.
 if ! devflow_run_full_suite_module "$LIB/test/modules/issue-audit-state.sh" \
-  "issue-audit-state" 230; then
+  "issue-audit-state" 238; then
   printf 'ERROR: issue-audit-state boundary could not record its result\n'
   exit 1
 fi