Agent-tool runs

def current_state(run_dir: Path) -> State:
    """Return the run's state, rebuilt from the log when ``state.json`` does not reflect all of it.

    The caller holds the lock.

    Args:
        run_dir: The run directory.

    Returns:
        The state after every logged call.

    Raises:
        RunError: If the run directory is malformed or incomplete.
    """
    manifest = read_manifest(run_dir)
    return _load(run_dir, manifest, read_calls(run_dir))[0]

def get_state(run_dir: Path) -> dict[str, Any]:
    """Return what a run holds now, for an agent deciding its next call.

    Args:
        run_dir: The run directory.

    Returns:
        Every node's id, type and statement, with its assessment where it has one; the relations; the
        refused records so far; how many calls the log holds; and whether the run is finished.

    Raises:
        RunError: If the run directory is malformed or incomplete.
    """
    with lock(run_dir):
        manifest = read_manifest(run_dir)
        all_calls = read_calls(run_dir)
        state = _reconcile(run_dir, manifest, _rubric(run_dir, manifest), all_calls)
        calls = len(all_calls)
    nodes = []
    for node in state.graph:
        item: dict[str, Any] = {"id": node.id, "type": node.kind, "text": node.statement}
        if "assessment" in node.attributes:
            item["assessment"] = node.attributes["assessment"]
        if "check" in node.attributes:
            item["result"] = node.attributes["check"].get("result")
        nodes.append(item)
    relations = [
        {"type": relation.type, "from": relation.source, "to": relation.target}
        for relation in state.graph.relations.values()
    ]
    return {
        "nodes": nodes,
        "relations": relations,
        "rejections": [dict(item) for item in state.rejections],
        "calls": calls,
        "finished": state.finished,
    }

@contextmanager
def lock(run_dir: Path, timeout: float = LOCK_TIMEOUT) -> Iterator[None]:
    """Hold the run's lock: one tool call at a time, across processes.

    The operating system releases the lock when its process exits, so a crashed agent leaves none
    behind. A call that cannot take the lock within ``timeout`` seconds reports the run busy.

    Args:
        run_dir: The run directory.
        timeout: How long to wait, in seconds.

    Yields:
        Nothing; the lock is held until the block ends.

    Raises:
        RunError: If the lock is not free within ``timeout``.
    """
    with (run_dir / LOCK).open("a+b") as handle:
        if handle.seek(0, os.SEEK_END) == 0:
            handle.write(b"\0")
            handle.flush()
        deadline = time.monotonic() + timeout
        while not _try_lock(handle):
            if time.monotonic() >= deadline:
                holder = _owner(run_dir)
                by = (
                    f" (process {holder.get('pid')}, since {holder.get('since')})" if holder and "pid" in holder else ""
                )
                raise RunError("run-busy", f"another tool call holds the lock of {run_dir.name!r}{by}")
            time.sleep(0.05)
        owner = run_dir / LOCK_OWNER
        try:
            owner.write_text(json.dumps({"pid": os.getpid(), "since": _now()}), encoding="utf-8")
            yield
        finally:
            owner.unlink(missing_ok=True)
            _unlock(handle)

def lock_holder(run_dir: Path) -> dict[str, Any] | None:
    """Return who holds a run's lock, or ``None`` when no tool call does.

    The lock is the operating system's: it is released when its process exits, so a crashed agent
    leaves none behind, and only a live process (a hung one, say) can hold it. A free lock's leftover
    owner record, from a process that died holding it, is removed.

    Args:
        run_dir: The run directory.

    Returns:
        ``{"pid", "since"}`` of the holding process (either may be missing when it wrote no record yet),
        or ``None``.

    Raises:
        RunError: If the directory is not a run.
    """
    if not (run_dir / CALLS).is_file():
        raise RunError("missing-run", f"{run_dir.name!r} is not a run directory")
    with (run_dir / LOCK).open("a+b") as handle:
        if handle.seek(0, os.SEEK_END) == 0:
            handle.write(b"\0")
            handle.flush()
        if _try_lock(handle):
            # A record left by a process that died holding the lock; removed while the lock is held, so a
            # call that takes it next cannot have its own record removed.
            try:
                (run_dir / LOCK_OWNER).unlink(missing_ok=True)
            finally:
                _unlock(handle)
            return None
    return _owner(run_dir) or {}

