cleveragents-core/tools/implementer_workspace.py

#!/usr/bin/env python3
"""Worker-side reader for the dispatcher's workspace handoff sentinel.

What this is
------------

The dispatcher pre-clones a PR's head_sha into
``/tmp/cleveragents-implementer-worktrees/pr-{N}-implementer-{tag}/``
and drops a sentinel JSON file alongside it at
``/tmp/cleveragents-implementer-worktrees/.handoff-pr-{N}.json``
describing the worktree. This script is what the implementer worker
calls (via the ``implementer-workspace`` skill) to discover that
worktree without re-cloning.

Why a script + skill instead of carrying the path in the prompt
---------------------------------------------------------------

Originally built to defeat OpenCode's per-``task``-hop prompt
summarisation in the pre-R3 wrapper chain
(``implementation-worker`` (depth 0) → ``tier-dispatcher`` (depth
1) → ``tier-0`` (depth 2) → ``task-implementor`` (depth 3)), where
by the time the chain reached ``task-implementor`` only
``BEGIN_PR_DIFF`` reliably survived — every other section was
dropped by intermediate agents' summarisation.

After the R3 wrapper-chain retirement (2026-05-17) the implementer
pool dispatches directly to ``task-implementor-tier-<slot>``
variants — no ``task`` hops between dispatcher and worker, so the
``## Pre-cloned working copy`` section now survives intact in the
prompt body. This script remains as the durable handshake: the
worker's contract still ALWAYS calls it (defensive against a future
regression that re-introduces summarisation), and the script is
deterministic / safe / fast (tens of ms).

Contract
--------

Subcommand ``discover``:

  ``python3 tools/implementer_workspace.py discover --pr <N>``

  Reads the sentinel at ``/tmp/cleveragents-implementer-worktrees/.handoff-pr-{N}.json``,
  validates it (schema version, file exists on disk, worktree dir
  still exists), and prints two lines on stdout:

      repo_dir=<absolute-path-to-worktree>
      branch=<branch-name-or-empty>

  If the sentinel is missing, stale, malformed, or points at a
  vanished worktree, the script prints

      repo_dir=
      branch=

  and exits with code 0. The worker calling this script checks the
  ``repo_dir=`` line; if empty, it falls through to its legacy
  workflow (``git-isolator-util`` for the implementer).

  This script NEVER exits non-zero on a "no handoff available" path
  — that's a normal condition, not a failure. Non-zero exit codes
  are reserved for usage errors (bad CLI args, schema-version
  mismatch with a sentinel we cannot read at all).

Exit codes
----------

- 0: success — output is on stdout. An empty ``repo_dir=`` line
  also means "no handoff" and is still exit 0.
- 2: usage error (bad / missing CLI arg, missing subcommand)
- 3: schema-version mismatch (the sentinel exists but its
  ``schema_version`` differs from this script's expectation; the
  worker should NOT proceed with the worktree because the field
  semantics may have changed). The worker still falls through to
  the legacy workflow.

Allow-rule
----------

This script is whitelisted as
``python3 tools/implementer_workspace.py discover *`` in the agent's
permission allowlist (see :file:`.opencode/agents/task-implementor.md`).
The cleanup of worktrees is the dispatcher's responsibility
(:meth:`tools._pr_clone.WorktreeHandle.cleanup`), not the worker's,
so this script is read-only by design — there is no ``cleanup``
subcommand.
"""
from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any


# Must match :data:`tools._pr_clone.WORKSPACE_HANDOFF_SCHEMA_VERSION`.
# Bump in both places when the on-disk shape changes.
EXPECTED_SCHEMA_VERSION = 1

_DEFAULT_WORKTREE_BASE = Path("/tmp/cleveragents-implementer-worktrees")


def _worktree_base() -> Path:
    """Resolve the worktree base. Honours
    ``IMPLEMENTER_DISPATCHER_WORKTREE_BASE`` so tests don't touch
    real ``/tmp`` and so an operator can sandbox a long-running
    dispatcher.
    """
    return Path(
        os.environ.get("IMPLEMENTER_DISPATCHER_WORKTREE_BASE")
        or str(_DEFAULT_WORKTREE_BASE)
    )


