parallel.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/testing/parallel.py
 
"""
Purpose
-------
Resource-keyed test parallelism (Plan 239 / TEST-01). A pytest worker
is a separate PROCESS, so in-process state is per-worker and safe;
what collides across workers is shared EXTERNAL state (a registry
database, a comms root, a tracked settings file, a bound port, writes
outside ``tmp_path``). This module classifies every test file by the
shared resources it touches, groups the conflicting files into the
connected components of that conflict graph, and hands the grouping to
pytest through xdist's ``--dist loadgroup``.
 
Responsibilities
----------------
- Pin the CLOSED resource vocabulary (Plan 239 P2), its KEYED
  refinement (Plan 247 P1: ``main_tree_write:<top>`` and
  ``env_subprocess:<family>``) and the ``unknown:<detail>``
  fail-closed escape hatch.
- STATIC detector: an :mod:`ast` pass over each test file plus the
  conftest fixtures it actually requests (:func:`analyse_source`,
  :func:`classify_source`, :func:`classify_tree`).
- DYNAMIC detector: the ``--oct-trace`` pytest plugin, which wraps
  ``open`` / ``os.environ`` / ``socket.bind`` / ``subprocess`` /
  ``os.chdir`` for the duration of each test and writes a per-process
  trace (:func:`load_traces`).
- Grouping and the committed manifest (:func:`build_groups`,
  :func:`build_manifest`, :func:`load_manifest`,
  :func:`resolve_groups`).
- The pytest-side seams a root conftest calls
  (:func:`resolve_worker_count`, :func:`configure_parallel_defaults`).
- The regeneration CLI: ``python -m oct.testing.parallel classify
  --root <subproject root>``.
 
Diagnostics
-----------
Domain: OCT-TESTING
Levels:
    L2: lifecycle (classification start/end, tracer install/remove)
    L3: semantic details (resolved cap, group counts)
    L4: deep tracing (per-file resource sets)
 
Contracts
---------
- A file's classification is the UNION of both detectors (P3):
  neither alone is authoritative. The union is why BOTH detectors
  must key what they can - one bare verdict re-widens a file the
  other had already narrowed.
- The vocabulary is CLOSED. A signature the detectors cannot map into
  it is recorded as ``unknown:<detail>``, which conflicts with every
  file that carries any resource at all (fail closed, P2/P4).
- A KEY only ever NARROWS (Plan 247 P1). Two keyed tokens of one
  family conflict when their keys are equal; a BARE family token
  conflicts with every member of its family, so anything a detector
  cannot attribute keeps precisely its pre-247 behaviour. Only
  ``unknown:*`` is a GLOBAL wildcard; a bare family token is not.
- The ENVIRONMENT is process-local for grouping (Plan 276 P1/P2). An
  ``os.environ`` write is recorded as ``env_mutation`` and connects
  the file to nothing: a worker is a separate PROCESS, so a write in
  one can never reach a child another one spawns. ``env_subprocess``
  is therefore recorded for a CHILD LAUNCH only, and a BARE launch is
  the one family wildcard that holds just the other bare launches -
  what a child does to genuinely shared state is what its
  :data:`LAUNCH_FAMILIES` entry declares (``main_tree_write:*``), and
  ``unknown:*`` still fails closed on everything else.
- Manifest keys are ROOT-RELATIVE POSIX paths, identical to the path
  part of a pytest nodeid. Every recorded value is machine
  independent: the manifest is committed, so no absolute path, user
  name or temp directory may ever reach it.
- A conftest contributes its AUTOUSE fixtures' resources to every file
  beneath it and its other fixtures' resources to the files that
  request them by name. Module-level conftest code and conftest hook
  functions are session-level uses that per-file grouping cannot
  separate, so they are deliberately not attributed to files.
- ``pytest`` is NEVER imported at module scope: it is an optional
  (``[test]`` extra) dependency of oct, so the plugin hooks below are
  plain functions that pytest discovers by name.
- No ``sys.exit`` outside the ``__main__`` CLI block.
 
Dependencies
------------
- oct.core.diagnostics (_dbg)
- oct.testing.errors (the spec-10 §12.4 taxonomy as data)
- oct.testing.manifest (the §9 workspace test blocks; ``max_workers``)
- oct.tools.atomic_write (atomic_write_text for writing the committed manifest)
- oct.core.workspace (lazy; find_workspace_root for the classify CLI)
- oct (lazy; the oct version recorded in the manifest)
- pytest (lazy; the xdist_group marker and UsageError in the plugin hooks)
- xdist (not imported; the pytest-xdist plugin whose worker count and loadgroup scheduling this module configures)
"""
 
from __future__ import annotations
 
import ast
import builtins
import functools
import io
import json
import os
import re
import socket
import subprocess
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any, Callable, Iterable, Mapping, Sequence
 
from oct.core.diagnostics import _dbg
from oct.testing.errors import (
    MANIFEST_TEST_BLOCK_INVALID,
    WORKSPACE_FILE_NOT_FOUND,
    OctTestError,
)
from oct.testing.manifest import load_test_manifest
from oct.tools.atomic_write import atomic_write_text
 
try:  # pragma: no cover - exercised by the shadowed-install path only
    from oct.testing.manifest import DEFAULT_MAX_WORKERS
except ImportError:  # pragma: no cover
    # An INSTALLED oct predating the P1 key (a worktree running against
    # an editable install rooted in another checkout). The literal is
    # the same shipped default the parser applies.
    DEFAULT_MAX_WORKERS = 24
 
 
# ---------------------------------------------------------------------
# P2 - the closed resource vocabulary
# ---------------------------------------------------------------------
 
#: The orchestration registry database (``agents.sqlite``).
RES_REGISTRY_SQLITE: str = "registry_sqlite"
 
#: The inter-agent comms root under ``.option_c/comms``.
RES_COMMS_ROOT: str = "comms_root"
 
#: The tracked runtime settings file.
RES_SETTINGS_FILE: str = "settings_file"
 
#: An ``octrc.json`` / ``.octrc.json`` tool-configuration file.
RES_OCTRC: str = "octrc"
 
#: A git worktree directory (creation, removal, or a write inside one).
RES_WORKTREE_DIR: str = "worktree_dir"
 
#: Any write outside ``tmp_path`` / ``tmp_path_factory`` and outside
#: the test's own package temp. KEYED by the top-level directory the
#: write lands in (P2): ``main_tree_write:docs`` and
#: ``main_tree_write:.option_c`` cannot collide, so they do not group.
RES_MAIN_TREE_WRITE: str = "main_tree_write"
 
#: A CHILD LAUNCH, and nothing else (Plan 276 P1). A child inherits
#: the environment, and what it does with it is invisible from here.
#: A child is OPAQUE to both detectors unless its argv names a family
#: the committed :data:`LAUNCH_FAMILIES` table vouches for, so a
#: launch is KEYED by that family (P3) and BARE otherwise.
RES_ENV_SUBPROCESS: str = "env_subprocess"
 
#: A write to ``os.environ`` outside ``monkeypatch`` (Plan 276 P1).
#: Recorded so the manifest keeps the full picture, but PROCESS-LOCAL:
#: an xdist worker is a separate process, so a write in worker A can
#: never reach a child worker B spawns. What CAN collide is what a
#: child does to shared state, and the keyed launch families already
#: declare that (``main_tree_write:*``) while ``unknown:*`` fails
#: closed on the rest.
RES_ENV_MUTATION: str = "env_mutation"
 
#: Outbound network use.
RES_NETWORK: str = "network"
 
#: A process-global working-directory change (``os.chdir``).
RES_CWD_CHANGE: str = "cwd_change"
 
#: A call into a known process-wide reset seam (session trust, grants,
#: runtime settings, the cooperative cancel flag, plan inventory).
RES_GLOBAL_SINGLETON_RESET: str = "global_singleton_reset"
 
#: Prefix of the parameterised port resource, ``port:<n>``.
PORT_PREFIX: str = "port:"
 
#: Prefix of the fail-closed escape hatch, ``unknown:<detail>``.
UNKNOWN_PREFIX: str = "unknown:"
 
#: Every non-parameterised member of the vocabulary.
RESOURCE_VOCABULARY: tuple[str, ...] = (
    RES_COMMS_ROOT,
    RES_CWD_CHANGE,
    RES_ENV_MUTATION,
    RES_ENV_SUBPROCESS,
    RES_GLOBAL_SINGLETON_RESET,
    RES_MAIN_TREE_WRITE,
    RES_NETWORK,
    RES_OCTRC,
    RES_REGISTRY_SQLITE,
    RES_SETTINGS_FILE,
    RES_WORKTREE_DIR,
)
 
#: Resources that are PROCESS-LOCAL: an xdist worker is a separate
#: process, so two files that both touch one of these can never
#: collide by running at the same time in different workers. They are
#: recorded (the manifest keeps the full picture) but they do NOT
#: connect files in the conflict graph.
#:
#: Grouping could not help them even if they did collide: a group
#: pins its files to ONE worker, which prevents CONCURRENCY, not
#: sequential adjacency - and adjacency is the only way a leaked
#: cwd or singleton reaches another test. Measured on the oc_chat
#: suite (2026-09-06): treating them as conflicting put 210 of 324
#: files into a single group, i.e. no parallelism at all, because
#: ``tests/unit/conftest.py`` resets the cooperative-cancel flag in
#: an autouse fixture for all 184 unit files.
#:
#: :data:`RES_ENV_MUTATION` joined them under Plan 276 for the same
#: reason: ``os.environ`` is per PROCESS, so a write in one worker is
#: invisible to every other worker and to the children they spawn.
#: Measured on the oc_chat suite (2026-09-15): treating an environment
#: write as evidence of a cross-worker conflict was one half of a
#: single 100-file group that held a quarter of the suite's tests.
PROCESS_LOCAL_RESOURCES: frozenset[str] = frozenset({
    RES_CWD_CHANGE,
    RES_ENV_MUTATION,
    RES_GLOBAL_SINGLETON_RESET,
})
 
 
#: Separator between a resource FAMILY and its key (P1).
KEY_SEPARATOR: str = ":"
 
#: The two families whose tokens may carry a key (Plan 247 P1). Both
#: were single undifferentiated tokens through plan 239, and both are
#: the reason the oc_chat conflict graph collapsed into one component
#: of 92 files: fifty files that each write somewhere different, and
#: fifty-eight that each launch something different, all conflicted.
#:
#: ``port:<n>`` and ``unknown:<detail>`` also carry a colon and are
#: deliberately NOT here: reading them as family-plus-key would make
#: ``unknown:a`` and ``unknown:b`` two non-conflicting resources, the
#: exact opposite of the fail-closed rule they exist for.
KEYED_FAMILIES: frozenset[str] = frozenset({
    RES_MAIN_TREE_WRITE,
    RES_ENV_SUBPROCESS,
})
 
#: Key of a traced write that landed ABOVE the package root and is not
#: temporary (P2). It is a constant, never the real directory name,
#: because the manifest that carries it is COMMITTED and no absolute
#: path may reach it; the angle brackets are not legal in a path
#: component, so it can never collide with a genuine top-level
#: directory called ``outside``.
OUTSIDE_KEY: str = "<outside>"
 
 
def keyed(family: str, key: str) -> str:
    """Build the keyed token ``<family>:<key>`` (P1).
 
    An empty *key* yields the BARE family token: a caller that could
    not attribute the use says so by producing the wildcard, never by
    inventing a key.
    """
    if not key:
        return family
    return f"{family}{KEY_SEPARATOR}{key}"
 
 
def resource_family(resource: str) -> str:
    """The family of *resource*, or the token itself when not keyed."""
    head, separator, _key = resource.partition(KEY_SEPARATOR)
    if separator and head in KEYED_FAMILIES:
        return head
    return resource
 
 
def resource_key(resource: str) -> str | None:
    """The key *resource* carries, or None when it is BARE."""
    head, separator, key = resource.partition(KEY_SEPARATOR)
    if separator and head in KEYED_FAMILIES:
        return key
    return None
 
 
def conflicts_with(left: str, right: str) -> bool:
    """True when two resource tokens can collide (the P1 rule).
 
    Equal keyed tokens conflict unless their launch family is known
    to share nothing (Plan 251 P4). A BARE token still conflicts with
    every member of its family, keyed or not - with ONE exception,
    :data:`RES_ENV_SUBPROCESS` (Plan 276 P2): a bare launch conflicts
    with other bare launches only.
 
    That exception is not a relaxation of the fail-closed rule; it is
    the rule applied to what an xdist worker actually is. A bare
    ``main_tree_write`` means "a write to a directory this pass could
    not read", and the directory it could not read may be any of the
    keyed ones, so it must conflict with all of them. A bare
    ``env_subprocess`` means "a child this pass could not name", and
    the one thing a keyed sibling would share with it - the inherited
    environment - is per PROCESS: a worker is a process, so that
    inheritance never crosses one. What a child does to state that IS
    shared is what its family declares through its own resources
    (``main_tree_write:*``), and what ``unknown:*`` fails closed on.
 
    Parameters
    ----------
    left, right
        Resource tokens as the detectors record them.
 
    Returns
    -------
    bool
        Whether a file carrying *left* may run beside one carrying
        *right*.
    """
    if left == right:
        return is_conflicting(left)
    family = resource_family(left)
    if family != resource_family(right):
        return False
    if family == RES_ENV_SUBPROCESS:
        return False
    return resource_key(left) is None or resource_key(right) is None
 
 
def is_conflicting(resource: str) -> bool:
    """True when *resource* can collide ACROSS worker processes.
 
    Every ``unknown:<detail>`` conflicts by construction. An empty
    launch-family entry has no self-conflict (Plan 251 P4), but its
    token remains visible to the GLOBAL ``unknown:*`` wildcard in
    :func:`build_groups` - no longer to a bare launch, which since
    Plan 276 P2 holds only the other bare launches. Other keyed
    resources remain shared.
    """
    family = resource_family(resource)
    if family == RES_ENV_SUBPROCESS and resource_key(resource) is not None:
        return LAUNCH_FAMILIES.get(resource_key(resource)) != ()
    return family not in PROCESS_LOCAL_RESOURCES
 
 
def is_known_resource(resource: str) -> bool:
    """True when *resource* is inside the closed P2 vocabulary.
 
    Parameters
    ----------
    resource
        A resource token as the detectors record it.
 
    Returns
    -------
    bool
        True for a vocabulary member, a keyed token of a
        :data:`KEYED_FAMILIES` member, or a well-formed ``port:<n>``;
        False for ``unknown:<detail>`` and anything unrecognised.
    """
    if resource in RESOURCE_VOCABULARY:
        return True
    if resource.startswith(PORT_PREFIX):
        return resource[len(PORT_PREFIX):].isdigit()
    return bool(resource_key(resource))
 
 