def operation(name: str) -> Callable[[Operation], Operation]:
    """Register a write tool's operation under its name.

    Args:
        name: The tool's name, as the call log records it.

    Returns:
        A decorator registering the operation.
    """

    def register(fn: Operation) -> Operation:
        OPERATIONS[name] = fn
        return fn

    return register

def read_calls(run_dir: Path) -> list[dict[str, Any]]:
    """Read the call log.

    Args:
        run_dir: The run directory.

    Returns:
        The calls, in the order they took effect.

    Raises:
        RunError: If a line is not a call object, or the calls are not numbered 1, 2, … from ``start_run``.
    """
    calls = []
    for number, line in enumerate(_read(run_dir, CALLS).splitlines(), start=1):
        try:
            call = json.loads(line)
        except ValueError as error:
            raise RunError("invalid-record", f"{CALLS} line {number} is not JSON: {error}") from None
        if not isinstance(call, dict) or call.get("seq") != number or not isinstance(call.get("tool"), str):
            raise RunError("invalid-record", f"{CALLS} line {number} is not call {number}")
        calls.append(call)
    if not calls or calls[0]["tool"] != START:
        raise RunError("invalid-record", f"{CALLS} does not begin with {START}")
    return calls

def read_manifest(run_dir: Path) -> dict[str, Any]:
    """Read a format-2 run's manifest.

    Args:
        run_dir: The run directory.

    Returns:
        The manifest.

    Raises:
        RunError: If the directory is not a run of this format and version.
    """
    if not (run_dir / MANIFEST).is_file():
        raise RunError("not-a-run", f"{run_dir.name!r} has no {MANIFEST}")
    try:
        manifest = json.loads(_read(run_dir, MANIFEST))
    except ValueError as error:
        raise RunError("invalid-manifest", f"{MANIFEST} is not JSON: {error}") from None
    if not isinstance(manifest, dict) or manifest.get("format") != FORMAT:
        raise RunError("not-a-run", f"{MANIFEST} is not a {FORMAT} manifest")
    if manifest.get("version") != VERSION:
        raise RunError("unsupported-version", f"{MANIFEST} is version {manifest.get('version')!r}, not {VERSION}")
    return manifest

def reconcile(run_dir: Path) -> None:
    """Bring a run's state, manifest and outputs back in line with its call log, under the lock.

    Args:
        run_dir: The run directory.

    Raises:
        RunError: If the run directory is malformed or incomplete.
    """
    with lock(run_dir):
        manifest = read_manifest(run_dir)
        _reconcile(run_dir, manifest, _rubric(run_dir, manifest), read_calls(run_dir))

def recorded_graph(run_dir: Path) -> bytes:
    """Return the run's graph after every logged call, as ``replay`` returns it.

    Args:
        run_dir: The run directory.

    Returns:
        The graph, as the core serialises it, with a final newline.
    """
    with lock(run_dir):
        state = current_state(run_dir)
    return _graph_text(state.graph)

