orchestrator/agent-team/agent_team/dispatcher.py

"""Trusted apply-path dispatcher (§4.3): carry a box-emitted diff into org CI.

The read-only R720 box (D2) EMITS a candidate diff + the
``workflow_dispatch`` inputs but holds **no write token**. This dispatcher is
the ONLY component that writes, and it runs on a TRUSTED host (the operator's
Mac, where the GitHub App key / operator ``gh`` auth lives) — never on the box.
It does two things and nothing else:

1. Pushes the candidate diff as a short-lived **head branch** (the PR head).
2. Triggers the ``agent-team-apply-verify.yml`` ``workflow_dispatch``, passing
   the diff as ``diff_b64`` plus the integrity inputs.

The CI workflow then RE-VERIFIES everything (materialize re-hashes; guard
re-hashes + denylist + scope; build-test builds/tests credential-less; the
pure-code gate decides pass/fail; the privileged job opens a DRAFT PR only on a
clean pass, gated by the ``agent-apply`` environment's required reviewer). This
dispatcher TRANSPORTS only — it makes no trust decision.

Design discipline (mirrors the rest of agent_team): the network/SDK/subprocess
side effects are behind INJECTED seams so the pure input-assembly + validation
logic is fully unit-testable with no git, no ``gh``, and no network. The real
default seams shell out to ``git`` / ``gh`` and are exercised only in
production.
"""

from __future__ import annotations

import base64
import re
from dataclasses import dataclass
from typing import Protocol

from agent_team.state_store import compute_content_hash

__all__ = [
    "DispatchInputs",
    "DispatcherError",
    "build_dispatch_inputs",
    "dispatch_apply_verify",
    "head_branch_for",
]

WORKFLOW_FILE = "agent-team-apply-verify.yml"

# The artifact carries exactly this filename; the workflow's materialize/guard
# steps look for ``candidate.diff`` (mirrors nodes.fixer.CANDIDATE_DIFF_FILENAME).
CANDIDATE_DIFF_FILENAME = "candidate.diff"

# Max candidate-diff size. NOT arbitrary: the diff is carried as the base64
# `diff_b64` workflow_dispatch INPUT, and GitHub caps total dispatch inputs at
# ~65,535 bytes. base64 inflates ~4/3, so a diff over ~45 KB cannot be
# dispatched at all. We cap at 40 KB (leaving headroom for the other inputs) and
# fail closed with a clear error rather than letting GitHub reject the dispatch
# opaquely — and to bound resource use on the operator host. Larger diffs need a
# branch-only transport (future), not an inline input.
MAX_DIFF_BYTES = 40_000

# task_id must match the workflow's gate-and-pr safety guard charset (it is
# embedded in the head ref and the PR text). A coordinator thread_id is a uuid,
# well within this set; we validate to fail closed on anything else.
_TASK_ID_RE = re.compile(r"\A[A-Za-z0-9._-]{1,200}\Z")
# owner/repo path-segment charset (mirrors ci_fetcher / github_intake).
_OWNER_REPO_RE = re.compile(r"\A[A-Za-z0-9_.-]{1,100}\Z")


class DispatcherError(Exception):
    """Raised on structurally invalid dispatch inputs (fail closed, never proceed)."""


def head_branch_for(task_id: str) -> str:
    """Return the head-branch ref the dispatcher pushes for ``task_id``.

    A stable, namespaced ref so a re-dispatch for the same task reuses/replaces
    one branch rather than littering refs. Validated to the safe charset so the
    ref can never carry shell/ref metacharacters.
    """
    if not _TASK_ID_RE.match(task_id or ""):
        raise DispatcherError(
            f"invalid task_id {task_id!r}: must match {_TASK_ID_RE.pattern}"
        )
    return f"agent-team/apply/{task_id}"