def unknown(detail: str) -> str:
    """Build the fail-closed ``unknown:<detail>`` token.
 
    *detail* must be machine independent (no absolute paths, no user
    names): the manifest that carries it is committed.
    """
    return f"{UNKNOWN_PREFIX}{detail}"
 
 
# ---------------------------------------------------------------------
# Manifest shape (P4)
# ---------------------------------------------------------------------
 
#: Schema version of ``parallel_manifest.json``.
SCHEMA_VERSION: str = "1.0"
 
#: Conventional basename of the committed manifest.
MANIFEST_FILENAME: str = "parallel_manifest.json"
 
#: Group of a file that touches no shared resource.
GROUP_FREE: str = "free"
 
#: Group applied at collection to a file with NO manifest entry.
GROUP_UNCLASSIFIED: str = "unclassified-serial"
 
#: Prefix of the per-test xdist group names a ``free`` file's items
#: carry (Plan 286 P1). The manifest still records the group ``free``
#: per FILE; only the name stamped on a collected ITEM is per test.
FREE_GROUP_PREFIX: str = f"{GROUP_FREE}:"
 
#: Marker that keeps ONE free file's tests in one group (Plan 286 P2).
#: The opt-out for a file whose tests depend on each other's order or
#: on module state; every use is named in the plan's report with the
#: witness that justified it.
WHOLE_FILE_MARKER: str = "whole_file"
 
 
@dataclass(frozen=True)
class FileEntry:
    """One classified file.
 
    Attributes
    ----------
    path
        Root-relative POSIX path (a pytest nodeid's path part).
    resources
        Sorted resource tokens, the UNION of both detectors.
    group
        The xdist group name the conftest applies to every item
        collected from this file.
    """
 
    path: str
    resources: tuple[str, ...]
    group: str
 
 
@dataclass(frozen=True)
class ParallelManifest:
    """A parsed ``parallel_manifest.json``."""
 
    schema_version: str
    generated_at: str
    tool_version: str
    files: Mapping[str, FileEntry]
 
 
# ---------------------------------------------------------------------
# Path -> resource mapping (shared by both detectors)
# ---------------------------------------------------------------------
 
#: Ordered ``(substring, resource)`` markers matched against a
#: lowercased, forward-slashed path hint. First match wins, so the
#: specific shared files come before the generic directories.
_PATH_MARKERS: tuple[tuple[str, str], ...] = (
    ("agents.sqlite", RES_REGISTRY_SQLITE),
    ("registry.sqlite", RES_REGISTRY_SQLITE),
    (".option_c/orchestration", RES_REGISTRY_SQLITE),
    (".option_c/comms", RES_COMMS_ROOT),
    ("octrc.json", RES_OCTRC),
    ("settings.json", RES_SETTINGS_FILE),
    ("option_c_branches", RES_WORKTREE_DIR),
    (".git/worktrees", RES_WORKTREE_DIR),
)
 
 
def resource_for_path_hint(hint: str) -> str:
    """Map a path-ish *hint* to its shared resource.
 
    Parameters
    ----------
    hint
        Any path fragment: string constants lifted out of the source,
        or a concrete path the dynamic tracer observed.
 
    Returns
    -------
    str
        The matching vocabulary member, or
        :data:`RES_MAIN_TREE_WRITE` when the hint names no known
        shared file - an unrecognised write is still a write, and
        still shared.
    """
    needle = hint.replace("\\", "/").lower()
    for marker, resource in _PATH_MARKERS:
        if marker in needle:
            return resource
    return RES_MAIN_TREE_WRITE
 
 
def path_key(hint: str) -> str | None:
    """The top-level component of a ROOT-RELATIVE path *hint* (P2).
 
    Parameters
    ----------
    hint
        A path relative to the package root, in either slash style.
 
    Returns
    -------
    str | None
        The first component - the directory the write lands in - or
        None for an absolute path, a drive-qualified path, or an
        upward escape. None means "no key", which produces the bare
        wildcard: a key that is not a real top inside the root would
        separate two writes that really can collide.
    """
    text = hint.replace("\\", "/").strip()
    if not text or text.startswith("/"):
        return None
    parts = [part for part in text.split("/") if part not in ("", ".")]
    if not parts or parts[0] == ".." or KEY_SEPARATOR in parts[0]:
        return None
    return parts[0]
 
 
def write_resource(hint: str, key: str | None) -> str:
    """The resource of a write to *hint*, keyed where P2 allows.
 
    Keying only ever refines the UNATTRIBUTED write: a hint the
    vocabulary already names something specific - the comms root, the
    registry database, an ``octrc.json`` - keeps that token, because
    two writes to the same shared FILE collide whichever directory
    they were reached through.
 
    Parameters
    ----------
    hint
        Any path fragment, as :func:`resource_for_path_hint` takes it.
    key
        The top-level directory the write landed in, or None when the
        caller could not read one.
 
    Returns
    -------
    str
        A specific vocabulary member, ``main_tree_write:<key>``, or
        the bare :data:`RES_MAIN_TREE_WRITE`.
    """
    resource = resource_for_path_hint(hint)
    if resource != RES_MAIN_TREE_WRITE or not key:
        return resource
    return keyed(RES_MAIN_TREE_WRITE, key)
 
 
# ---------------------------------------------------------------------
# P3 - launch families (shared by both detectors)
# ---------------------------------------------------------------------
 
#: Stand-in for ``sys.executable`` in a STATICALLY read argv. It is
#: the one non-constant argv head whose meaning is fixed, and the
#: module name beside it is a literal; without it the interpreter
#: launches would stay bare and re-widen every file the tracer had
#: just keyed, because a classification is the UNION of both
#: detectors.
PYTHON_ARGV0: str = "<python>"
 
#: Program stems that are a Python interpreter.
_PYTHON_PROGRAM = re.compile(r"^python[0-9tw.]*$", re.IGNORECASE)
 
#: Value-less interpreter flags that may sit between the interpreter
#: and ``-m``. The oc_chat suite launches ``python -P -m oc_chat``
#: (safe path) in five places; without skipping these the family reads
#: as ``python`` and the launch stays bare for no reason.
_PYTHON_FLAGS: frozenset[str] = frozenset({
    "-b", "-bb", "-B", "-d", "-E", "-i", "-I", "-O", "-OO", "-P",
    "-q", "-s", "-S", "-u", "-v", "-x",
})
 
#: What each launch FAMILY is KNOWN to touch, beyond its own keyed
#: ``env_subprocess:<family>`` token (P3). This table is the ONLY
#: evidence that narrows a child: a family absent from it contributes
#: the BARE token and conflicts with every launcher, because a child
#: process is otherwise opaque to an ast pass and an in-process
#: tracer alike.
#:
#: Per-entry justification (Plan 247; widen an entry the moment a
#: trace shows it touching more, never narrow one on a guess):
#:
#: - ``git`` - every git launch in the suite runs against a repository
#:   the test built under ``tmp_path``; git writes inside the
#:   repository it is pointed at and nowhere else.
#: - ``oc_chat.testing`` - the §20 stand-in. Its sessions have lived
#:   under temp since plan 239. The existing empty resource entry is
#:   process-local under Plan 251 P4.
#: - ``oct`` / ``oct.cli`` - the toolchain writes the timing ledger and
#:   its caches under ``.option_c``, so a launch joins the files that
#:   write there directly.
LAUNCH_FAMILIES: Mapping[str, tuple[str, ...]] = {
    "git": (),
    "oc_chat.testing": (),
    # Plan 251 serial child audit: 4 console/module and 9 replay launches
    # completed; cross-PID diagnostic-log cleanup writes the shared logs.
    # Registry stays opaque: its 67 native Git children were not probed.
    "oc_chat": (f"{RES_MAIN_TREE_WRITE}{KEY_SEPARATOR}logs",),
    "oc_chat.scripts.replay": (f"{RES_MAIN_TREE_WRITE}{KEY_SEPARATOR}logs",),
    "oct": (f"{RES_MAIN_TREE_WRITE}{KEY_SEPARATOR}.option_c",),
    "oct.cli": (f"{RES_MAIN_TREE_WRITE}{KEY_SEPARATOR}.option_c",),
}
 
 
def launch_family(argv: Any) -> str | None:
    """The program family of a child launch, or None when opaque (P3).
 
    ``python -m <module>`` names the MODULE, because that is what the
    child actually IS - ``oc_chat.testing``, ``oct.cli`` and
    ``oc_chat.registry`` are three families, not one interpreter. Any
    other program names its own basename without extension. A bare
    interpreter (``python -c``, ``python script.py``) names only
    ``python``, which the table deliberately does not vouch for.
 
    Parameters
    ----------
    argv
        The argv as the caller passed it: a sequence whose unreadable
        elements are None, a shell string, or anything else.
 
    Returns
    -------
    str | None
        The family name, or None when the argv has no readable head.
    """
    if not isinstance(argv, (list, tuple)) or not argv:
        return None
    head = argv[0]
    if head is None:
        return None
    if head == PYTHON_ARGV0:
        stem = "python"
    else:
        try:
            text = os.fspath(head)
        except TypeError:
            return None
        stem = PurePosixPath(str(text).replace("\\", "/")).stem
    if not stem:
        return None
    if _PYTHON_PROGRAM.match(stem):
        index = 1
        while index < len(argv) and argv[index] in _PYTHON_FLAGS:
            index += 1
        if index + 1 < len(argv) and argv[index] == "-m" and argv[index + 1]:
            return str(argv[index + 1])
    return stem
 
 
def launch_resources(argv: Any) -> frozenset[str]:
    """The resources one child launch contributes (P3).
 
    A family named in :data:`LAUNCH_FAMILIES` contributes its own
    keyed token plus everything the table vouches for. A family the
    table does not name - and an argv with no readable family at all -
    contributes the BARE token, which conflicts with every launcher.
    """
    func = "launch_resources"
    family = launch_family(argv)
    if family is None or family not in LAUNCH_FAMILIES:
        _dbg("OCT-TESTING", func, f"unattributed launch: {family}", 4)
        return frozenset({RES_ENV_SUBPROCESS})
    return frozenset({
        keyed(RES_ENV_SUBPROCESS, family), *LAUNCH_FAMILIES[family],
    })
 
 
# ---------------------------------------------------------------------
# P3a - the static detector
# ---------------------------------------------------------------------
 
#: Fixture/parameter names that seed the temp-path taint set.
_TMP_SEEDS: frozenset[str] = frozenset({
    "tmp_path", "tmp_path_factory", "tmpdir", "tmpdir_factory",
})
 
#: Calls whose result is a temporary path.
_TMP_CALLS: frozenset[str] = frozenset({
    "mkdtemp", "mkstemp", "mktemp", "gettempdir", "TemporaryDirectory",
    "NamedTemporaryFile", "TemporaryFile",
})
 
#: ``pathlib.Path`` methods that write.
_PATH_WRITES: frozenset[str] = frozenset({
    "write_text", "write_bytes", "mkdir", "touch", "unlink", "rmdir",
    "rename", "replace", "symlink_to", "hardlink_to", "chmod",
})
 
#: ``os`` functions that write.
_OS_WRITES: frozenset[str] = frozenset({
    "remove", "unlink", "mkdir", "makedirs", "rmdir", "removedirs",
    "rename", "renames", "replace", "truncate", "symlink", "link",
    "chmod", "utime", "mknod",
})
 
#: ``shutil`` functions that write, mapped to the argument index that
#: names the DESTINATION (the path actually written).
_SHUTIL_WRITES: Mapping[str, int] = {
    "copy": 1, "copy2": 1, "copyfile": 1, "copytree": 1, "move": 1,
    "rmtree": 0, "make_archive": 0, "unpack_archive": 1, "chown": 0,
}
 
#: Subprocess launch seams reached as ``<module>.<attr>``.
_SUBPROCESS_ATTRS: frozenset[str] = frozenset({
    "run", "Popen", "call", "check_call", "check_output",
    "create_subprocess_exec", "create_subprocess_shell",
})
 
#: Modules whose ``_SUBPROCESS_ATTRS`` really do launch a child.
_SUBPROCESS_MODULES: frozenset[str] = frozenset({
    "subprocess", "asyncio",
})
 
#: Bare function names that launch a child process.
_SUBPROCESS_NAMES: frozenset[str] = frozenset({
    "run_cancellable", "getoutput", "getstatusoutput",
})
 
#: Network seams reached by attribute.
_NETWORK_ATTRS: frozenset[str] = frozenset({
    "urlopen", "urlretrieve", "create_connection",
})
 
#: Modules whose use is outbound network use.
_NETWORK_MODULES: frozenset[str] = frozenset({
    "requests", "httpx", "urllib",
})
 
#: Known process-wide reset seams (P2 ``global_singleton_reset``).
_RESET_NAMES: frozenset[str] = frozenset({
    "reset_tool_cancel", "signal_tool_cancel", "reset_session_trust",
    "reset_inventory_for_tests", "reset_runtime_settings_for_tests",
    "revoke_session_grants", "grant_work_dir_for_session",
})
 
#: Imported module basenames whose entry points reach a shared file
#: unless the call is handed a temp-derived path.
_MODULE_HINTS: Mapping[str, str] = {
    "agent_instances": RES_REGISTRY_SQLITE,
    "agent_registry": RES_REGISTRY_SQLITE,
    "orchestration": RES_REGISTRY_SQLITE,
}
 
 
@dataclass(frozen=True)
class ModuleAnalysis:
    """The static reading of one module (a test file or a conftest).
 
    Attributes
    ----------
    resources
        Union of every resource the module's own code touches.
    fixtures
        Fixture name -> the resources that fixture's body touches.
    autouse
        Names of the module's ``autouse=True`` fixtures.
    tmp_fixtures
        Fixtures whose value is derived from a temp path; requesting
        one taints the requesting parameter.
    requested
        Fixture/parameter names every ``def`` in the module requests.
    """
 
    resources: frozenset[str]
    fixtures: Mapping[str, frozenset[str]]
    autouse: frozenset[str]
    tmp_fixtures: frozenset[str]
    requested: frozenset[str]
 
 
def _target_names(node: ast.AST) -> set[str]:
    """Every plain name bound by an assignment target *node*."""
    return {
        sub.id for sub in ast.walk(node) if isinstance(sub, ast.Name)
    }
 
 