def _handoff_path(pr_number: int) -> Path:
    """Resolve the sentinel path for ``pr_number``. Mirrors
    :func:`tools._pr_clone._workspace_handoff_path`."""
    return _worktree_base() / f".handoff-pr-{int(pr_number)}.json"


def _load_handoff(pr_number: int) -> tuple[dict[str, Any] | None, str | None]:
    """Read + parse + validate the sentinel.

    Returns a tuple of ``(payload, error)``:

    - ``(payload, None)`` — sentinel exists, parses, schema matches,
      and the referenced ``repo_dir`` is on disk.
    - ``(None, None)`` — sentinel does not exist (the dispatcher
      didn't pre-clone, or already cleaned up). Normal "no handoff"
      condition; the worker falls through to its legacy path.
    - ``(None, error)`` — sentinel exists but is unusable (malformed,
      schema mismatch, repo_dir vanished). ``error`` is a short
      diagnostic emitted on stderr so the worker can log the reason.
    """
    target = _handoff_path(pr_number)
    if not target.exists():
        return None, None
    try:
        raw = target.read_text(encoding="utf-8")
        payload = json.loads(raw)
    except (OSError, ValueError) as e:
        return None, f"handoff read/parse failed at {target}: {e}"
    if not isinstance(payload, dict):
        return None, f"handoff at {target} is not a JSON object"
    schema = payload.get("schema_version")
    if schema != EXPECTED_SCHEMA_VERSION:
        return None, (
            f"handoff at {target} has schema_version={schema!r}, "
            f"expected {EXPECTED_SCHEMA_VERSION}"
        )
    repo_dir = payload.get("repo_dir")
    if not isinstance(repo_dir, str) or not repo_dir:
        return None, f"handoff at {target} has no usable repo_dir"
    if not Path(repo_dir).is_dir():
        return None, (
            f"handoff at {target} points at vanished worktree {repo_dir}"
        )
    return payload, None


def _cmd_discover(args: argparse.Namespace) -> int:
    """Implement ``discover``. Always prints both lines and exits
    0 on the happy "no handoff available" path; exits 3 only if the
    sentinel exists but has a schema mismatch (a config issue an
    operator must notice). All other "unusable sentinel" cases
    treat the absence as "no handoff" and exit 0."""
    payload, error = _load_handoff(int(args.pr))
    if error:
        print(f"warning: {error}", file=sys.stderr)
        # Schema mismatch is the one situation the worker should
        # NOT silently swallow — different field semantics between
        # script and dispatcher could mis-route the worker. Surface
        # via exit code while still printing the empty contract
        # output so even a worker that ignores exit codes falls back
        # safely.
        if "schema_version" in error:
            print("repo_dir=")
            print("branch=")
            return 3
        print("repo_dir=")
        print("branch=")
        return 0
    if payload is None:
        # No handoff exists — the dispatcher didn't pre-clone (e.g.
        # preclone gated off, fetch failed, no head_sha yet). Normal
        # path; emit the empty contract and exit 0.
        print("repo_dir=")
        print("branch=")
        return 0
    print(f"repo_dir={payload['repo_dir']}")
    print(f"branch={payload.get('branch') or ''}")
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="implementer_workspace",
        description=(
            "Worker-side reader for the dispatcher's workspace handoff "
            "sentinel. See module docstring for the full contract."
        ),
    )
    sub = parser.add_subparsers(dest="cmd", required=True)

    p_discover = sub.add_parser(
        "discover",
        help=(
            "Read the workspace handoff sentinel for PR N. Prints "
            "repo_dir=<path> and branch=<name> on stdout (both empty "
            "if no usable handoff). Always exits 0 except on schema "
            "mismatch (exit 3)."
        ),
    )
    p_discover.add_argument(
        "--pr", required=True, type=int, help="PR number"
    )

    args = parser.parse_args(argv)
    if args.cmd == "discover":
        return _cmd_discover(args)
    parser.error(f"unknown subcommand {args.cmd!r}")
    return 2  # pragma: no cover — argparse exits before reaching here


if __name__ == "__main__":
    sys.exit(main())