[FIX] Rectify: Evidence-Bound Injected Instruction Policy

.autoskillit/test-filter-manifest.yaml

@@ -245,6 +245,8 @@ scripts/check_pyi_stub_format.py:
   - infra/
 scripts/check_pyi_stub_symbols.py:
   - infra/
+scripts/measure_codex_read_repetition.py:
+  - infra/
 scripts/sync_versions.py:
   - infra/
 

docs/decisions/0005-output-budget-protocol.md

@@ -203,11 +203,13 @@ insertion control, not a measurement of remaining context.
   vs 258,400 in cli 0.144.1, but the effective window is server/catalog-controlled
   and oscillated during July 2026 (openai/codex#31860, #32806) — re-verify, do not
   assume an upgrade restores headroom.
-- openai/codex#25458 / #27830: `fork_turns "none"` task-envelope delivery bug gates
-  the intake digest's sub-agent spawn rule — until the upstream bug is fixed,
-  `codex --json` sessions cannot reliably deliver task-envelope context to
-  sub-agents, so the intake discipline's "do not spawn sub-agents" guard remains
-  the operative constraint.
+- openai/codex#25458 / #27830: `fork_turns "none"` task-envelope delivery bug — until
+  the upstream bug is fixed, `codex --json` sessions cannot reliably deliver
+  task-envelope context to sub-agents, so the intake discipline requires sub-agents to
+  be spawned with `fork_turns "none"` passed explicitly and given an explicit narrow
+  brief; sub-agents return a summary of their own task work, not raw file contents,
+  and must not be delegated the reading or interpretation of the caller's instruction
+  files.
 - openai/codex#33881: agent-TOML `model`/`model_reasoning_effort` reportedly ignored
   on 0.144.5 — affects the pins `_generate_agent_tomls` writes.
 - Upstream auto-compact semantics are scope-dependent
@@ -222,3 +224,10 @@ insertion control, not a measurement of remaining context.
 - Recipe and ordinary result decisions are explicit and independent from history retention.
 - Artifact/invariant failures become explicit bounded errors instead of context floods.
 - The accepted gaps above remain visible work rather than implicit guarantees.
+- The Codex context-intake policy (#4351) is advisory prose with no runtime gate, so it
+  is governed by the evidence-bound `CODEX_INTAKE_RULES` registry rather than by
+  `INVARIANT_REGISTRY`, which maps prohibitions to runtime enforcement targets.
+- The intake digest is deliberately excluded from `PROBE_POLICY_IDENTITY`. Wiring it in
+  would make every intake-prose edit probe-gated under this ADR's Forward Obligations,
+  a real recurring operational cost; excluding it keeps that a conscious future decision
+  rather than a side effect of this change.

scripts/measure_codex_read_repetition.py

@@ -0,0 +1,289 @@
+#!/usr/bin/env python3
+"""Measure the Codex intake-discipline repeat-read rate across rollout sessions (#4351).
+
+Three repeat-read definitions have been published for this signal; this tool
+implements the REPORT method, and states here how the other two differ:
+
+  - report method (implemented here): counts, per session, only bounded-read
+    command shapes whose *leading* verb (before any pipe) is `sed -n`,
+    `head`/`tail` applied directly to a path, `nl -ba | sed -n`, or `rg -n`.
+    A trailing `| head -c N` output-safety wrapper is NOT itself a
+    bounded-read signal — nearly every command in this harness carries one,
+    so counting it naively would classify the vast majority of all commands
+    as "reads" and destroy the signal. A repeat is the 2nd+ bounded read of
+    the same resolved path within one session (rollout file).
+  - leading-command method (2026-07-24, 60-rollout sample): the same
+    leading-shape classifier described above, applied to a smaller, earlier
+    sample. This tool reproduces that method exactly at full corpus size, so
+    its output supersedes that one-off run.
+  - #4351 issue-body method (22,282 exec calls, 07-22 -> 07-24): counted
+    every exec_command call as a "read" with no bounded-shape filter and no
+    repeat-of-same-file grouping. It is not reproduced here because it does
+    not isolate repeat reads of a file already resident in context, which is
+    the friction signal this tool exists to measure.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+DEFAULT_LOG_ROOT = Path("~/.local/share/autoskillit/logs/codex-sessions").expanduser()
+DEFAULT_SESSIONS_INDEX = Path("~/.local/share/autoskillit/logs/sessions.jsonl").expanduser()
+DEFAULT_OUT = Path(".autoskillit/temp/codex_read_repetition_report.json")
+
+# Leading-command bounded-read shapes. Anchored at ^ so a trailing
+# `| head -c N` safety wrapper elsewhere in the command does not qualify it.
+_LEADING_SED_N_PAT = re.compile(r"^\s*(?:\{\s*)?sed\s+-n\s+'[^']*'\s+\S+")
+_LEADING_HEAD_PAT = re.compile(r"^\s*(?:\{\s*)?head\s+-[cn]\s*\d+\s+\S+")
+_LEADING_TAIL_PAT = re.compile(r"^\s*(?:\{\s*)?tail\s+-[cn]\s*\d+\s+\S+")
+_LEADING_NL_BA_PAT = re.compile(r"^\s*(?:\{\s*)?nl\s+-ba\s+\S+\s*\|\s*sed\s+-n\b")
+_LEADING_RG_N_PAT = re.compile(r"^\s*(?:\{\s*)?rg\s+(?:-\S+\s+)*(-n\b|--line-number\b)")
+
+_BOUNDED_PATTERNS = (
+    _LEADING_SED_N_PAT,
+    _LEADING_HEAD_PAT,
+    _LEADING_TAIL_PAT,
+    _LEADING_NL_BA_PAT,
+    _LEADING_RG_N_PAT,
+)
+
+_PATH_TOKEN_PAT = re.compile(r"(?:[./~][\w./\-]+|[\w\-]+/[\w./\-]+)")
+
+# A quoted span (single- or double-) is consumed whole, so a literal `|` inside an
+# rg search pattern -- POSIX ERE alternation, this repo's documented idiom for
+# `pattern` arguments -- does not truncate the match. Only an unquoted `|`/`;`
+# (a real shell pipe/separator) ends the span. Unquantified: callers apply their
+# own `*`/`*?` to match the surrounding regex's greediness.
+_QUOTED_OR_UNPIPED = r"(?:'[^']*'|\"[^\"]*\"|[^|;])"
+
+# The rarer custom_tool_call/exec shape embeds the command in a JS-template
+# string: tools.exec_command({cmd:"..."}).
+_CUSTOM_TOOL_CALL_CMD_PAT = re.compile(
+    r"tools\.exec_command\(\{\s*cmd\s*:\s*[\"']((?:[^\"'\\]|\\.)*)[\"']"
+)
+
+_VERSION_V2_MARKER = "Context Intake Discipline v2:"
+_VERSION_V1_MARKER = "Context Intake Discipline v1:"
+
+# The YYYY/MM directory pair is month precision only; day precision lives in the
+# filename itself, e.g. rollout-2026-05-26T07-30-33-<thread-id>.jsonl.
+_ROLLOUT_DATE_PAT = re.compile(r"rollout-(\d{4}-\d{2}-\d{2})T")
+
+
+def is_bounded_read(cmd: str) -> bool:
+    """Return True when cmd's *leading* command (before any pipe) is a bounded file read."""
+    return any(pattern.search(cmd) for pattern in _BOUNDED_PATTERNS)
+
+
+def extract_target_path(cmd: str) -> str | None:
+    """Best-effort extraction of the file path a bounded-read command targets."""
+    m = re.search(r"sed\s+-n\s+'[^']*'\s+([^\s|;]+)", cmd)
+    if m:
+        return m.group(1)
+    m = re.search(
+        rf"\brg\s+{_QUOTED_OR_UNPIPED}*?(-n\b|--line-number\b){_QUOTED_OR_UNPIPED}*", cmd
+    )
+    if m:
+        # Strip a trailing `2>&1` (or `2>/dev/null`, `>&2`, ...) redirect token
+        # before tokenizing -- an unstripped redirect was mis-attributed as a
+        # path in 8.5% of matches during calibration.
+        segment = re.sub(r"\d*[<>]&?\d*(/\S+)?\s*$", "", m.group(0)).strip()
+        tokens = [t for t in segment.split() if not t.startswith("-")]
+        candidates = [t for t in tokens if t != "rg" and not re.match(r"^\d*[<>]", t)]
+        if candidates:
+            return candidates[-1].strip("'\"")
+    m = re.search(r"\b(?:head|tail)\s+-[cn]\s*\d+\s+([^\s|;]+)", cmd)
+    if m:
+        return m.group(1)
+    m = re.search(r"\bnl\s+-ba\s+([^\s|;]+)", cmd)
+    if m:
+        return m.group(1)
+    tokens = [t for t in _PATH_TOKEN_PAT.findall(cmd) if len(t) > 3]
+    return max(tokens, key=len) if tokens else None
+
+
+def classify_rollout_records(rollout_path: Path) -> tuple[list[tuple[int, str]], int]:
+    """Return (exec_commands, unclassified_count) for one rollout JSONL file.
+
+    Malformed JSONL lines are skipped, never fatal. A record that is
+    exec-shaped (function_call/exec_command or custom_tool_call/exec) but
+    whose command could not be extracted increments the unclassified count
+    instead of silently vanishing.
+    """
+    commands: list[tuple[int, str]] = []
+    unclassified = 0
+    with rollout_path.open("r", encoding="utf-8") as f:
+        for idx, line in enumerate(f):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                rec = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if not isinstance(rec, dict) or rec.get("type") != "response_item":
+                continue
+            payload = rec.get("payload")
+            if not isinstance(payload, dict):
+                continue
+            ptype = payload.get("type")
+            if ptype == "function_call" and payload.get("name") == "exec_command":
+                cmd = _extract_function_call_cmd(payload.get("arguments"))
+                if cmd:
+                    commands.append((idx, cmd))
+                else:
+                    unclassified += 1
+            elif ptype == "custom_tool_call" and payload.get("name") == "exec":
+                raw_input = payload.get("input")
+                match = (
+                    _CUSTOM_TOOL_CALL_CMD_PAT.search(raw_input)
+                    if isinstance(raw_input, str)
+                    else None
+                )
+                if match:
+                    commands.append((idx, match.group(1)))
+                else:
+                    unclassified += 1
+    return commands, unclassified
+
+
+def _extract_function_call_cmd(raw_args: Any) -> str | None:
+    if not raw_args:
+        return None
+    try:
+        parsed = json.loads(raw_args)
+    except (json.JSONDecodeError, TypeError):
+        return None
+    return parsed.get("cmd") if isinstance(parsed, dict) else None
+
+
+def classify_policy_cohort(rollout_path: Path) -> str:
+    """Return 'v2', 'v1', or 'none' from the intake-discipline header seen on the wire.
+
+    An unreadable rollout is treated as 'none' rather than raised -- one bad file
+    in a batch of thousands must not abort the whole measurement run.
+    """
+    try:
+        text = rollout_path.read_text(encoding="utf-8", errors="ignore")
+    except OSError:
+        return "none"
+    if _VERSION_V2_MARKER in text:
+        return "v2"
+    if _VERSION_V1_MARKER in text:
+        return "v1"
+    return "none"
+
+
+def measure_rollout(rollout_path: Path) -> dict[str, Any]:
+    """Measure one rollout's bounded-read and repeat-read counts."""
+    commands, unclassified = classify_rollout_records(rollout_path)
+    seen: dict[str, int] = defaultdict(int)
+    bounded = 0
+    repeats = 0
+    for _idx, cmd in commands:
+        if not is_bounded_read(cmd):
+            continue
+        bounded += 1
+        target = extract_target_path(cmd)
+        if target is None:
+            continue
+        seen[target] += 1
+        if seen[target] > 1:
+            repeats += 1
+    return {
+        "session": rollout_path.name,
+        "cohort": classify_policy_cohort(rollout_path),
+        "exec_commands": len(commands),
+        "bounded_reads": bounded,
+        "repeat_reads": repeats,
+        "unclassified": unclassified,
+        "worst_paths": sorted(seen.items(), key=lambda kv: -kv[1])[:5],
+    }
+
+
+def aggregate_report(rollout_paths: list[Path]) -> dict[str, Any]:
+    """Aggregate per-rollout measurements into a per-cohort report."""
+    cohorts: dict[str, dict[str, Any]] = {}
+    worst_overall: list[tuple[str, str, int]] = []
+    total_unclassified = 0
+    for path in rollout_paths:
+        row = measure_rollout(path)
+        agg = cohorts.setdefault(
+            row["cohort"], {"session_count": 0, "bounded_read_count": 0, "repeat_count": 0}
+        )
+        agg["session_count"] += 1
+        agg["bounded_read_count"] += row["bounded_reads"]
+        agg["repeat_count"] += row["repeat_reads"]
+        total_unclassified += row["unclassified"]
+        for path_name, count in row["worst_paths"]:
+            if count > 1:
+                worst_overall.append((row["session"], path_name, count))
+    for cohort_stats in cohorts.values():
+        bounded = cohort_stats["bounded_read_count"]
+        cohort_stats["repeat_read_rate"] = (
+            cohort_stats["repeat_count"] / bounded if bounded else None
+        )
+    worst_overall.sort(key=lambda row: -row[2])
+    return {
+        "cohorts": cohorts,
+        "worst_offenders": worst_overall[:20],
+        "unclassified_record_count": total_unclassified,
+    }
+
+
+def _date_within_bound(date: str, bound: str, *, is_lower: bool) -> bool:
+    """Compare a day-precision date against a since/until bound of either precision.
+
+    A month-precision bound (`YYYY-MM`) is matched against the date's month only, so
+    every day in that month satisfies it. A day-precision bound (`YYYY-MM-DD`)
+    compares in full, per the inclusive-boundary contract documented on --since/--until.
+    """
+    if len(bound) == len("YYYY-MM"):
+        date = date[: len(bound)]
+    return date >= bound if is_lower else date <= bound
+
+
+def _find_rollouts(log_root: Path, since: str | None, until: str | None) -> list[Path]:
+    paths = sorted(log_root.glob("*/*/rollout-*.jsonl"))
+    if since is None and until is None:
+        return paths
+    filtered = []
+    for p in paths:
+        m = _ROLLOUT_DATE_PAT.search(p.name)
+        date = m.group(1) if m else None
+        if date is not None:
+            if since is not None and not _date_within_bound(date, since, is_lower=True):
+                continue
+            if until is not None and not _date_within_bound(date, until, is_lower=False):
+                continue
+        filtered.append(p)
+    return filtered
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    parser.add_argument("--log-root", type=Path, default=DEFAULT_LOG_ROOT)
+    parser.add_argument("--since", default=None, help="Inclusive date prefix, e.g. 2026-07-18")
+    parser.add_argument("--until", default=None, help="Inclusive date prefix, e.g. 2026-07-24")
+    parser.add_argument("--sessions-index", type=Path, default=DEFAULT_SESSIONS_INDEX)
+    parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
+    args = parser.parse_args(argv)
+
+    rollouts = _find_rollouts(args.log_root, args.since, args.until)
+    report = aggregate_report(rollouts)
+    report["rollouts_scanned"] = len(rollouts)
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(json.dumps(report, indent=2))
+    print(f"CODEX_READ_REPETITION=PASS rollouts={len(rollouts)} out={args.out}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/autoskillit/core/__init__.pyi

@@ -177,9 +177,16 @@ from .types import CODEX_ACTIVE_VIEWS_SUBDIR as CODEX_ACTIVE_VIEWS_SUBDIR
 from .types import CODEX_ARCHIVED_SESSIONS_SUBDIR as CODEX_ARCHIVED_SESSIONS_SUBDIR
 from .types import CODEX_CONTEXT_EXHAUSTION_MARKER as CODEX_CONTEXT_EXHAUSTION_MARKER
 from .types import CODEX_COOK_RESERVED_ENV_VARS as CODEX_COOK_RESERVED_ENV_VARS
+from .types import (
+    CODEX_DISCIPLINE_SUFFIX_BYTE_BUDGET as CODEX_DISCIPLINE_SUFFIX_BYTE_BUDGET,
+)
 from .types import CODEX_EFFORT_MAPPING as CODEX_EFFORT_MAPPING
+from .types import (
+    CODEX_INTAKE_DISCIPLINE_BYTE_BUDGET as CODEX_INTAKE_DISCIPLINE_BYTE_BUDGET,
+)
 from .types import CODEX_INTAKE_DISCIPLINE_DIGEST as CODEX_INTAKE_DISCIPLINE_DIGEST
 from .types import CODEX_INTAKE_DISCIPLINE_VERSION as CODEX_INTAKE_DISCIPLINE_VERSION
+from .types import CODEX_INTAKE_RULES as CODEX_INTAKE_RULES
 from .types import CODEX_INTERACTIVE_REQUIRED_ENV as CODEX_INTERACTIVE_REQUIRED_ENV
 from .types import CODEX_MCP_ENV_FORWARD_VARS as CODEX_MCP_ENV_FORWARD_VARS
 from .types import CODEX_MODEL_ALIASES as CODEX_MODEL_ALIASES
@@ -439,6 +446,7 @@ from .types import InputSpecType as InputSpecType
 from .types import InspectorCallback as InspectorCallback
 from .types import InspectorEvidence as InspectorEvidence
 from .types import InspectorVerdict as InspectorVerdict
+from .types import IntakeRuleDef as IntakeRuleDef
 from .types import InvariantDef as InvariantDef
 from .types import IssueLabelState as IssueLabelState
 from .types import KillReason as KillReason
@@ -619,6 +627,7 @@ from .types import parse_plan_paths as parse_plan_paths
 from .types import recipe_section_digest as recipe_section_digest
 from .types import recipe_section_element_digest as recipe_section_element_digest
 from .types import recipe_section_plan_digest as recipe_section_plan_digest
+from .types import render_intake_digest as render_intake_digest
 from .types import render_target_skill_command as render_target_skill_command
 from .types import resolve_payload_field as resolve_payload_field
 from .types import resolve_skill_name as resolve_skill_name

src/autoskillit/core/types/AGENTS.md

@@ -37,6 +37,7 @@ Type re-export hub and all typed building blocks for the autoskillit package (IL
 | `_type_helpers.py` | Text processing, skill-name extraction, and shared content-free validation utilities |
 | `_type_inspector.py` | Health Inspector types: `InspectorEvidence`, `InspectorVerdict`, `InspectorCallback` (issue #3533) |
 | `_type_invariant_registry.py` | Invariant registry: `InvariantDef` dataclass and `INVARIANT_REGISTRY` mapping prose prohibitions to runtime gates |
+| `_type_intake_policy.py` | Codex context-intake rule registry: `IntakeRuleDef`, `CODEX_INTAKE_RULES`, rendered digest, byte budgets |
 | `_type_phoropter.py` | Phoropter family/phase types: `PhoropterPrescription`, `ReadingToken`, `READING_TOKEN_PATTERN`, `PhoropterPhaseSkip`, `CrossDomainPrescription`, `CrossDomainAssessment` |
 | `_type_resume.py` | `ResumeSpec` discriminated union: `NoResume | BareResume | NamedResume` |
 | `_type_plugin_source.py` | `DirectInstall` (projection input) and `ProjectedPluginRoot` (the sole `PluginSource`) |
@@ -45,7 +46,7 @@ Type re-export hub and all typed building blocks for the autoskillit package (IL
 
 ## Architecture Notes
 
-Internal dependency DAG: enums -> constants_registries -> constants_features; enums -> results -> protocols -> helpers; enums -> phoropter; enums + phoropter -> tradition_manifest. All modules have zero `autoskillit` imports outside this sub-package (IL-0 hard constraint). Production code imports from `autoskillit.core`, not from this package directly.
+Internal dependency DAG: enums -> constants_registries -> constants_features; enums -> results -> protocols -> helpers; enums -> phoropter; enums + phoropter -> tradition_manifest. `_type_intake_policy` is a DAG leaf — stdlib-only, zero sibling imports. All modules have zero `autoskillit` imports outside this sub-package (IL-0 hard constraint). Production code imports from `autoskillit.core`, not from this package directly.
 
 ## Extension Bundle Pattern
 

src/autoskillit/core/types/__init__.py

@@ -36,6 +36,8 @@
 from ._type_helpers import __all__ as _helpers_all
 from ._type_inspector import *  # noqa: F401, F403
 from ._type_inspector import __all__ as _inspector_all
+from ._type_intake_policy import *  # noqa: F401, F403
+from ._type_intake_policy import __all__ as _intake_policy_all
 from ._type_invariant_registry import *  # noqa: F401, F403
 from ._type_invariant_registry import __all__ as _invariant_registry_all
 from ._type_phoropter import *  # noqa: F401, F403
@@ -93,6 +95,7 @@
     + _figure_spec_all
     + _helpers_all
     + _inspector_all
+    + _intake_policy_all
     + _invariant_registry_all
     + _phoropter_all
     + _plugin_source_all