def _is_tmp_expr(node: ast.AST | None, tainted: frozenset[str]) -> bool:
    """True when *node* is derived from a temporary directory."""
    if node is None:
        return False
    for sub in ast.walk(node):
        if isinstance(sub, ast.Name) and sub.id in tainted:
            return True
        if isinstance(sub, ast.Call):
            func = sub.func
            name = getattr(func, "attr", None) or getattr(func, "id", None)
            if name in _TMP_CALLS:
                return True
    return False
 
 
def _fixture_decorated(node: ast.AST) -> tuple[bool, bool]:
    """Return ``(is_fixture, is_autouse)`` for a function definition."""
    is_fixture = False
    autouse = False
    for dec in getattr(node, "decorator_list", []):
        call = dec if isinstance(dec, ast.Call) else None
        target = call.func if call is not None else dec
        name = getattr(target, "attr", None) or getattr(target, "id", None)
        if name != "fixture":
            continue
        is_fixture = True
        for keyword in (call.keywords if call is not None else []):
            if keyword.arg == "autouse" and isinstance(
                keyword.value, ast.Constant,
            ):
                autouse = bool(keyword.value.value)
    return is_fixture, autouse
 
 
def _returns_tmp(node: ast.AST, tainted: frozenset[str]) -> bool:
    """True when every value a fixture body returns/yields is temp."""
    seen = False
    for sub in ast.walk(node):
        if isinstance(sub, (ast.Return, ast.Yield)):
            if sub.value is None:
                continue
            seen = True
            if not _is_tmp_expr(sub.value, tainted):
                return False
    return seen
 
 
def _collect_taint(tree: ast.AST, seeds: Iterable[str]) -> frozenset[str]:
    """Fixpoint over the names bound to a temporary-path value.
 
    Parameters
    ----------
    tree
        The parsed module.
    seeds
        Extra names to seed with: the temp-returning fixtures of the
        conftests above this module.
 
    Returns
    -------
    frozenset[str]
        Every name a write target may safely be derived from.
    """
    func = "_collect_taint"
    tainted: set[str] = set(_TMP_SEEDS) | set(seeds)
    for _ in range(4):
        before = len(tainted)
        frozen = frozenset(tainted)
        for node in ast.walk(tree):
            if isinstance(node, (ast.Assign, ast.AnnAssign, ast.NamedExpr)):
                if _is_tmp_expr(node.value, frozen):
                    targets = getattr(node, "targets", None) or [node.target]
                    for target in targets:
                        tainted |= _target_names(target)
            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                is_fixture, _autouse = _fixture_decorated(node)
                if is_fixture and _returns_tmp(node, frozen):
                    tainted.add(node.name)
            elif isinstance(node, ast.withitem):
                if node.optional_vars is not None and _is_tmp_expr(
                    node.context_expr, frozen,
                ):
                    tainted |= _target_names(node.optional_vars)
            elif isinstance(node, ast.For):
                if _is_tmp_expr(node.iter, frozen):
                    tainted |= _target_names(node.target)
        if len(tainted) == before:
            break
    _dbg("OCT-TESTING", func, f"tainted names: {len(tainted)}", 4)
    return frozenset(tainted)
 
 
def _scope_nodes(tree: ast.AST) -> Iterable[ast.AST]:
    """Walk one lexical body without importing names from nested bodies."""
    pending = list(ast.iter_child_nodes(tree))
    while pending:
        node = pending.pop()
        yield node
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef,
                             ast.ClassDef, ast.Lambda, ast.ListComp,
                             ast.SetComp, ast.DictComp, ast.GeneratorExp)):
            continue
        pending.extend(ast.iter_child_nodes(node))
 
 
def _proven_tmp_expr(node: ast.AST | None, tainted: frozenset[str]) -> bool:
    """Read a helper actual without assuming an opaque call returns its input."""
    if isinstance(node, ast.Name):
        return node.id in tainted
    if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Div):
        return _proven_tmp_expr(node.left, tainted)
    if isinstance(node, ast.JoinedStr):
        return bool(node.values) and isinstance(
            node.values[0], ast.FormattedValue,
        ) and _proven_tmp_expr(node.values[0].value, tainted)
    if not isinstance(node, ast.Call):
        return False
    name = getattr(node.func, 'attr', None) or getattr(node.func, 'id', None)
    if name in _TMP_CALLS:
        return True
    if name in _PATH_CONSTRUCTORS | {'str', 'fspath'}:
        return bool(node.args) and _proven_tmp_expr(node.args[0], tainted)
    if isinstance(node.func, ast.Attribute) and name in (
        'joinpath', 'absolute', 'resolve', 'with_name', 'with_suffix',
    ):
        return _proven_tmp_expr(node.func.value, tainted)
    return False
 
 
def _scope_proofs(
    tree: ast.AST,
    seeds: Iterable[str],
    *,
    parameters: Iterable[str] = (),
) -> frozenset[str]:
    """Prove local aliases only when every binding has a temporary origin."""
    bindings: dict[str, list[ast.AST | None]] = {}
    for node in _scope_nodes(tree):
        pairs = []
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            fixture = isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and (
                isinstance(tree, ast.Module) and _fixture_decorated(node)[0]
                and node.name in seeds
            )
            if not fixture:
                bindings.setdefault(node.name, []).append(None)
        elif isinstance(node, (ast.Import, ast.ImportFrom)):
            for alias in node.names:
                bindings.setdefault(alias.asname or alias.name.split(".")[0], []).append(None)
        elif isinstance(node, (ast.ExceptHandler, ast.MatchAs, ast.MatchStar)):
            if node.name:
                bindings.setdefault(node.name, []).append(None)
        elif isinstance(node, ast.MatchMapping) and node.rest:
            bindings.setdefault(node.rest, []).append(None)
        if isinstance(node, (ast.Assign, ast.AnnAssign, ast.NamedExpr)):
            targets = getattr(node, 'targets', None) or [node.target]
            pairs = [(target, node.value) for target in targets]
        elif isinstance(node, ast.withitem) and node.optional_vars is not None:
            pairs = [(node.optional_vars, node.context_expr)]
        elif isinstance(node, (ast.For, ast.AsyncFor)):
            pairs = [(node.target, node.iter)]
        elif isinstance(node, (ast.AugAssign, ast.Delete)):
            targets = getattr(node, 'targets', None) or [node.target]
            pairs = [(target, None) for target in targets]
        for target, value in pairs:
            for bound in ast.walk(target):
                if isinstance(bound, ast.Name) and isinstance(
                    bound.ctx, (ast.Store, ast.Del),
                ):
                    bindings.setdefault(bound.id, []).append(value)
    tainted = set(seeds) - set(bindings)
    if isinstance(tree, (ast.FunctionDef, ast.AsyncFunctionDef)):
        tainted -= _parameter_names(tree) - set(parameters)
    while True:
        before = set(tainted)
        for name, values in bindings.items():
            if all(_proven_tmp_expr(value, frozenset(tainted)) for value in values):
                tainted.add(name)
        if tainted == before:
            return frozenset(tainted)
 
 
def _parameter_names(node: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]:
    """All names bound by one function's argument specification."""
    spec = node.args
    names = {arg.arg for arg in (*spec.posonlyargs, *spec.args, *spec.kwonlyargs)}
    names.update(arg.arg for arg in (spec.vararg, spec.kwarg) if arg is not None)
    return names
 
 
def _helper_parameter_taints(
    tree: ast.Module,
    base: frozenset[str],
    seeds: Iterable[str] = (),
) -> dict[ast.AST, frozenset[str]]:
    """Prove helper parameters from every visible call, without name leakage.
 
    Only direct references to uniquely named module functions qualify.
    An escaping reference, reflection, rebinding or starred call keeps the
    helper opaque. Inferred names belong to that function's scope only.
    """
    functions = {node for node in ast.walk(tree) if isinstance(
        node, (ast.FunctionDef, ast.AsyncFunctionDef),
    )}
    parents = {child: node for node in ast.walk(tree)
               for child in ast.iter_child_nodes(node)}
    boundaries = functions | {node for node in ast.walk(tree) if isinstance(
        node, (ast.ClassDef, ast.Lambda, ast.ListComp, ast.SetComp,
               ast.DictComp, ast.GeneratorExp),
    )}
    scopes: dict[ast.AST, ast.AST] = {}
    for node in ast.walk(tree):
        parent = parents.get(node)
        while parent is not None and parent not in boundaries:
            parent = parents.get(parent)
        scopes[node] = parent or tree
    fixture_seeds = set(_TMP_SEEDS) | set(seeds)
    fixture_seeds.update(
        node.name for node in functions
        if _fixture_decorated(node)[0] and node.name in base
    )
    # A nested global/nonlocal declaration can change a visible origin
    # without rebinding it in the caller body. Keep those names opaque.
    mutable_origins = {name for node in ast.walk(tree)
                       if isinstance(node, (ast.Global, ast.Nonlocal))
                       for name in node.names}
    module_proofs = _scope_proofs(tree, fixture_seeds) - mutable_origins
    proofs = {
        node: _scope_proofs(
            node, module_proofs,
            parameters=(fixture_seeds if node.name.startswith("test_")
                        or _fixture_decorated(node)[0] else ()),
        ) for node in functions
    }
    proofs.update({node: frozenset() for node in boundaries - functions})
    proofs[tree] = module_proofs
    taints = {}
    candidates = {}
    for helper in tree.body:
        if helper not in functions or helper.name.startswith("test_"):
            continue
        if _fixture_decorated(helper)[0]:
            continue
        # Existing module-wide name taint cannot prove a helper's own
        # parameters or locals: both can shadow unrelated temporary names.
        local = _parameter_names(helper) | {
            node.id for node in _scope_nodes(helper)
            if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store)
        }
        taints[helper] = (base - local) | _scope_proofs(helper, module_proofs)
        calls = _closed_helper_calls(tree, helper, parents)
        if calls:
            candidates[helper] = calls
    while True:
        changed = False
        for helper, calls in candidates.items():
            inferred = set()
            parameters = (*helper.args.posonlyargs, *helper.args.args,
                          *helper.args.kwonlyargs)
            for parameter in parameters:
                name = parameter.arg
                if all(_proven_tmp_expr(
                    _helper_actual(helper, call, name), proofs[scopes[call]],
                ) for call in calls):
                    inferred.add(name)
            # A write through a reassigned parameter is not proved by its
            # incoming value. Keep it opaque even if another call is safe.
            inferred -= _rebound_parameters(helper, inferred)
            proved = _scope_proofs(
                helper, module_proofs | inferred, parameters=inferred,
            )
            local = _parameter_names(helper) | {
                node.id for node in _scope_nodes(helper)
                if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store)
            }
            taints[helper] = (base - local) | proved
            if proved != proofs[helper]:
                proofs[helper] = proved
                changed = True
        if not changed:
            break
    return taints
 
 
def _closed_helper_calls(
    tree: ast.Module,
    helper: ast.FunctionDef | ast.AsyncFunctionDef,
    parents: Mapping[ast.AST, ast.AST],
) -> list[ast.Call]:
    """Return every direct call, or no proof when the helper can escape."""
    calls = []
    if helper.decorator_list and not _fixture_decorated(helper)[0]:
        return calls
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            if node is not helper and node.name == helper.name:
                return []
        if isinstance(node, ast.arg) and node.arg == helper.name:
            return []
        if isinstance(node, ast.alias) and (
            node.asname or node.name.split('.')[0]
        ) == helper.name:
            return []
        if isinstance(node, ast.Attribute) and node.attr == helper.name:
            return []
        if isinstance(node, ast.Call) and (
            (getattr(node.func, 'id', None) or getattr(node.func, 'attr', None))
            in {'getattr', 'globals', 'locals', 'vars', 'eval', 'exec'}
        ):
            return []
        if not isinstance(node, ast.Name) or node.id != helper.name:
            continue
        parent = parents.get(node)
        if not (isinstance(node.ctx, ast.Load)
                and isinstance(parent, ast.Call) and parent.func is node):
            return []
        if any(isinstance(arg, ast.Starred) for arg in parent.args):
            return []
        if any(keyword.arg is None for keyword in parent.keywords):
            return []
        calls.append(parent)
    return calls
 
 
def _helper_actual(
    helper: ast.FunctionDef | ast.AsyncFunctionDef,
    call: ast.Call,
    name: str,
) -> ast.AST | None:
    """Read a parameter's actual argument; omissions stay unresolved."""
    positional = [arg.arg for arg in (*helper.args.posonlyargs, *helper.args.args)]
    if len(call.args) > len(positional) and helper.args.vararg is None:
        return None
    if name in {arg.arg for arg in helper.args.posonlyargs} and any(
        keyword.arg == name for keyword in call.keywords
    ):
        return None
    values = [keyword.value for keyword in call.keywords if keyword.arg == name]
    if name in positional and positional.index(name) < len(call.args):
        values.append(call.args[positional.index(name)])
    return values[0] if len(values) == 1 else None
 
 
def _rebound_parameters(tree: ast.AST, names: set[str]) -> set[str]:
    """Reject new parameter proofs when the body assigns their names."""
    return {node.id for node in ast.walk(tree)
            if isinstance(node, ast.Name) and node.id in names
            and isinstance(node.ctx, (ast.Store, ast.Del))}
 
 
def _import_hints(tree: ast.AST) -> dict[str, str]:
    """Map names imported from a hinted module to their resource."""
    hints: dict[str, str] = {}
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom):
            resource = _hint_for_dotted(node.module or "")
            if resource is None:
                continue
            for alias in node.names:
                hints[alias.asname or alias.name] = resource
        elif isinstance(node, ast.Import):
            for alias in node.names:
                resource = _hint_for_dotted(alias.name)
                if resource is None:
                    continue
                bound = alias.asname or alias.name.split(".")[0]
                hints[bound] = resource
    return hints
 
 
def _hint_for_dotted(dotted: str) -> str | None:
    """The hinted resource of a dotted module path, if any."""
    for part in dotted.split("."):
        if part in _MODULE_HINTS:
            return _MODULE_HINTS[part]
    return None
 
 
def _string_hint(node: ast.AST | None) -> str:
    """Join every string constant, attribute and name inside *node*."""
    if node is None:
        return ""
    parts: list[str] = []
    for sub in ast.walk(node):
        if isinstance(sub, ast.Constant) and isinstance(sub.value, str):
            parts.append(sub.value)
        elif isinstance(sub, ast.Attribute):
            parts.append(sub.attr)
        elif isinstance(sub, ast.Name):
            parts.append(sub.id)
    return "/".join(parts)
 
 
#: Constructors whose all-literal arguments still name a literal path.
_PATH_CONSTRUCTORS: frozenset[str] = frozenset({
    "Path", "PurePath", "PurePosixPath", "PureWindowsPath",
    "PosixPath", "WindowsPath",
})
 