@dataclass(frozen=True)
class DispatchInputs:
    """The exact ``workflow_dispatch`` inputs for the apply/verify workflow.

    Field names match ``on.workflow_dispatch.inputs`` 1:1 so :meth:`as_inputs`
    can be handed straight to ``gh workflow run -f k=v``.
    """

    task_id: str
    diff_artifact_name: str
    expected_diff_hash: str
    declared_scope: str
    diff_b64: str
    head_branch: str

    def as_inputs(self) -> dict[str, str]:
        return {
            "task_id": self.task_id,
            "diff_artifact_name": self.diff_artifact_name,
            "expected_diff_hash": self.expected_diff_hash,
            "declared_scope": self.declared_scope,
            "diff_b64": self.diff_b64,
            "head_branch": self.head_branch,
        }


def build_dispatch_inputs(
    *,
    task_id: str,
    diff_text: str,
    declared_scope: str,
    artifact_prefix: str = "agent-team-diff",
) -> DispatchInputs:
    """Assemble the dispatch inputs from a task id + the candidate diff (PURE).

    Computes the sha256 the workflow binds to (``expected_diff_hash``), base64s
    the diff (``diff_b64``, carried as an input so the credential-less
    materialize job can re-create the artifact), derives the head branch, and
    names the artifact. No I/O, no network — fully testable.

    Fails closed (:class:`DispatcherError`) on an empty diff / empty scope /
    unsafe task id, so a malformed task can never be transported.
    """
    if not isinstance(diff_text, str) or not diff_text.strip():
        raise DispatcherError("diff_text must be a non-empty diff")
    if not isinstance(declared_scope, str) or not declared_scope.strip():
        raise DispatcherError(
            "declared_scope must be non-empty (an empty scope allows any path)"
        )
    head_branch = head_branch_for(task_id)  # validates task_id
    raw = diff_text.encode("utf-8")
    if len(raw) > MAX_DIFF_BYTES:
        raise DispatcherError(
            f"diff too large ({len(raw)} bytes > {MAX_DIFF_BYTES}); the diff is "
            "carried as a base64 workflow_dispatch input (GitHub caps inputs at "
            "~64 KB). Split the change or use a branch-only transport."
        )
    return DispatchInputs(
        task_id=task_id,
        diff_artifact_name=f"{artifact_prefix}-{task_id}",
        expected_diff_hash=compute_content_hash(raw),
        declared_scope=declared_scope,
        diff_b64=base64.b64encode(raw).decode("ascii"),
        head_branch=head_branch,
    )


class BranchPusher(Protocol):
    """Pushes the candidate diff as the head branch (the only WRITE)."""

    def __call__(
        self, *, owner: str, repo: str, base: str, head_branch: str, diff_text: str
    ) -> None: ...


class WorkflowDispatcher(Protocol):
    """Triggers the apply/verify ``workflow_dispatch`` with the assembled inputs."""

    def __call__(
        self, *, owner: str, repo: str, inputs: dict[str, str], ref: str
    ) -> None: ...


def dispatch_apply_verify(
    *,
    owner: str,
    repo: str,
    task_id: str,
    diff_text: str,
    declared_scope: str,
    base: str = "main",
    pusher: BranchPusher | None = None,
    dispatcher: WorkflowDispatcher | None = None,
) -> DispatchInputs:
    """Transport one candidate diff into org CI: push the head branch, dispatch.

    The trusted apply path. Validates owner/repo, assembles the dispatch inputs,
    pushes the diff as the head branch via ``pusher``, then triggers the
    workflow via ``dispatcher``. Returns the :class:`DispatchInputs` used (for
    the ledger/audit). ``pusher`` / ``dispatcher`` are injected so this is
    testable without git/gh/network; the real defaults shell out to git/gh.

    NOTE: this never runs on the box (the box has no write token, D2). CI owns
    every trust decision; this only moves bytes.
    """
    if not _OWNER_REPO_RE.match(owner or "") or not _OWNER_REPO_RE.match(repo or ""):
        raise DispatcherError(f"invalid owner/repo {owner!r}/{repo!r}")

    inputs = build_dispatch_inputs(
        task_id=task_id, diff_text=diff_text, declared_scope=declared_scope
    )
    push = pusher if pusher is not None else _default_branch_pusher()
    fire = dispatcher if dispatcher is not None else _default_workflow_dispatcher()

    # Push the head branch FIRST: the draft-PR step opens against an
    # already-pushed --head, so the branch must exist before the run reaches it.
    push(
        owner=owner,
        repo=repo,
        base=base,
        head_branch=inputs.head_branch,
        diff_text=diff_text,
    )
    fire(owner=owner, repo=repo, inputs=inputs.as_inputs(), ref=base)
    return inputs