def replay(run_dir: Path, *, source: Path | None = None, sources: Path | None = None) -> bytes:
    """Re-ingest a run's document and apply every logged call again, without the network.

    Each call's result must have the digest the log recorded; the graph the last call leaves is
    returned, as ``state.json`` holds it, for the caller to compare with the run's.

    Args:
        run_dir: The run directory.
        source: The document, for a run that did not store it.
        sources: The fetched sources the run did not store, as files named by their digest.

    Returns:
        The rebuilt graph, as the core serialises it, with a final newline.

    Raises:
        RunError: If the run directory is malformed or incomplete, the document, rubric or a cited
            copy differs from what the run recorded, or a call's result differs from its record.
    """
    manifest = read_manifest(run_dir)
    calls = read_calls(run_dir)
    start = calls[0]["args"]
    recorded = start["source"]
    if recorded.get("stored"):
        source = run_dir / SOURCE / recorded["file"]
    elif source is None:
        raise RunError("missing-source", "the run did not store its document: pass the source document")
    try:
        source_bytes = source.read_bytes()
    except OSError as error:
        raise RunError("missing-source", f"cannot read {source.name!r}: {error.strerror or error}") from None
    if bytes_digest(source_bytes) != recorded["digest"]:
        raise RunError("source-mismatch", f"{source.name!r} is not the document the run read")
    rubric = _rubric(run_dir, manifest)
    for name, digest in start["cited"].items():
        if bytes_digest(_read(run_dir, f"{CITED}/{name}")) != digest:
            raise RunError("record-mismatch", f"the cited copy {name!r} differs from the one start_run made")
    graph = _ingest(source, source_bytes, start["document"], rubric)
    graph = stamp(graph, Graph(), {"run": manifest["run"], "tool": START})
    state = State(graph)
    _check_digests(calls[0], state, _start_result(graph, rubric, manifest["run"]))
    for call in calls[1:]:
        context = _context(run_dir, manifest, rubric, calls, call, replaying=True, sources=sources)
        state, result = _apply(context, state, call)
        _check_digests(call, state, result)
    return _graph_text(state.graph)

def rubric_problems(rubric: Rubric) -> list[str]:
    """Return why a rubric cannot be used for an agent-tool run, or nothing if it can.

    Args:
        rubric: The rubric.

    Returns:
        Every problem: an edge rule that creates a relation other than ``requires``, and a
        ``[premises]`` table without the analyst ``assumption`` type it needs.
    """
    problems = []
    refused = edge_rules_not_requires(rubric)
    if refused:
        problems.append(
            f"the edge rules of {refused} create relations other than requires, which would let a document "
            "raise the credence of its own claims"
        )
    if rubric.premises is not None:
        problem = premises_rubric_problem(rubric)
        if problem:
            problems.append(f"{problem}, which its [premises] table needs")
    return problems

def stamp(graph: Graph, before: Graph, mark: Mapping[str, Any]) -> Graph:
    """Stamp every node of ``graph`` that is new or changed since ``before`` with ``attributes.extraction``.

    Args:
        graph: The graph after a change.
        before: The graph before it.
        mark: The stamp, such as ``{"run": <run id>, "tool": <tool>}``.

    Returns:
        The stamped graph.
    """
    stamped = Graph()
    for node in graph:
        changed = before.nodes.get(node.id) != node
        attributes = {**node.attributes, "extraction": dict(mark)}
        stamped.add_node(replace(node, attributes=attributes) if changed else node)
    for relation in graph.relations.values():
        stamped.add_relation(relation)
    return stamped