#: ``pathlib`` methods that return the SAME location they were called
#: on, so a resolved literal passes through them unchanged (Plan 284
#: P1). They are what stands between ``Path(__file__)`` and the
#: ``.parent`` beside it in the shape the suite actually writes.
_PATH_IDENTITY_CALLS: frozenset[str] = frozenset({"resolve", "absolute"})
 
#: Stand-in for the pid inside a resolved f-string (Plan 284 P1). The
#: value is irrelevant - :func:`_literal_fstring` only admits a
#: placeholder that cannot reach the component :func:`path_key` reads -
#: but it must be a legal path character, because the joined result is
#: parsed as a path.
_PID_PLACEHOLDER: str = "0"
 
#: Ceiling on ``.parents[N]``. A literal index this large has already
#: escaped every real tree, and the bound keeps a pathological constant
#: from walking a loop the classifier gains nothing from.
_MAX_PARENTS_INDEX: int = 64
 
 
def _is_absolute_literal(text: str) -> bool:
    """True when a literal path fragment is not root-relative.
 
    The same three shapes :func:`path_key` refuses - a leading slash, a
    drive letter, and anything else carrying the key separator in its
    first component - because a fragment that is absolute swallows
    everything joined BEFORE it.
    """
    head = text.replace("\\", "/")
    return head.startswith("/") or KEY_SEPARATOR in head.split("/")[0]
 
 
def _join_literal(left: str, right: str) -> str:
    """Join two literal fragments the way ``pathlib`` joins them.
 
    An absolute right operand DISCARDS the left one - ``Path('docs') /
    '/var/lib'`` is ``/var/lib`` - so reading the left as the key would
    invent a directory the write never touches. Returning the right
    operand whole hands :func:`path_key` an absolute path, which it
    already refuses.
    """
    if _is_absolute_literal(right):
        return right
    if not right:
        return left
    if not left:
        return right
    return f"{left.rstrip('/')}/{right}"
 
 
def _parent_literal(text: str) -> str | None:
    """The parent of a root-relative literal fragment, or None (P1).
 
    Climbing past the package root yields a ``..`` prefix rather than
    an empty string, because the two mean different things and only the
    first is an escape: the root itself is a legitimate location with
    no top-level component, while ``..`` is outside the tree the
    manifest describes. :func:`path_key` refuses both, which is the
    fail-closed answer for each.
    """
    if _is_absolute_literal(text):
        return None
    parts = [part for part in text.replace("\\", "/").split("/")
             if part not in ("", ".")]
    if parts and parts[-1] != "..":
        parts.pop()
    else:
        parts.append("..")
    return "/".join(parts)
 
 
def _is_pid_expr(node: ast.AST | None) -> bool:
    """True when *node* is ``os.getpid()``, or a ``str()`` of one."""
    if not isinstance(node, ast.Call):
        return False
    name = getattr(node.func, "attr", None) or getattr(node.func, "id", None)
    if name == "getpid":
        return True
    return name == "str" and len(node.args) == 1 and _is_pid_expr(node.args[0])
 
 
def _literal_fstring(node: ast.JoinedStr) -> str | None:
    """An f-string read as a path when its placeholders cannot move the key.
 
    The key is the FIRST component, so a placeholder is harmless
    exactly while every separator in the string comes BEFORE it: a pid
    stamped into the basename leaves the directory literal, and
    ``f'{os.getpid()}/run.txt'`` does not. Only the pid qualifies as a
    placeholder at all - a temp-derived one never reaches here, because
    :meth:`_StaticDetector._write` drops the write before it asks for a
    key, and anything else is a value this pass cannot read.
    """
    parts: list[str] = []
    placed = False
    for value in node.values:
        if isinstance(value, ast.Constant) and isinstance(value.value, str):
            if placed and ("/" in value.value or "\\" in value.value):
                return None
            parts.append(value.value)
            continue
        if not isinstance(value, ast.FormattedValue):
            return None
        if not _is_pid_expr(value.value):
            return None
        placed = True
        parts.append(_PID_PLACEHOLDER)
    return "".join(parts)
 
 
def _is_os_path_join(func: ast.AST) -> bool:
    """True for the fully spelled ``os.path.join`` attribute chain."""
    return (
        isinstance(func, ast.Attribute)
        and func.attr == "join"
        and isinstance(func.value, ast.Attribute)
        and func.value.attr == "path"
        and getattr(func.value.value, "id", None) == "os"
    )
 
 
def _literal_parents_index(
    node: ast.Subscript,
    constants: Mapping[str, str],
) -> str | None:
    """``<literal>.parents[N]`` for a constant, non-negative literal N."""
    owner = node.value
    if not (isinstance(owner, ast.Attribute) and owner.attr == "parents"):
        return None
    index = node.slice
    if not isinstance(index, ast.Constant) or isinstance(index.value, bool):
        return None
    if not isinstance(index.value, int) or not (
        0 <= index.value <= _MAX_PARENTS_INDEX
    ):
        return None
    resolved = _literal_path(owner.value, constants)
    for _ in range(index.value + 1):
        if resolved is None:
            return None
        resolved = _parent_literal(resolved)
    return resolved
 
 
def _literal_path(
    node: ast.AST | None,
    constants: Mapping[str, str] | None = None,
) -> str | None:
    """The path *node* names when this pass can read it in full (P2).
 
    ``_string_hint`` deliberately mashes names and attributes together
    so a MARKER can be spotted anywhere in an expression; that mash is
    useless as a key, because ``Path('docs/x')`` would key on ``Path``
    and land in a different group from ``open('docs/y', 'w')`` - two
    writes into the same directory, separated. So keying reads the
    expression strictly instead: a constant, a ``Path(...)`` of
    constants, or a ``/`` join of those.
 
    Plan 284 P1 widens WHAT counts as readable without widening how
    loosely it is read. *constants* carries the module's own
    unambiguous names - ``__file__`` among them, bound to the module's
    ROOT-RELATIVE path - so ``Path(__file__).resolve().parent /
    'golden'`` names a real directory instead of an opaque one. Each
    added shape either resolves completely or yields None; none of them
    guesses at a component it could not read.
 
    Parameters
    ----------
    node
        The expression naming a write target.
    constants
        Module-level names this pass proved to be literal paths, as
        :func:`_module_constants` builds them. Absent, the function
        reads exactly what it read before Plan 284.
 
    Returns
    -------
    str | None
        A path in :func:`path_key`'s space - root-relative unless the
        expression itself was absolute or escaped upward - or None when
        any part of the expression is unreadable.
    """
    names: Mapping[str, str] = constants if constants is not None else {}
    if isinstance(node, ast.Constant):
        return node.value if isinstance(node.value, str) else None
    if isinstance(node, ast.Name):
        return names.get(node.id)
    if isinstance(node, ast.JoinedStr):
        return _literal_fstring(node)
    if isinstance(node, ast.Subscript):
        return _literal_parents_index(node, names)
    if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Div):
        left = _literal_path(node.left, names)
        right = _literal_path(node.right, names)
        if left is None or right is None:
            return None
        return _join_literal(left, right)
    if isinstance(node, ast.Attribute):
        if node.attr != "parent":
            return None
        base = _literal_path(node.value, names)
        return None if base is None else _parent_literal(base)
    if not isinstance(node, ast.Call):
        return None
    name = getattr(node.func, "attr", None) or getattr(node.func, "id", None)
    if isinstance(node.func, ast.Attribute) and name in _PATH_IDENTITY_CALLS:
        return _literal_path(node.func.value, names) if not node.args else None
    if _is_os_path_join(node.func):
        return _literal_sequence(node, names, joiner=_join_literal)
    if name not in _PATH_CONSTRUCTORS or node.keywords or not node.args:
        return None
    return _literal_sequence(node, names, joiner=_join_literal)
 
 
def _literal_sequence(
    node: ast.Call,
    constants: Mapping[str, str],
    *,
    joiner: Callable[[str, str], str],
) -> str | None:
    """Fold a call's positional arguments through *joiner*, or None."""
    if not node.args or node.keywords:
        return None
    resolved: str | None = None
    for arg in node.args:
        part = _literal_path(arg, constants)
        if part is None:
            return None
        resolved = part if resolved is None else joiner(resolved, part)
    return resolved
 
 
def _binding_counts(tree: ast.AST) -> dict[str, int]:
    """How many times each name is BOUND anywhere in a module (P1).
 
    A module constant is only readable while it is a constant, so the
    resolver admits a name bound exactly once - by the module-level
    assignment it read. Everything that can put a different value
    behind a name counts here: a second assignment, a parameter, a
    ``def``/``class``, an import alias, an ``except`` or ``match``
    capture. A ``global``/``nonlocal`` declaration counts twice, which
    disqualifies the name outright: it can rewrite the constant from a
    scope this walk cannot order.
    """
    counts: dict[str, int] = {}
 
    def add(name: str, weight: int = 1) -> None:
        counts[name] = counts.get(name, 0) + weight
 
    for node in ast.walk(tree):
        if isinstance(node, ast.Name) and isinstance(
            node.ctx, (ast.Store, ast.Del),
        ):
            add(node.id)
        elif isinstance(node, ast.arg):
            add(node.arg)
        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef,
                               ast.ClassDef)):
            add(node.name)
        elif isinstance(node, ast.alias):
            add(node.asname or node.name.split(".")[0])
        elif isinstance(node, (ast.ExceptHandler, ast.MatchAs, ast.MatchStar)):
            if node.name:
                add(node.name)
        elif isinstance(node, ast.MatchMapping) and node.rest:
            add(node.rest)
        elif isinstance(node, (ast.Global, ast.Nonlocal)):
            for name in node.names:
                add(name, 2)
    return counts
 
 
def _module_constants(tree: ast.Module, module_path: str | None) -> dict[str, str]:
    """The module-level names that unambiguously name a literal path (P1).
 
    Read in SOURCE ORDER, so a constant may be built from one declared
    above it - the shape the suite writes as ``ROOT = ...`` then
    ``SEED = ROOT / 'seed'`` - and never from one declared below, which
    does not exist yet at that point in the module either.
 
    Parameters
    ----------
    tree
        The parsed module.
    module_path
        The module's ROOT-RELATIVE POSIX path, or None when the caller
        analysed a source string with no place in the tree. It is what
        ``__file__`` resolves to; without it the anchor stays opaque
        and every shape built on it stays bare.
 
    Returns
    -------
    dict[str, str]
        Name -> literal path, for names bound exactly once.
    """
    func = "_module_constants"
    bound = _binding_counts(tree)
    constants: dict[str, str] = {}
    if module_path is not None and not bound.get("__file__"):
        constants["__file__"] = module_path
    for node in tree.body:
        if isinstance(node, ast.Assign) and len(node.targets) == 1:
            target: ast.AST = node.targets[0]
        elif isinstance(node, ast.AnnAssign):
            target = node.target
        else:
            continue
        if not isinstance(target, ast.Name) or bound.get(target.id) != 1:
            continue
        resolved = _literal_path(getattr(node, "value", None), constants)
        if resolved is not None:
            constants[target.id] = resolved
    _dbg("OCT-TESTING", func, f"{len(constants)} literal constant(s)", 4)
    return constants
 
 
def _argv_element(node: ast.AST) -> str | None:
    """One argv element as a literal string, or None when unreadable."""
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value
    if (
        isinstance(node, ast.Attribute)
        and node.attr == "executable"
        and getattr(node.value, "id", None) == "sys"
    ):
        return PYTHON_ARGV0
    return None
 
 
def _literal_argv(node: ast.Call, attr: str) -> list[str | None] | None:
    """The argv of a launch call, as far as the literals allow (P3).
 
    ``create_subprocess_exec`` takes the program and its arguments as
    separate positional parameters; every other seam takes one
    sequence.
 
    A STARRED element hides the length and the position of everything
    AFTER it, so the argv is truncated there and only the readable
    PREFIX comes back (Plan 276 P3). The prefix decides the family and
    nothing else: the family is the PROGRAM - ``argv[0]``, plus the
    ``-m <module>`` pair when ``argv[0]`` is an interpreter - and a
    starred tail cannot change what the program is. A prefix too short
    to name a module still reads as the bare interpreter, which the
    table deliberately does not vouch for, and an argv whose FIRST
    element is starred has no readable program at all and stays bare.
    Measured on the oc_chat suite (2026-09-15): nine ``['git', *args]``
    helpers and one ``[sys.executable, '-m', 'oc_chat.scripts.replay',
    *args]`` helper were bare for this reason alone.
    """
    if not node.args:
        return None
    if attr == "create_subprocess_exec":
        return [_argv_element(arg) for arg in node.args]
    first = node.args[0]
    if not isinstance(first, (ast.List, ast.Tuple)):
        return None
    argv: list[str | None] = []
    for element in first.elts:
        if isinstance(element, ast.Starred):
            break
        argv.append(_argv_element(element))
    return argv or None
 
 
def _write_mode(node: ast.Call) -> bool:
    """True when an ``open``-like call requests a writing mode."""
    mode: str | None = None
    if len(node.args) > 1 and isinstance(node.args[1], ast.Constant):
        value = node.args[1].value
        mode = value if isinstance(value, str) else None
    for keyword in node.keywords:
        if keyword.arg == "mode" and isinstance(keyword.value, ast.Constant):
            value = keyword.value.value
            mode = value if isinstance(value, str) else mode
    if mode is None:
        return False
    return any(flag in mode for flag in ("w", "a", "x", "+"))
 
 
def _is_path_replace(node: ast.Call) -> bool:
    """True only for the one-argument ``pathlib.Path.replace`` (P1).
 
    ``replace`` is the one name in :data:`_PATH_WRITES` that several
    unrelated types also carry, and the ARITY separates them exactly.
    ``Path.replace(target)`` takes one argument, positionally or by
    that keyword name; ``str.replace(old, new)`` needs at least two;
    ``datetime.replace`` names a field, never ``target``; and
    ``dataclasses.replace(record, **changes)`` passes the record plus
    its changes. Measured on the oc_chat suite (Plan 284 P0): fifteen
    of the thirty-eight files carrying the BARE ``main_tree_write``
    carried it for a write that does not exist, and a bare token welds
    every keyed writer of its family into one group.
 
    This narrows how the CALL is read, never the vocabulary. A real
    ``Path.replace`` still records its write, and ``os.replace`` /
    ``shutil`` are matched by their own earlier arms and never reach
    here.
    """
    if len(node.args) + len(node.keywords) != 1:
        return False
    if node.keywords:
        return node.keywords[0].arg == "target"
    return True
 
 
def _is_environ(node: ast.AST | None) -> bool:
    """True when *node* is ``os.environ`` or a subscript of it."""
    if isinstance(node, ast.Subscript):
        node = node.value
    return isinstance(node, ast.Attribute) and node.attr == "environ"
 
 