def _default_branch_pusher() -> BranchPusher:
    """Real pusher: clone-free apply of the diff onto a fresh branch via git/gh.

    Deferred to call time (no subprocess at import). Uses the operator's git
    credentials / the GitHub App token present on the trusted host — NEVER a
    box-held token. Implemented as a thin shell-out; the heavy lifting is the
    pure :func:`build_dispatch_inputs` above, so this stays small.
    """

    def _push(
        *, owner: str, repo: str, base: str, head_branch: str, diff_text: str
    ) -> None:
        import subprocess
        import tempfile
        from pathlib import Path

        # Operator host only. Use a throwaway worktree, apply the diff, push the
        # branch with the host's credentials. Kept intentionally minimal; the
        # security comes from CI re-verifying the pushed content by hash.
        with tempfile.TemporaryDirectory() as tmp:
            tmpdir = Path(tmp)
            diff_path = tmpdir / CANDIDATE_DIFF_FILENAME
            diff_path.write_text(diff_text, encoding="utf-8")
            clone = tmpdir / "repo"
            subprocess.run(
                [
                    "gh",
                    "repo",
                    "clone",
                    f"{owner}/{repo}",
                    str(clone),
                    "--",
                    "--depth",
                    "1",
                    "--branch",
                    base,
                ],
                check=True,
                capture_output=True,
            )
            subprocess.run(
                ["git", "-C", str(clone), "checkout", "-B", head_branch],
                check=True,
                capture_output=True,
            )
            # --index applies AND stages exactly the diff's changes (incl. new
            # files) and NOTHING else — so the committed head tree is precisely
            # base+diff, never stray untracked worktree content. This keeps the
            # PR head bound to the same bytes CI hash-verified (LOGIC-1). No
            # separate `git add -A` (which would stage unrelated content).
            subprocess.run(
                ["git", "-C", str(clone), "apply", "--index", str(diff_path)],
                check=True,
                capture_output=True,
            )
            subprocess.run(
                [
                    "git",
                    "-C",
                    str(clone),
                    "-c",
                    "user.name=agent-team",
                    "-c",
                    "user.email=agent-team@seahavenind.com",
                    "commit",
                    "-m",
                    f"agent-team apply: {head_branch}",
                ],
                check=True,
                capture_output=True,
            )
            subprocess.run(
                [
                    "git",
                    "-C",
                    str(clone),
                    "push",
                    "--force-with-lease",
                    "origin",
                    head_branch,
                ],
                check=True,
                capture_output=True,
            )

    return _push


def _default_workflow_dispatcher() -> WorkflowDispatcher:
    """Real dispatcher: ``gh workflow run`` (operator auth has ``actions:write``)."""

    def _fire(*, owner: str, repo: str, inputs: dict[str, str], ref: str) -> None:
        import subprocess

        args = [
            "gh",
            "workflow",
            "run",
            WORKFLOW_FILE,
            "--repo",
            f"{owner}/{repo}",
            "--ref",
            ref,
        ]
        for key, value in inputs.items():
            args += ["-f", f"{key}={value}"]
        subprocess.run(args, check=True, capture_output=True)

    return _fire