def start_run(  # noqa: PLR0913 - each argument is a separate choice of the caller
    source: Path,
    rubric: Path,
    run: Path,
    *,
    harness: str,
    model: str,
    agent: str | None = None,
    note: str | None = None,
    document: str | None = None,
    cited: Path | None = None,
    store_source: bool = False,
) -> dict[str, Any]:
    """Start a run: check the rubric, ingest the document, copy the cited works, and log the call.

    Everything is read and checked before the run directory is written, so a rubric the tools cannot
    use, an unreadable document or a cited directory holding a link leaves nothing behind.

    Args:
        source: The document's main ``.tex`` file.
        rubric: The rubric file.
        run: The run directory to create; it must not exist or be empty.
        harness: The agent harness, as it reports itself.
        model: The model making the call.
        agent: A name for the agent.
        note: Free text kept with the call.
        document: The document id of every anchor; the source's file name by default.
        cited: The directory of cited works, each a file named by its bibliography key; copied into
            the run with each file's digest, and the copy is the only one the citation tool reads.
        store_source: Copy the document into the run.

    Returns:
        The run's id, the chunks, the document types with their form fields and edge rules, the hedge
        phrases, the check types the tools run, and those the rubric declares but the tools do not.

    Raises:
        RunError: If the run directory is in use, the rubric cannot be used, the document cannot be
            read, or the cited works are not plain files.
    """
    identity = _identity(model, agent, note)
    if not isinstance(harness, str) or not harness.strip():
        raise RunError("invalid-argument", "harness must name the agent harness")
    if run.exists() and any(run.iterdir()):
        raise RunError("run-exists", f"{run.name!r} exists and is not empty")
    read = _read_start(source, rubric, cited)
    if document is not None and not isinstance(document, str):
        raise RunError("invalid-argument", "document must be an id for the document, a string")
    if document is not None and (
        not document.strip() or any(sep in document for sep in ("/", "\\")) or PureWindowsPath(document).drive
    ):
        raise RunError("invalid-argument", f"document {document!r} must be an id for the document, not a path")
    document = document or source.name
    run_id = new_run_id()
    graph = stamp(_ingest(source, read.source_bytes, document, read.rubric), Graph(), {"run": run_id, "tool": START})
    args = {
        "document": document,
        "source": {"file": source.name, "digest": bytes_digest(read.source_bytes), "stored": store_source},
        "rubric_digest": bytes_digest(read.rubric_bytes),
        "cited": {name: bytes_digest(data) for name, data in read.cited.items()},
        "harness": harness,
    }
    result = _start_result(graph, read.rubric, run_id)
    # Build the run in a temporary directory beside it and rename it into place, so a start that fails
    # part-way leaves nothing that a second start_run would refuse as a run in use.
    building = run.parent / f".{run.name}.{run_id}.starting"
    building.mkdir(parents=True)
    try:
        _populate(building, read, graph, result, args, identity, run_id, store_source, source.name, document)
        if run.exists():
            run.rmdir()
        building.rename(run)
    except BaseException:
        shutil.rmtree(building, ignore_errors=True)
        raise
    return result

def versions() -> dict[str, str]:
    """Return the versions a run depends on: Python, the core, this package and sympy.

    Returns:
        The versions by name.
    """
    return {
        "python": platform.python_version(),
        "credencegraph": version("credencegraph"),
        "credencegraph-extract": __version__,
        "sympy": version("sympy"),
    }

def write(  # noqa: PLR0913 - the call and who makes it
    run_dir: Path,
    tool: str,
    args: Mapping[str, Any],
    *,
    model: str,
    agent: str | None = None,
    note: str | None = None,
) -> dict[str, Any]:
    """Make one write to a run: apply a tool's operation under the lock and commit it.

    Args:
        run_dir: The run directory.
        tool: The tool, a registered operation.
        args: Its arguments, as JSON values.
        model: The model making the call, as the harness reports it.
        agent: A name for the agent, for runs with several.
        note: Free text kept with the call, never used in inference.

    Returns:
        The tool's result.

    Raises:
        RunError: If the tool is unknown, the run is busy or malformed, or the operation refuses the
            call as a whole.
    """
    identity = _identity(model, agent, note)
    if tool not in _operations():
        raise RunError("unknown-tool", f"{tool!r} is not a tool that changes a run")
    with lock(run_dir):
        manifest = read_manifest(run_dir)
        rubric = _rubric(run_dir, manifest)
        calls = read_calls(run_dir)
        state = _reconcile(run_dir, manifest, rubric, calls)
        seq = len(calls) + 1
        # The call is applied to its arguments as the log will hold them, so a replay, which reads them
        # back from the log, gives the same result whatever order the agent wrote their keys in.
        call = {"seq": seq, "tool": tool, "args": json.loads(_line(dict(args))), **identity}
        after, result = _apply(_context(run_dir, manifest, rubric, calls, call), state, call)
        line = _line({**call, "call_digest": _call_digest(call), **_digests(after, result), "at": _now()})
        log_digest = bytes_digest(_read(run_dir, CALLS) + line)
        temporary = run_dir / (STATE + ".tmp")
        _durable_write(temporary, _state_file(after, seq, log_digest))
        with (run_dir / CALLS).open("ab") as log:
            log.write(line)
            log.flush()
            os.fsync(log.fileno())
        temporary.replace(run_dir / STATE)
        _replace(run_dir / MANIFEST, _manifest(manifest, [*calls, json.loads(line)], after, log_digest))
        _check_files(run_dir, after)
        _outputs(run_dir, rubric, after)
    return result