class _StaticDetector(ast.NodeVisitor):
    """Record the P2 resources one subtree touches."""
 
    def __init__(
        self,
        tainted: frozenset[str],
        hints: Mapping[str, str],
        scopes: Mapping[ast.AST, frozenset[str]] | None = None,
        constants: Mapping[str, str] | None = None,
    ) -> None:
        self.tainted = tainted
        self.base_tainted = tainted
        self.scopes = scopes or {}
        self.hints = hints
        self.constants = dict(constants or {})
        self.resources: set[str] = set()
        self.env_write = False
        self.spawns = False
 
    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        """Apply newly inferred taint only inside the owning helper."""
        previous = self.tainted
        try:
            for child in ast.iter_child_nodes(node):
                self.tainted = (
                    self.scopes.get(node, self.base_tainted)
                    if child in node.body else previous
                )
                self.visit(child)
        finally:
            self.tainted = previous
 
    visit_AsyncFunctionDef = visit_FunctionDef
 
    def visit_ClassDef(self, node: ast.ClassDef) -> None:
        """Nested class bindings cannot inherit a helper parameter proof."""
        previous = self.tainted
        try:
            self.tainted = self.base_tainted
            self.generic_visit(node)
        finally:
            self.tainted = previous
 
    def visit_Lambda(self, node: ast.Lambda) -> None:
        """A lambda has its own parameter bindings, which remain opaque."""
        previous = self.tainted
        try:
            self.tainted = self.base_tainted
            self.generic_visit(node)
        finally:
            self.tainted = previous
 
    # -- helpers ----------------------------------------------------
    def _write(self, target: ast.AST | None) -> None:
        """Record a write to *target* unless it is temp-derived.
 
        Keyed by the top-level directory when *target* is a literal
        path this pass can read in full (P2); bare otherwise.
        """
        if _is_tmp_expr(target, self.tainted):
            return
        literal = _literal_path(target, self.constants)
        key = path_key(literal) if literal is not None else None
        self.resources.add(write_resource(_string_hint(target), key))
 
    def _launch(self, node: ast.Call, attr: str) -> None:
        """Record a child launch, keyed by family when readable (P3)."""
        self.resources |= set(launch_resources(_literal_argv(node, attr)))
 
    def _bind(self, node: ast.Call) -> None:
        """Record a socket bind as ``port:<n>``, or fail closed."""
        port: int | None = None
        if node.args and isinstance(node.args[0], ast.Tuple):
            for element in node.args[0].elts:
                if isinstance(element, ast.Constant) and isinstance(
                    element.value, int,
                ):
                    port = element.value
        if port is None:
            self.resources.add(unknown("socket-bind-dynamic-port"))
        elif port > 0:
            self.resources.add(f"{PORT_PREFIX}{port}")
 
    # -- visitors ---------------------------------------------------
    def visit_Assign(self, node: ast.Assign) -> None:
        """Catch ``os.environ[...] = ...``."""
        for target in node.targets:
            if _is_environ(target):
                self.env_write = True
        self.generic_visit(node)
 
    def visit_Delete(self, node: ast.Delete) -> None:
        """Catch ``del os.environ[...]``."""
        for target in node.targets:
            if _is_environ(target):
                self.env_write = True
        self.generic_visit(node)
 
    def visit_Call(self, node: ast.Call) -> None:
        """Dispatch one call expression through every signature."""
        func = node.func
        name = getattr(func, "id", None)
        attr = getattr(func, "attr", None)
        owner = getattr(func, "value", None)
        owner_name = getattr(owner, "id", None)
 
        if name == "open" and _write_mode(node):
            self._write(node.args[0] if node.args else None)
        elif name in _RESET_NAMES or attr in _RESET_NAMES:
            self.resources.add(RES_GLOBAL_SINGLETON_RESET)
        elif name in _SUBPROCESS_NAMES:
            self.spawns = True
        elif attr is not None:
            self._visit_attr_call(node, attr, owner, owner_name)
 
        hinted = self.hints.get(name or "") or self.hints.get(owner_name or "")
        if hinted is not None and not any(
            _is_tmp_expr(arg, self.tainted) for arg in node.args
        ):
            self.resources.add(hinted)
        if any(keyword.arg == "env" for keyword in node.keywords):
            self.env_write = True
        self.generic_visit(node)
 
    def _visit_attr_call(
        self,
        node: ast.Call,
        attr: str,
        owner: ast.AST | None,
        owner_name: str | None,
    ) -> None:
        """The ``<something>.<attr>(...)`` half of :meth:`visit_Call`."""
        args = node.args
        if owner_name == "os" and attr == "chdir":
            self.resources.add(RES_CWD_CHANGE)
        elif owner_name == "os" and attr in _OS_WRITES:
            self._write(args[0] if args else None)
        elif owner_name == "os" and attr in ("system", "popen", "execv"):
            self.spawns = True
        elif owner_name == "shutil" and attr in _SHUTIL_WRITES:
            index = _SHUTIL_WRITES[attr]
            self._write(args[index] if len(args) > index else None)
        elif owner_name in _SUBPROCESS_MODULES and attr in _SUBPROCESS_ATTRS:
            self._launch(node, attr)
        elif owner_name == "sqlite3" and attr == "connect":
            target = args[0] if args else next(
                (kw.value for kw in node.keywords if kw.arg == "database"), None,
            )
            value = target.value if isinstance(target, ast.Constant) else None
            if not (isinstance(value, str) and (
                value == ":memory:" or value.startswith("file::memory:")
            )):
                self._write(target)
        elif attr == "bind":
            self._bind(node)
        elif attr in _NETWORK_ATTRS or owner_name in _NETWORK_MODULES:
            self.resources.add(RES_NETWORK)
        elif _is_environ(owner) and attr in (
            "update", "pop", "setdefault", "clear",
        ):
            self.env_write = True
        elif attr == "open" and _write_mode(node):
            self._write(owner)
        elif attr in _PATH_WRITES:
            if attr == "replace" and not _is_path_replace(node):
                return
            self._write(owner)
 
    def result(self) -> frozenset[str]:
        """The recorded resources, with the launch and mutation joins.
 
        A child launch alone is enough for :data:`RES_ENV_SUBPROCESS`:
        it inherits the environment, and what it does with it is
        invisible from here. ``self.spawns`` carries only the launches
        this pass could NOT attribute - a shell string, a computed
        argv, a bare ``os.system`` - and those contribute the bare
        wildcard (P3).
 
        An ``os.environ`` write contributes :data:`RES_ENV_MUTATION`
        and nothing else (Plan 276 P1): it is recorded, so the manifest
        keeps the full picture, but it is process-local, so on its own
        it leaves the file free.
        """
        resources = set(self.resources)
        if self.spawns:
            resources.add(RES_ENV_SUBPROCESS)
        if self.env_write:
            resources.add(RES_ENV_MUTATION)
        return frozenset(resources)
 
 
def _requested_names(tree: ast.AST) -> frozenset[str]:
    """Every parameter name any function in *tree* requests."""
    names: set[str] = set()
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            spec = node.args
            for arg in (*spec.posonlyargs, *spec.args, *spec.kwonlyargs):
                names.add(arg.arg)
    return frozenset(names)
 
 
def analyse_source(
    source: str,
    *,
    filename: str = "<test>",
    tmp_seeds: Iterable[str] = (),
) -> ModuleAnalysis:
    """Statically analyse one module.
 
    Parameters
    ----------
    source
        Module source text.
    filename
        Reported in diagnostics only.
    tmp_seeds
        Fixture names from an enclosing conftest whose value is a
        temporary path; requesting one taints the parameter.
 
    Returns
    -------
    :class:`ModuleAnalysis`
 
    Notes
    -----
    A file that does not parse is not silently clean: it comes back
    carrying ``unknown:parse-error``, which conflicts with everything.
    """
    func = "analyse_source"
    try:
        tree = ast.parse(source, filename=filename)
    except SyntaxError as exc:
        _dbg("OCT-TESTING", func, f"parse error in {filename}: {exc}", 2)
        return ModuleAnalysis(
            resources=frozenset({unknown("parse-error")}),
            fixtures={},
            autouse=frozenset(),
            tmp_fixtures=frozenset(),
            requested=frozenset(),
        )
 
    hints = _import_hints(tree)
    tainted = _collect_taint(tree, tmp_seeds)
    scopes = _helper_parameter_taints(tree, tainted, tmp_seeds)
    constants = _module_constants(
        tree, filename if filename.endswith(".py") else None,
    )
    module = _StaticDetector(tainted, hints, scopes, constants)
    module.visit(tree)
 
    fixtures: dict[str, frozenset[str]] = {}
    autouse: set[str] = set()
    tmp_fixtures: set[str] = set()
    for node in ast.walk(tree):
        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            continue
        is_fixture, is_autouse = _fixture_decorated(node)
        if not is_fixture:
            continue
        detector = _StaticDetector(
            scopes.get(node, tainted), hints, scopes, constants,
        )
        for child in node.body:
            detector.visit(child)
        fixtures[node.name] = detector.result()
        if is_autouse:
            autouse.add(node.name)
        if node.name in tainted:
            tmp_fixtures.add(node.name)
 
    return ModuleAnalysis(
        resources=module.result(),
        fixtures=fixtures,
        autouse=frozenset(autouse),
        tmp_fixtures=frozenset(tmp_fixtures),
        requested=_requested_names(tree),
    )
 
 
def classify_source(
    source: str,
    *,
    filename: str = "<test>",
    conftests: Sequence[ModuleAnalysis] = (),
) -> frozenset[str]:
    """Classify one test module, folding in the conftests above it.
 
    A file inherits the resources of every AUTOUSE conftest fixture on
    its path, plus those of the conftest fixtures it requests by name
    (P3a).
    """
    func = "classify_source"
    seeds: set[str] = set()
    for analysis in conftests:
        seeds |= set(analysis.tmp_fixtures)
    own = analyse_source(source, filename=filename, tmp_seeds=seeds)
 
    resources = set(own.resources)
    for analysis in conftests:
        for name in analysis.autouse:
            resources |= analysis.fixtures.get(name, frozenset())
        for name in own.requested:
            resources |= analysis.fixtures.get(name, frozenset())
    _dbg("OCT-TESTING", func, f"{filename}: {sorted(resources)}", 4)
    return frozenset(resources)
 
 
def _read(path: Path) -> str:
    """Read *path* as UTF-8, tolerating a stray decoding error."""
    return path.read_text(encoding="utf-8", errors="replace")
 
 
def _conftest_chain(
    root: Path,
    path: Path,
    cache: dict[Path, ModuleAnalysis],
) -> list[ModuleAnalysis]:
    """Analyse every ``conftest.py`` from *root* down to *path*."""
    directories: list[Path] = []
    current = path.parent
    while True:
        directories.append(current)
        if current == root or current.parent == current:
            break
        current = current.parent
 
    chain: list[ModuleAnalysis] = []
    for directory in reversed(directories):
        conftest = directory / "conftest.py"
        if not conftest.is_file():
            continue
        if conftest not in cache:
            cache[conftest] = analyse_source(
                _read(conftest),
                filename=conftest.relative_to(root).as_posix(),
            )
        chain.append(cache[conftest])
    return chain
 
 
def classify_tree(
    root: Path,
    test_roots: Sequence[str],
    *,
    pattern: str = "test_*.py",
) -> dict[str, frozenset[str]]:
    """Statically classify every test file under *test_roots*.
 
    Parameters
    ----------
    root
        Subproject root; manifest keys are relative to it.
    test_roots
        Root-relative test directories (the workspace manifest's
        ``test.test_roots``).
    pattern
        Test-file glob, matching pytest's ``python_files``.
 
    Returns
    -------
    dict[str, frozenset[str]]
        Root-relative POSIX path -> resource set.
    """
    func = "classify_tree"
    _dbg("OCT-TESTING", func, f"root={root} roots={list(test_roots)}", 2)
    cache: dict[Path, ModuleAnalysis] = {}
    results: dict[str, frozenset[str]] = {}
 
    for test_root in test_roots:
        base = root / test_root
        if not base.is_dir():
            _dbg("OCT-TESTING", func, f"missing test root {test_root}", 2)
            continue
        for path in sorted(base.rglob(pattern)):
            if not path.is_file():
                continue
            rel = path.relative_to(root).as_posix()
            results[rel] = classify_source(
                _read(path),
                filename=rel,
                conftests=_conftest_chain(root, path, cache),
            )
    _dbg("OCT-TESTING", func, f"classified {len(results)} file(s)", 2)
    return results
 
 
# ---------------------------------------------------------------------
# P4 - grouping: connected components of the conflict graph
# ---------------------------------------------------------------------
 
def _find(parent: dict[str, str], path: str) -> str:
    """Union-find root of *path*, with path compression."""
    while parent[path] != path:
        parent[path] = parent[parent[path]]
        path = parent[path]
    return path
 
 
def _union(parent: dict[str, str], left: str, right: str) -> None:
    """Merge the components of *left* and *right*."""
    root_left, root_right = _find(parent, left), _find(parent, right)
    if root_left != root_right:
        parent[root_right] = root_left
 
 
def _slug(resource: str) -> str:
    """Group-name form of a resource token."""
    return resource.replace(":", "-").replace("/", "-")
 
 
def _named_components(
    components: Mapping[str, list[str]],
    sets: Mapping[str, frozenset[str]],
) -> list[tuple[str, list[str]]]:
    """Assign deterministic names to the conflict components."""
    ordered: list[tuple[tuple[str, ...], list[str]]] = []
    for members in components.values():
        resources: set[str] = set()
        for member in members:
            resources |= sets[member]
        ordered.append((tuple(sorted(resources)), sorted(members)))
    ordered.sort(key=lambda item: (item[0], item[1][0]))
 
    named: list[tuple[str, list[str]]] = []
    used: dict[str, int] = {}
    for resources, members in ordered:
        stem = _slug(resources[0])
        used[stem] = used.get(stem, 0) + 1
        name = stem if used[stem] == 1 else f"{stem}-{used[stem]}"
        named.append((name, members))
    return named
 
 
def build_groups(
    classified: Mapping[str, Iterable[str]],
    *,
    pins: Mapping[str, str] | None = None,
) -> dict[str, str]:
    """Group files into the connected components of the conflict graph.
 
    Files sharing any CONFLICTING resource are connected, where
    "sharing" follows :func:`conflicts_with`: two KEYED tokens share
    only when their keys are equal, and a BARE family token shares
    with every keyed member of its own family (Plan 247 P1). A file
    carrying an ``unknown:*`` token connects to every file that
    carries any conflicting resource (fail closed, P2); a file left
    with none - no resources at all, or only
    :data:`PROCESS_LOCAL_RESOURCES` - is :data:`GROUP_FREE` and
    connects to nothing. Names derive from each component's own
    sorted resources, so the same input always produces the same
    names.
 
    ``env_subprocess`` is the one family whose BARE token does not
    reach its keyed members (Plan 276 P2). A worker is a separate
    PROCESS, so the environment a child inherits never crosses one,
    and an unattributable launch therefore holds only the other
    unattributable launches. Measured on the oc_chat suite
    (2026-09-15): the removed weld alone put 100 files carrying a
    quarter of the suite's tests into a single group, of which 46
    held the bare token.
 
    Parameters
    ----------
    classified
        Root-relative path -> resource set.
    pins
        Operator-pinned ``path -> group`` overrides (P7). A pinned
        file leaves the graph entirely and keeps its own group name.
 
    Returns
    -------
    dict[str, str]
        Root-relative path -> group name.
    """
    func = "build_groups"
    pins = dict(pins or {})
    sets = {path: frozenset(res) for path, res in classified.items()}
    # A non-conflicting launch token must remain visible to the GLOBAL
    # wildcard, which fails closed on everything. A BARE launch no
    # longer makes it busy (Plan 276 P2).
    unpinned = [res for path, res in sets.items() if path not in pins]
    global_wildcard = any(r.startswith(UNKNOWN_PREFIX)
                          for res in unpinned for r in res)
    busy = {
        path: frozenset(r for r in resources if is_conflicting(r) or (
            resource_family(r) == RES_ENV_SUBPROCESS and global_wildcard
        ))
        for path, resources in sets.items()
        if path not in pins
    }
    busy = {path: res for path, res in busy.items() if res}
 
    parent: dict[str, str] = {path: path for path in busy}
    by_resource: dict[str, list[str]] = {}
    by_family: dict[str, list[str]] = {}
    family_wildcards: dict[str, list[str]] = {}
    wildcards: list[str] = []
    for path, resources in busy.items():
        for resource in resources:
            by_resource.setdefault(resource, []).append(path)
            family = resource_family(resource)
            if family in KEYED_FAMILIES:
                by_family.setdefault(family, []).append(path)
                if resource_key(resource) is None:
                    family_wildcards.setdefault(family, []).append(path)
            if resource.startswith(UNKNOWN_PREFIX):
                wildcards.append(path)
    for resource, members in by_resource.items():
        if not is_conflicting(resource):
            continue
        for other in members[1:]:
            _union(parent, members[0], other)
    # A BARE family token conflicts with every keyed member of its own
    # family and with nothing else (P1); only ``unknown:*`` is global.
    # ``env_subprocess`` is exempt (Plan 276 P2): its bare token holds
    # only the other bare tokens, which the equal-token pass above has
    # already joined.
    for family, holders in family_wildcards.items():
        if family == RES_ENV_SUBPROCESS:
            continue
        for path in holders:
            for other in by_family.get(family, ()):
                _union(parent, path, other)
    for path in wildcards:
        for other in busy:
            _union(parent, path, other)
 
    components: dict[str, list[str]] = {}
    for path in busy:
        components.setdefault(_find(parent, path), []).append(path)
 
    groups: dict[str, str] = {}
    for name, members in _named_components(components, busy):
        for member in members:
            groups[member] = name
    for path in sets:
        if path in pins:
            groups[path] = pins[path]
        elif path not in groups:
            groups[path] = GROUP_FREE
    _dbg(
        "OCT-TESTING", func,
        f"{len(components)} conflict group(s) over {len(busy)} file(s)", 3,
    )
    return groups
 
 
# ---------------------------------------------------------------------
# Manifest read / write
# ---------------------------------------------------------------------
 
def build_manifest(
    classified: Mapping[str, Iterable[str]],
    *,
    tool_version: str,
    pins: Mapping[str, str] | None = None,
    generated_at: str | None = None,
) -> dict[str, Any]:
    """Build the committed manifest document (the P4 shape)."""
    func = "build_manifest"
    groups = build_groups(classified, pins=pins)
    files = {
        path: {
            "resources": sorted(classified[path]),
            "group": groups[path],
        }
        for path in sorted(classified)
    }
    stamp = generated_at or datetime.now(timezone.utc).strftime(
        "%Y-%m-%dT%H:%M:%SZ",
    )
    _dbg("OCT-TESTING", func, f"manifest over {len(files)} file(s)", 2)
    return {
        "schema_version": SCHEMA_VERSION,
        "generated_at": stamp,
        "tool_version": tool_version,
        "files": files,
    }
 
 
def write_manifest(path: Path, document: Mapping[str, Any]) -> None:
    """Write *document* to *path* as stable, newline-terminated JSON."""
    func = "write_manifest"
    path.parent.mkdir(parents=True, exist_ok=True)
    text = json.dumps(document, indent=2) + "\n"
    atomic_write_text(path, text, encoding="utf-8", newline="\n")
    _dbg("OCT-TESTING", func, f"wrote {path}", 2)
 
 
def load_manifest(path: Path) -> ParallelManifest:
    """Load and validate a committed ``parallel_manifest.json``.
 
    Raises
    ------
    OctTestError
        ``9102`` when the file is unreadable, is not JSON, or does not
        carry the P4 shape. A malformed manifest never degrades to an
        empty one: that would silently widen parallelism.
    """
    func = "load_manifest"
    try:
        raw = json.loads(Path(path).read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"parallel manifest {path} is unreadable: {exc}",
        ) from exc
    if not isinstance(raw, dict) or not isinstance(raw.get("files"), dict):
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"parallel manifest {path} has no 'files' object",
        )
 
    files: dict[str, FileEntry] = {}
    for key, value in raw["files"].items():
        group = value.get("group") if isinstance(value, dict) else None
        resources = value.get("resources") if isinstance(value, dict) else None
        if not isinstance(group, str) or not isinstance(resources, list):
            raise OctTestError(
                MANIFEST_TEST_BLOCK_INVALID,
                f"parallel manifest entry {key!r} is malformed",
            )
        files[key] = FileEntry(
            path=key, resources=tuple(resources), group=group,
        )
    _dbg("OCT-TESTING", func, f"loaded {len(files)} entry/entries", 3)
    return ParallelManifest(
        schema_version=str(raw.get("schema_version", "")),
        generated_at=str(raw.get("generated_at", "")),
        tool_version=str(raw.get("tool_version", "")),
        files=files,
    )
 
 
def resolve_groups(
    manifest: ParallelManifest,
    relpaths: Iterable[str],
) -> tuple[dict[str, str], tuple[str, ...]]:
    """Map collected file paths to groups, failing closed on absence.
 
    Returns
    -------
    tuple
        ``(path -> group, missing)``. A collected file with no
        manifest entry maps to :data:`GROUP_UNCLASSIFIED` and is
        listed in *missing*, so a stale manifest can never silently
        widen parallelism.
    """
    func = "resolve_groups"
    groups: dict[str, str] = {}
    missing: set[str] = set()
    for relpath in relpaths:
        entry = manifest.files.get(relpath)
        if entry is None:
            groups[relpath] = GROUP_UNCLASSIFIED
            missing.add(relpath)
        else:
            groups[relpath] = entry.group
    if missing:
        _dbg(
            "OCT-TESTING", func,
            f"{len(missing)} collected file(s) absent from the manifest", 2,
        )
    return groups, tuple(sorted(missing))
 
 
# ---------------------------------------------------------------------
# P1/P5 - the worker cap and the pytest option seams
# ---------------------------------------------------------------------
 
def _raw_max_workers(workspace_root: Path, package: str) -> int:
    """Read ``max_workers`` straight from the manifest JSON.
 
    The compatibility path for an INSTALLED ``oct.testing.manifest``
    that predates the P1 key; it applies the same validation the
    parser does, so an invalid value still refuses rather than
    degrading to the default.
    """
    func = "_raw_max_workers"
    path = workspace_root / ".option_c_workspace.json"
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        _dbg("OCT-TESTING", func, f"raw manifest read failed: {exc}", 2)
        return DEFAULT_MAX_WORKERS
    for entry in raw.get("subprojects", []):
        if not isinstance(entry, dict) or entry.get("name") != package:
            continue
        block = entry.get("test")
        if not isinstance(block, dict) or "max_workers" not in block:
            return DEFAULT_MAX_WORKERS
        value = block["max_workers"]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise OctTestError(
                MANIFEST_TEST_BLOCK_INVALID,
                f"subprojects[{package}].test.max_workers must be a "
                f"positive integer, got {value!r}",
            )
        return value
    return DEFAULT_MAX_WORKERS
 
 
def resolve_max_workers(workspace_root: Path, package: str) -> int:
    """Read ``subprojects[].test.max_workers`` for *package* (P1/P5).
 
    REFUSES a manifest-level failure instead of degrading to the
    shipped default (Plan 247 P5): a workspace manifest that EXISTS
    and cannot be loaded, and a package that manifest does not
    declare, are both configuration errors, and the only caller is a
    root conftest deciding how many workers to inject - degrading
    there ran two dozen workers against a workspace nobody could
    parse, silently.
 
    A manifest that is ABSENT ENTIRELY is not a misconfiguration and
    still resolves the shipped default (operator ruling 2026-09-08):
    there is no configuration to be wrong about, and the §20 stand-in
    is asked to run over standalone fixture trees that carry no
    workspace manifest by design - refusing there declined to
    parallelise fourteen witnesses that had nothing wrong with them.
    An ABSENT ``max_workers`` key on a DECLARED package likewise still
    defaults, as does an installed manifest parser that predates the
    key, through :func:`_raw_max_workers`.
 
    Raises
    ------
    OctTestError
        Whatever :func:`load_test_manifest` raises for a workspace
        manifest that exists but is unusable, and ``9102`` when
        *package* is not declared in one that loaded.
    """
    func = "resolve_max_workers"
    try:
        manifest = load_test_manifest(Path(workspace_root))
    except OctTestError as exc:
        if exc.code != WORKSPACE_FILE_NOT_FOUND:
            raise
        _dbg("OCT-TESTING", func, f"no workspace at {workspace_root}", 2)
        return DEFAULT_MAX_WORKERS
    config = manifest.packages_by_name.get(package)
    if config is None:
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"package {package!r} is not declared in the workspace "
            f"manifest at {workspace_root}, so its worker cap cannot "
            f"be resolved",
        )
    value = getattr(config, "max_workers", None)
    if value is None:
        value = _raw_max_workers(Path(workspace_root), package)
    _dbg("OCT-TESTING", func, f"{package} max_workers={value}", 3)
    return int(value)
 
 
def available_cpu_count() -> int:
    """The affinity-aware CPU count (ADR-T-11's v0.4 amendment).
 
    The ONE normative resolution every worker-count formula reads::
 
        os.process_cpu_count()             # Python >= 3.13
        else len(os.sched_getaffinity(0))  # where the OS exposes it
        else (os.cpu_count() or 1)
 
    A bare ``os.cpu_count()`` overcounts inside a container and under
    CPU affinity (OI-027) - exactly where over-spawning hurts most.
    """
    func = "available_cpu_count"
    getter = getattr(os, "process_cpu_count", None)
    if getter is not None:
        count = getter()
        if count:
            _dbg("OCT-TESTING", func, f"process_cpu_count={count}", 4)
            return int(count)
    affinity = getattr(os, "sched_getaffinity", None)
    if affinity is not None:
        try:
            count = len(affinity(0))
        except OSError as exc:  # pragma: no cover - platform specific
            _dbg("OCT-TESTING", func, f"affinity unavailable: {exc}", 3)
        else:
            if count:
                _dbg("OCT-TESTING", func, f"affinity={count}", 4)
                return count
    return os.cpu_count() or 1
 
 
def resolve_worker_count(
    workspace_root: Path,
    package: str,
    *,
    cpu_count: int | None = None,
) -> int:
    """Resolve ``auto`` to ``min(max_workers, available_cpu_count())``.
 
    P1 for the cap, ADR-T-11's v0.4 amendment for the CPU count (P5).
    """
    func = "resolve_worker_count"
    cap = resolve_max_workers(workspace_root, package)
    cores = cpu_count if cpu_count is not None else available_cpu_count()
    count = max(1, min(cap, max(1, cores)))
    _dbg(
        "OCT-TESTING", func,
        f"cap={cap} cores={cores} -> {count} worker(s)", 3,
    )
    return count
 
 
def free_group_name(nodeid: str, *, whole_file: bool = False) -> str:
    """The per-test group name a ``free`` item carries (Plan 286 P1).
 
    A free file shares no resource with anything, so its tests may run
    anywhere - including in different workers. The name is therefore
    the item's own nodeid, not the group ``free`` every free file used
    to share: ``--dist loadgroup`` keeps ONE group on ONE worker, so
    the shared name pinned the whole free set to a single worker
    (measured on the oc_chat suite from this worktree, BASE
    ``3297fae7``: of twenty workers, gw0 ran all 3877 tests of all 283
    free files and only three other workers ran anything at all).
 
    *whole_file* is the P2 opt-out: the name is then the FILE's path,
    so that one file's tests stay together while every other free file
    still spreads.
 
    Parameters
    ----------
    nodeid
        The collected item's nodeid, ``<relpath>::<test>``.
    whole_file
        True when the item carries :data:`WHOLE_FILE_MARKER`.
    """
    stem = nodeid.split("::")[0] if whole_file else nodeid
    return f"{FREE_GROUP_PREFIX}{stem}"
 
 
def apply_group_marks(
    config: Any,
    items: Iterable[Any],
    manifest_path: Path,
) -> tuple[tuple[str, ...], str | None]:
    """Stamp every collected item with its scheduling group (P4/P5).
 
    Call from a root conftest's ``pytest_collection_modifyitems``
    under ``@pytest.hookimpl(tryfirst=True)``: xdist's worker-side
    hook is what turns the mark into the ``@<group>`` nodeid suffix
    its scheduler reads, and it runs FIRST otherwise, so a later mark
    is collected and then ignored.
 
    A file in a real conflict group, and a file the manifest does not
    classify, are stamped with that group's ONE name, exactly as
    before: both mean "these tests must not run beside each other".
    A ``free`` file's items each carry their own name through
    :func:`free_group_name` (Plan 286 P1), because a free file shares
    nothing and one name is one worker.
 
    Parameters
    ----------
    config
        The pytest ``Config``.
    items
        The collected items, modified in place.
    manifest_path
        The committed ``parallel_manifest.json``.
 
    Returns
    -------
    tuple
        ``(missing, manifest_error)``: the collected files with no
        manifest entry, and the load failure when the manifest itself
        could not be read. Both are fail-closed states - every
        affected item is stamped :data:`GROUP_UNCLASSIFIED`, so it
        runs serially with respect to the other unclassified files
        and a completeness check can fail the run by name.
    """
    func = "apply_group_marks"
    import pytest  # local: an optional dependency of oct
 
    collected = list(items)
    groups: dict[str, str] = {}
    missing: tuple[str, ...] = ()
    error: str | None = None
    try:
        manifest = load_manifest(Path(manifest_path))
    except OctTestError as exc:
        error = f"{Path(manifest_path).name}: {exc}"
        _dbg("OCT-TESTING", func, f"manifest unusable: {error}", 2)
    else:
        groups, missing = resolve_groups(
            manifest,
            {item.nodeid.split("::")[0] for item in collected},
        )
    if config.pluginmanager.hasplugin("xdist"):
        for item in collected:
            group = groups.get(
                item.nodeid.split("::")[0], GROUP_UNCLASSIFIED,
            )
            if group == GROUP_FREE:
                group = free_group_name(
                    item.nodeid,
                    whole_file=item.get_closest_marker(
                        WHOLE_FILE_MARKER,
                    ) is not None,
                )
            item.add_marker(pytest.mark.xdist_group(name=group))
    return missing, error
 
 
def is_xdist_worker(config: Any) -> bool:
    """True when *config* belongs to an xdist worker process.
 
    The guard that stops a worker re-injecting ``-n`` and spawning its
    own workers: xdist resets ``numprocesses`` to ``None`` in the
    worker's config, which otherwise looks exactly like "no ``-n`` was
    given" to :func:`configure_parallel_defaults`. The environment
    half also catches a NESTED pytest run launched from inside a
    worker, which inherits ``PYTEST_XDIST_WORKER`` and must stay
    serial.
    """
    return hasattr(config, "workerinput") or bool(
        os.environ.get("PYTEST_XDIST_WORKER"),
    )
 
 
def _adopt_group_dist_in_worker(config: Any, option: Any, dist: str) -> None:
    """Re-assert ``loadgroup`` inside an xdist worker process.
 
    xdist ships the CONTROLLER's ORIGINAL command line to each worker
    (``workermanage.WorkerController.setup`` sends
    ``config.invocation_params.args``), so an in-memory ``--dist``
    chosen by :func:`configure_parallel_defaults` is invisible there.
    The worker's ``remote.setup_config`` derives its private
    ``option.loadgroup`` flag from the dist it can see and then forces
    ``dist='no'``; that flag is what makes the worker append the
    ``@<group>`` nodeid suffix the ``loadgroup`` scheduler reads, so
    without this the marks are collected and then ignored, and a group
    silently spreads across workers (measured, 2026-09-06).
 
    An explicit ``--dist`` on the command line is left alone: there
    the worker already computed the flag from the operator's choice.
    """
    func = "_adopt_group_dist_in_worker"
    args = [str(arg) for arg in (config.invocation_params.args or ())]
    if any(arg.startswith("--dist") for arg in args):
        return
    option.loadgroup = dist == "loadgroup"
    _dbg("OCT-TESTING", func, f"worker loadgroup={option.loadgroup}", 3)
 
 
def configure_parallel_defaults(
    config: Any,
    *,
    workspace_root: Path,
    package: str,
    dist: str = "loadgroup",
) -> int | None:
    """Turn parallelism on by default for a run (P5).
 
    Call this from a root conftest's ``pytest_cmdline_main`` under
    ``@pytest.hookimpl(tryfirst=True)``: xdist's own ``tryfirst`` hook
    is what turns ``numprocesses`` into ``tx`` endpoints, so a later
    assignment engages nothing.
 
    Returns
    -------
    int | None
        The resolved worker count, or ``None`` when the run was left
        alone: an xdist worker process, a nested run inside one,
        ``-p no:xdist``, or an explicit ``-n`` on the command line.
    """
    func = "configure_parallel_defaults"
    option = getattr(config, "option", None)
    if option is None or not hasattr(option, "numprocesses"):
        _dbg("OCT-TESTING", func, "xdist absent; serial run", 2)
        return None
    if hasattr(config, "workerinput"):
        _adopt_group_dist_in_worker(config, option, dist)
        return None
    if is_xdist_worker(config):
        _dbg("OCT-TESTING", func, "nested run inside a worker; serial", 2)
        return None
 
    resolved: int | None = None
    if option.numprocesses is None:
        resolved = resolve_worker_count(workspace_root, package)
        option.numprocesses = resolved
    if option.numprocesses and getattr(option, "dist", "no") == "no":
        option.dist = dist
    _dbg(
        "OCT-TESTING", func,
        f"numprocesses={option.numprocesses} dist={option.dist}", 2,
    )
    return resolved
 
 
# ---------------------------------------------------------------------
# P3b - the dynamic detector (``--oct-trace``)
# ---------------------------------------------------------------------
 
#: Command-line flag that switches the dynamic detector on.
TRACE_OPTION: str = "--oct-trace"
 
#: Path fragments whose writes are never a shared-resource use.
_TRACE_IGNORED: tuple[str, ...] = (
    "__pycache__", "site-packages", ".pytest_cache", "/.git/",
    ".hypothesis", ".egg-info",
)
 
 
def _from_pytest_internals(depth: int = 4) -> bool:
    """True when the calling code is pytest's own machinery.
 
    ``monkeypatch.setenv`` writes through ``os.environ`` like any
    other caller, but it records an undo, so it is not a leak and must
    not be recorded as one. Neither is pytest's own
    ``PYTEST_CURRENT_TEST`` bookkeeping - which reaches ``__delitem__``
    through ``MutableMapping.pop`` in ``<frozen _collections_abc>``,
    so the search skips FROZEN frames and judges the first real one.
    Without that skip every file in a suite reads as an environment
    writer (measured on all 324 oc_chat files, 2026-09-06).
    """
    frame: Any = sys._getframe(2)
    for _ in range(depth):
        if frame is None:
            return False
        filename = frame.f_code.co_filename.replace("\\", "/")
        if "/_pytest/" in filename:
            return True
        if not filename.startswith("<"):
            return False
        frame = frame.f_back
    return False
 
 
def _is_pid_stamped(basename: str, pid: str) -> bool:
    """True when *basename* carries *pid* as a whole token.
 
    A pid-stamped artefact - oc_diagnostics writes
    ``logs/unified_debug_logs_<stamp>_p<pid>.txt`` - is per-process by
    construction, so it can never collide across workers, and it is
    written lazily by whichever test happens to initialise diagnostics
    first, so attributing it would make the classification depend on
    run order.
 
    The match is on WHOLE tokens, never a substring. A bare
    ``str(pid) in basename`` silently swallowed a GENUINE
    ``main_tree_write``: with pid 648, the artefact filename
    ``..._1788651058142648000_6d28ee.txt`` (a nanosecond stamp) matched
    by coincidence, which dropped
    ``tests/unit/test_203_event_payload_completeness.py`` out of its
    conflict group and made the classification differ between two runs
    of the same tree (gate finding R1, measured 2026-09-06).
    """
    tokens = re.split(r"[^0-9A-Za-z]+", basename)
    return any(token in (pid, f"p{pid}") for token in tokens)
 
 
class _Tracer:
    """Wrap the process-global seams and attribute them to a file.
 
    Only WRITE-shaped events are recorded - a read cannot collide.
    Every observation is attributed to the test file that was running
    when it happened, so the trace shares the static pass's key space.
 
    Every wrapper is ``functools.wraps``-ed onto the seam it replaces,
    so a wrapped function still reports the original's ``__name__``,
    ``__qualname__``, ``__doc__`` and ``__wrapped__``: code under test
    that inspects the identity of a patched binding must see what it
    would see untraced (gate finding R2).
    """
 
    def __init__(self, root: Path) -> None:
        self.root = Path(root).resolve()
        self.records: dict[str, set[str]] = {}
        self.env_writers: set[str] = set()
        self.spawners: set[str] = set()
        self.current: str | None = None
        self._installed: list[tuple[Any, str, Any, bool]] = []
        self._temp = os.path.realpath(
            os.environ.get("TEMP") or os.environ.get("TMPDIR") or "/tmp",
        ).replace("\\", "/").lower()
        self._pid = str(os.getpid())
 
    # -- attribution ------------------------------------------------
    def record(self, resource: str) -> None:
        """Attribute *resource* to the file under test."""
        if self.current is not None:
            self.records.setdefault(self.current, set()).add(resource)
 
    def record_path(self, target: Any) -> None:
        """Attribute a write to *target* to the file under test."""
        if self.current is None:
            return
        try:
            text = os.path.abspath(os.fspath(target))
        except (TypeError, ValueError):
            return
        normalised = text.replace("\\", "/")
        if any(part in normalised for part in _TRACE_IGNORED):
            return
        if _is_pid_stamped(normalised.rsplit("/", 1)[-1], self._pid):
            return
        try:
            # A write INSIDE the tree under test is a shared write by
            # definition, checked BEFORE the temp filters: a package
            # can legitimately be rooted inside a temp directory (the
            # W2 fixture tree is), and the tree under test always
            # wins over "it looks temporary".
            hint = Path(text).relative_to(self.root).as_posix()
        except ValueError:
            if normalised.lower().startswith(self._temp):
                return
            if "/pytest-of-" in normalised:
                return
            # Above the root: the real directory name is machine
            # specific and the manifest is committed, so the key is
            # the constant sentinel (P2).
            self.record(write_resource(normalised, OUTSIDE_KEY))
            return
        self.record(write_resource(hint, path_key(hint)))
 
    def record_launch(self, argv: Any, *, cwd: Any = None) -> None:
        """Attribute a child launch to the file under test (P3).
 
        The tracer sees the REAL argv, so it keys what the static pass
        can only guess at; an argv whose family the committed table
        does not vouch for still contributes the bare wildcard.
        """
        if self.current is None:
            return
        self.spawners.add(self.current)
        if launch_family(argv) == "git" and cwd is not None:
            if not self._git_is_temporary(argv, cwd):
                self.record(RES_ENV_SUBPROCESS)
                return
        for resource in launch_resources(argv):
            self.record(resource)
 
    def _git_is_temporary(self, argv: Sequence[Any], cwd: Any) -> bool:
        """Vouch for Git only when cwd and every -C remain temporary.
 
        Global configuration and explicit repository overrides can target
        locations outside cwd and therefore keep the launch unattributed.
        """
        try:
            target = Path(cwd).resolve()
            targets = [target]
            index = 1
            while index < len(argv):
                arg = os.fspath(argv[index])
                if arg in ("--global", "--system") or arg.startswith((
                    "--git-dir", "--work-tree", "--exec-path",
                )):
                    return False
                if arg == "-C":
                    index += 1
                    if index >= len(argv):
                        return False
                    target = (target / os.fspath(argv[index])).resolve()
                    targets.append(target)
                elif arg.startswith("-C"):
                    target = (target / arg[2:]).resolve()
                    targets.append(target)
                index += 1
            temporary = Path(self._temp)
            return all(
                not path.is_relative_to(self.root) and (
                    path.is_relative_to(temporary)
                    or "/pytest-of-" in path.as_posix()
                ) for path in targets
            )
        except (TypeError, ValueError, OSError):
            return False
 
    def resources(self) -> dict[str, set[str]]:
        """The trace as recorded.
 
        The P2 env/subprocess join happens at the seam now:
        :meth:`record_launch` records the family's own tokens, or the
        bare one, at the moment of the launch, and the environment
        seam records :data:`RES_ENV_MUTATION` (Plan 276 P1) - which is
        process-local, so it never groups anything. :attr:`spawners`
        and :attr:`env_writers` stay as the diagnostic record of WHICH
        files launched a child and WHICH wrote the environment.
        """
        return {path: set(res) for path, res in self.records.items()}
 
    # -- install / remove -------------------------------------------
    def _patch(self, owner: Any, name: str, factory: Callable) -> None:
        """Replace ``owner.name`` with ``factory(original)``."""
        original = getattr(owner, name)
        owned = name in getattr(owner, "__dict__", {})
        self._installed.append((owner, name, original, owned))
        setattr(owner, name, factory(original))
 
    def install(self) -> None:
        """Wrap every traced seam."""
        func = "install"
        self._patch(builtins, "open", self._open_wrapper)
        self._patch(io, "open", self._open_wrapper)
        self._patch(os, "chdir", self._chdir_wrapper)
        self._patch(type(os.environ), "__setitem__", self._env_wrapper)
        self._patch(type(os.environ), "__delitem__", self._env_wrapper)
        self._patch(socket.socket, "bind", self._bind_wrapper)
        self._patch(subprocess.Popen, "__init__", self._popen_wrapper)
        _dbg("OCT-TESTING", func, f"{len(self._installed)} seams wrapped", 2)
 
    def remove(self) -> None:
        """Restore every wrapped seam, newest first."""
        func = "remove"
        for owner, name, original, owned in reversed(self._installed):
            if owned:
                setattr(owner, name, original)
            else:
                delattr(owner, name)
        self._installed.clear()
        _dbg("OCT-TESTING", func, "seams restored", 2)
 
    # -- wrappers ---------------------------------------------------
    def _open_wrapper(self, original: Callable) -> Callable:
        @functools.wraps(original)
        def wrapper(
            file: Any = None, mode: str = "r", *args: Any, **kwargs: Any,
        ) -> Any:
            if any(flag in mode for flag in ("w", "a", "x", "+")):
                self.record_path(file)
            return original(file, mode, *args, **kwargs)
        return wrapper
 
    def _chdir_wrapper(self, original: Callable) -> Callable:
        @functools.wraps(original)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            self.record(RES_CWD_CHANGE)
            return original(*args, **kwargs)
        return wrapper
 
    def _env_wrapper(self, original: Callable) -> Callable:
        @functools.wraps(original)
        def wrapper(mapping: Any, *args: Any, **kwargs: Any) -> Any:
            if self.current is not None and not _from_pytest_internals():
                self.env_writers.add(self.current)
                self.record(RES_ENV_MUTATION)
            return original(mapping, *args, **kwargs)
        return wrapper
 
    def _bind_wrapper(self, original: Callable) -> Callable:
        @functools.wraps(original)
        def wrapper(
            sock: Any, address: Any, *args: Any, **kwargs: Any,
        ) -> Any:
            port = address[1] if isinstance(address, tuple) and len(
                address,
            ) > 1 else None
            if isinstance(port, int) and port > 0:
                self.record(f"{PORT_PREFIX}{port}")
            return original(sock, address, *args, **kwargs)
        return wrapper
 
    def _popen_wrapper(self, original: Callable) -> Callable:
        @functools.wraps(original)
        def wrapper(popen: Any, *args: Any, **kwargs: Any) -> Any:
            if self.current is not None:
                self.record_launch(
                    args[0] if args else kwargs.get("args"),
                    cwd=(kwargs.get("cwd", args[9] if len(args) > 9 else None)
                         or os.getcwd()),
                )
            return original(popen, *args, **kwargs)
        return wrapper
 
 
#: The active tracer, or None. Module state, because the pytest hooks
#: below are plain functions (no ``pytest`` import at module scope).
_TRACER: _Tracer | None = None
 
#: Where the active tracer writes its per-process trace.
_TRACE_PATH: Path | None = None
 
 
#: Refusal text when ``--oct-trace`` meets a parallel run (R3).
TRACE_REQUIRES_SERIAL: str = (
    "--oct-trace requires a SERIAL run: re-run with -n 0. The tracer "
    "wraps process-global seams, so under xdist it would record only "
    "this process's activity and write a partial trace, and a partial "
    "trace silently misclassifies the files it never saw."
)
 
 
def pytest_addoption(parser: Any) -> None:
    """Register ``--oct-trace`` (pytest plugin hook, P3b)."""
    parser.addoption(
        TRACE_OPTION,
        action="store",
        default=None,
        metavar="PATH",
        help=(
            "record the shared resources each test file actually "
            "touches and write them to PATH; requires a serial run "
            "(-n 0), and refuses under xdist workers"
        ),
    )
 
 
def pytest_configure(config: Any) -> None:
    """Install the tracer when ``--oct-trace`` was given.
 
    Refuses a PARALLEL traced run outright (R3). Measured on the
    oc_chat suite 2026-09-06: with workers, only the controller wrote
    a trace file, so the classification would have been rebuilt from
    collection-time activity alone. Refusing at configure time, before
    any worker starts, is the fail-closed choice; a partial trace that
    looks complete is the failure mode this whole mechanism exists to
    prevent.
    """
    func = "pytest_configure"
    global _TRACER, _TRACE_PATH
    target = config.getoption(TRACE_OPTION, default=None)
    if not target:
        return
    workers = getattr(config.option, "numprocesses", None) or 0
    if workers and not hasattr(config, "workerinput"):
        import pytest  # local: an optional dependency of oct
 
        _dbg("OCT-TESTING", func, f"refusing: {workers} worker(s)", 2)
        raise pytest.UsageError(TRACE_REQUIRES_SERIAL)
    _TRACE_PATH = Path(target)
    _TRACER = _Tracer(Path(str(config.rootpath)))
    _TRACER.install()
    _dbg("OCT-TESTING", func, f"tracing to {_TRACE_PATH}", 2)
 
 
def pytest_runtest_logstart(nodeid: str, location: Any) -> None:
    """Attribute the next observations to *nodeid*'s file."""
    if _TRACER is not None:
        _TRACER.current = nodeid.split("::")[0]
 
 
def pytest_runtest_logfinish(nodeid: str, location: Any) -> None:
    """Stop attributing observations to a test."""
    if _TRACER is not None:
        _TRACER.current = None
 
 
def pytest_sessionfinish(session: Any, exitstatus: Any) -> None:
    """Write this process's trace and restore the wrapped seams."""
    func = "pytest_sessionfinish"
    global _TRACER
    if _TRACER is None or _TRACE_PATH is None:
        return
    tracer, _TRACER = _TRACER, None
    tracer.remove()
    document = {
        "schema_version": SCHEMA_VERSION,
        "files": {
            path: sorted(res)
            for path, res in sorted(tracer.resources().items())
        },
        # Informational, not part of the classification: the files
        # that wrote to os.environ OUTSIDE monkeypatch, i.e. the P6
        # leak candidates. An env write is process-local and never a
        # cross-worker conflict on its own.
        "env_writers": sorted(tracer.env_writers),
    }
    path = _TRACE_PATH.with_name(
        f"{_TRACE_PATH.stem}.{os.getpid()}{_TRACE_PATH.suffix or '.json'}",
    )
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(document, indent=2) + "\n", encoding="utf-8", newline="\n",
    )
    _dbg("OCT-TESTING", func, f"trace written to {path}", 2)
 
 
def load_traces(sources: Sequence[Path]) -> dict[str, set[str]]:
    """Merge trace documents from files and/or directories."""
    func = "load_traces"
    merged: dict[str, set[str]] = {}
    for source in sources:
        paths = sorted(source.glob("*.json")) if source.is_dir() else [source]
        for path in paths:
            raw = json.loads(path.read_text(encoding="utf-8"))
            for key, values in raw.get("files", {}).items():
                merged.setdefault(key, set()).update(values)
    _dbg("OCT-TESTING", func, f"merged {len(merged)} traced file(s)", 2)
    return merged
 
 
# ---------------------------------------------------------------------
# CLI - ``python -m oct.testing.parallel classify --root <path>``
# ---------------------------------------------------------------------
 
def _package_for_root(root: Path) -> tuple[Path, str, tuple[str, ...]]:
    """Resolve *root* to its workspace, package name and test roots."""
    from oct.core.workspace import find_workspace_root
 
    workspace_root = find_workspace_root(root)
    if workspace_root is None:
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"no workspace manifest above {root}",
        )
    manifest = load_test_manifest(workspace_root)
    for package in manifest.packages:
        if package.path.resolve() == root.resolve():
            return workspace_root, package.name, package.test_roots
    raise OctTestError(
        MANIFEST_TEST_BLOCK_INVALID,
        f"{root} is not a declared subproject of {workspace_root}",
    )
 
 
def default_manifest_path(root: Path, test_roots: Sequence[str]) -> Path:
    """The manifest path implied by *test_roots* (their common parent)."""
    parts = [Path(entry).parts for entry in test_roots]
    common: list[str] = []
    for index in range(min((len(part) for part in parts), default=0)):
        column = {part[index] for part in parts}
        if len(column) != 1:
            break
        common.append(column.pop())
    return root.joinpath(*common) / MANIFEST_FILENAME
 
 
def _parse_pins(values: Sequence[str]) -> dict[str, str]:
    """Parse repeated ``--pin <path>=<group>`` arguments."""
    pins: dict[str, str] = {}
    for value in values:
        path, separator, group = value.partition("=")
        if not separator or not path or not group:
            raise OctTestError(
                MANIFEST_TEST_BLOCK_INVALID,
                f"--pin expects <path>=<group>, got {value!r}",
            )
        pins[path] = group
    return pins
 
 
def _tool_version() -> str:
    """The oct version string recorded in the manifest."""
    try:
        from oct import __version__
    except ImportError:  # pragma: no cover - oct is always importable
        return "oct unknown"
    return f"oct {__version__}"
 
 
def _print_summary(
    document: Mapping[str, Any],
    out: Path,
    workspace_root: Path,
    package: str,
) -> None:
    """Print an ASCII-only summary of a regeneration."""
    files = document["files"]
    counts: dict[str, int] = {}
    groups: dict[str, int] = {}
    for entry in files.values():
        groups[entry["group"]] = groups.get(entry["group"], 0) + 1
        for resource in entry["resources"]:
            counts[resource] = counts.get(resource, 0) + 1
    print(f"manifest: {out}")
    print(f"files: {len(files)}  groups: {len(groups)}")
    print(f"workers: {resolve_worker_count(workspace_root, package)}")
    for resource, count in sorted(counts.items(), key=lambda kv: -kv[1]):
        print(f"  resource {resource}: {count} file(s)")
    for name, count in sorted(groups.items(), key=lambda kv: -kv[1])[:8]:
        print(f"  group {name}: {count} file(s)")
 
 
#: Exit code of ``classify --check`` when the committed manifest and a
#: fresh classification disagree (§12.7 configuration class).
EXIT_MANIFEST_DRIFT: int = 4
 
 
def diff_manifests(
    committed: Mapping[str, Any],
    fresh: Mapping[str, Any],
) -> list[str]:
    """Compare two manifest documents by CONTENT, file by file.
 
    Only ``files`` is compared: ``generated_at`` and ``tool_version``
    differ on every regeneration by design.
 
    Returns
    -------
    list[str]
        One line per disagreeing file, each naming the file and the
        reason. Empty when the two documents classify identically.
    """
    func = "diff_manifests"
    left = committed.get("files", {}) or {}
    right = fresh.get("files", {}) or {}
    lines: list[str] = []
    for path in sorted(set(left) | set(right)):
        if path not in left:
            lines.append(f"{path}: absent from the committed manifest")
            continue
        if path not in right:
            lines.append(f"{path}: in the manifest but no longer collected")
            continue
        old, new = left[path], right[path]
        old_res = sorted(old.get("resources", []))
        new_res = sorted(new.get("resources", []))
        if old_res != new_res:
            lines.append(
                f"{path}: resources {old_res} -> {new_res}",
            )
        elif old.get("group") != new.get("group"):
            lines.append(
                f"{path}: group {old.get('group')!r} -> "
                f"{new.get('group')!r}",
            )
    _dbg("OCT-TESTING", func, f"{len(lines)} drifted file(s)", 2)
    return lines
 
 
def run_classify(argv: Sequence[str]) -> int:
    """Regenerate (or CHECK) a subproject's parallel manifest.
 
    ``--check`` classifies in memory from the same inputs and diffs the
    result against the committed manifest instead of overwriting it:
    exit 0 on identity, :data:`EXIT_MANIFEST_DRIFT` listing every file
    whose resources or group differ. The completeness check inside the
    suite only proves a file is PRESENT; this is what proves the
    entries are still TRUE (gate finding R1).
 
    A trace passed with ``--trace`` must come from a SERIAL traced run
    (``pytest -n 0 -p oct.testing.parallel --oct-trace=<path>``); the
    plugin refuses to trace a parallel run, because a partial trace
    would quietly reclassify every file it did not observe.
    """
    func = "run_classify"
    import argparse
 
    parser = argparse.ArgumentParser(
        prog="python -m oct.testing.parallel classify",
        description="classify test files by shared resource and group them",
    )
    parser.add_argument("--root", required=True, type=Path)
    parser.add_argument("--out", type=Path, default=None)
    parser.add_argument("--trace", type=Path, action="append", default=[])
    parser.add_argument("--pin", action="append", default=[])
    parser.add_argument(
        "--check",
        action="store_true",
        help=(
            "do not write: diff a fresh classification against the "
            "committed manifest and exit non-zero on any difference"
        ),
    )
    args = parser.parse_args(list(argv))
 
    root = args.root.resolve()
    workspace_root, package, test_roots = _package_for_root(root)
    _dbg("OCT-TESTING", func, f"classifying {package} at {root}", 2)
 
    classified: dict[str, set[str]] = {
        path: set(resources)
        for path, resources in classify_tree(root, test_roots).items()
    }
    traced = load_traces(args.trace)
    for path, resources in traced.items():
        if path in classified:
            classified[path] |= resources
    document = build_manifest(
        classified,
        tool_version=_tool_version(),
        pins=_parse_pins(args.pin),
    )
    out = args.out or default_manifest_path(root, test_roots)
 
    if args.check:
        return _run_check(out, document)
    write_manifest(out, document)
    _print_summary(document, out, workspace_root, package)
    return 0
 
 
def _run_check(out: Path, document: Mapping[str, Any]) -> int:
    """The ``--check`` half of :func:`run_classify`."""
    func = "_run_check"
    try:
        committed = json.loads(Path(out).read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        print(f"drift: {out} is unreadable: {exc}")
        return EXIT_MANIFEST_DRIFT
    lines = diff_manifests(committed, document)
    if not lines:
        print(f"manifest: {out}")
        print(f"check: OK, {len(document['files'])} file(s) identical")
        return 0
    print(f"manifest: {out}")
    print(f"check: DRIFT, {len(lines)} file(s) differ")
    for line in lines:
        print(f"  {line}")
    print("remedy: regenerate with the same --trace and --pin arguments")
    _dbg("OCT-TESTING", func, f"drift on {len(lines)} file(s)", 2)
    return EXIT_MANIFEST_DRIFT
 
 
def main(argv: Sequence[str] | None = None) -> int:
    """CLI entry point; ``classify`` is the only subcommand."""
    args = list(sys.argv[1:] if argv is None else argv)
    if not args or args[0] != "classify":
        print("usage: python -m oct.testing.parallel classify --root PATH")
        return 4
    try:
        return run_classify(args[1:])
    except OctTestError as exc:
        print(f"error {exc.code}: {exc.message}")
        return exc.exit_code
 
 
__all__ = [
    "DEFAULT_MAX_WORKERS",
    "EXIT_MANIFEST_DRIFT",
    "FREE_GROUP_PREFIX",
    "FileEntry",
    "GROUP_FREE",
    "GROUP_UNCLASSIFIED",
    "KEYED_FAMILIES",
    "KEY_SEPARATOR",
    "LAUNCH_FAMILIES",
    "MANIFEST_FILENAME",
    "ModuleAnalysis",
    "OUTSIDE_KEY",
    "PORT_PREFIX",
    "PROCESS_LOCAL_RESOURCES",
    "PYTHON_ARGV0",
    "ParallelManifest",
    "RESOURCE_VOCABULARY",
    "RES_COMMS_ROOT",
    "RES_CWD_CHANGE",
    "RES_ENV_SUBPROCESS",
    "RES_GLOBAL_SINGLETON_RESET",
    "RES_MAIN_TREE_WRITE",
    "RES_NETWORK",
    "RES_OCTRC",
    "RES_REGISTRY_SQLITE",
    "RES_SETTINGS_FILE",
    "RES_WORKTREE_DIR",
    "SCHEMA_VERSION",
    "TRACE_OPTION",
    "TRACE_REQUIRES_SERIAL",
    "UNKNOWN_PREFIX",
    "WHOLE_FILE_MARKER",
    "analyse_source",
    "apply_group_marks",
    "available_cpu_count",
    "build_groups",
    "build_manifest",
    "classify_source",
    "classify_tree",
    "configure_parallel_defaults",
    "conflicts_with",
    "default_manifest_path",
    "diff_manifests",
    "free_group_name",
    "is_conflicting",
    "is_known_resource",
    "is_xdist_worker",
    "keyed",
    "launch_family",
    "launch_resources",
    "load_manifest",
    "load_traces",
    "main",
    "path_key",
    "resolve_groups",
    "resolve_max_workers",
    "resolve_worker_count",
    "resource_family",
    "resource_for_path_hint",
    "resource_key",
    "unknown",
    "write_manifest",
    "write_resource",
]
 
 
if __name__ == "__main__":
    sys.exit(main())