nisse/nisse-dev/src/nisse_dev/delegation.py

"""Delegation harness (architecture L9).

Escalation is delegation, not consultation: a frontier model receives a scoped
sub-task, works autonomously with full tool access, and hands back. The local
model does not interpret or re-derive its reasoning — the handback note plus
the resulting file state is the entire interface back.

Three properties do the work here:

**Isolation.** The delegation runs on its own branch, taken at start. Abort
discards the branch, which gives a delegation the same atomicity that boot
reconciliation gives an interrupted state transition.

**Caps that are hard.** Turns, tokens, and wall clock. A breach is a defined
failure return, not a nudge: the local loop gates to the human rather than
resuming on top of a half-edited tree.

**Checkpointing as it happens.** Every tool call, result, and assistant turn
is written to the canonical log when it occurs, never batched at handback.
That is what makes a delegation resumable after a crash (L8).

Scope is bounded by construction: a delegation is handed a sub-task the
approved plan already contains. It is smart, not authorised to widen the work.
"""

import time
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any, Literal

from nisse_dev.cancel import Cancel, skip_calls
from nisse_dev.canonical import (
    AssistantText,
    DelegationCompleted,
    DelegationStarted,
    Fact,
    PlanSubmitted,
    StoredEvent,
    ToolCall,
    ToolResult,
)
from nisse_dev.git import GitError, GitRepo
from nisse_dev.llm import LLMClient, LLMError, UpstreamUnavailable
from nisse_dev.reconcile import Reconciler
from nisse_dev.state import Intent, StateStore, Transcript
from nisse_dev.tools import EXECUTION_TOOLS, MUTATING_TOOLS, Workspace

DELEGATION_INTENT = "delegation"

SYSTEM_PROMPT = """You are a delegate working on one scoped sub-task for
nisse-dev. You have full tool access and are expected to finish the sub-task
yourself, including running the tests.

Stay inside the sub-task you were given. It comes from a plan a human already
approved; widening it is not yours to do. If the sub-task turns out to be
wrong or impossible, say so in the handback rather than substituting your own.

When you are done, call hand_back exactly once. Write it for {reader},
which cannot see your reasoning and will act on your note alone:

- what changed
- what is now true that was not before
- what to do next

Do not call hand_back until the work is actually done."""

HAND_BACK_TOOL: dict[str, Any] = {
    "type": "function",
    "function": {
        "name": "hand_back",
        "description": (
            "End the delegation and report. Call exactly once, when the "
            "sub-task is complete or you have determined it cannot be."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "changed": {
                    "type": "string",
                    "description": "What you changed, concretely.",
                },
                "now_true": {
                    "type": "string",
                    "description": "What is now true that was not before.",
                },
                "next": {
                    "type": "string",
                    "description": "What the caller should do next.",
                },
                "succeeded": {
                    "type": "boolean",
                    "description": "Whether the sub-task was completed.",
                },
            },
            "required": ["changed", "now_true", "next", "succeeded"],
        },
    },
}

DELEGATE_TOOLS = [*EXECUTION_TOOLS, HAND_BACK_TOOL]


@dataclass(frozen=True)
class Caps:
    turns: int = 30
    tokens: int = 200_000
    wall_clock_s: float = 900.0

    def as_payload(self) -> dict[str, Any]:
        return {
            "turns": self.turns,
            "tokens": self.tokens,
            "wall_clock_s": self.wall_clock_s,
        }


@dataclass(frozen=True)
class Handback:
    """The entire interface back to the local loop, alongside file state."""

    changed: str
    now_true: str
    next: str
    succeeded: bool

    def render(self) -> str:
        return (
            f"What changed: {self.changed}\n"
            f"What is now true: {self.now_true}\n"
            f"What to do next: {self.next}"
        )


@dataclass
class DelegationResult:
    status: Literal["handback", "capped", "failed", "interrupted", "stopped"]
    note: str
    handback: Handback | None = None
    branch: str | None = None
    artifacts: list[str] = field(default_factory=list)
    tokens: int = 0
    duration_s: float = 0.0

    @property
    def interrupted(self) -> bool:
        """Stopped by something external rather than by the delegate.

        The work on the branch may be perfectly good, so it is preserved and
        the intent left pending: the next boot treats it exactly like a crash
        mid-delegation (L8) and resumes if the cut was clean. Discarding here
        would throw away work because the budget ran out.
        """
        return self.status == "interrupted"

    @property
    def usable(self) -> bool:
        """Only a successful handback may be built upon. A cap breach, a
        failure or an operator's stop leaves a half-edited tree, which L9
        says must gate rather than be resumed."""
        return self.status == "handback" and self.handback is not None


class Delegation:
    def __init__(
        self,
        client: LLMClient,
        store: StateStore,
        workspace: Workspace,
        git_repo: GitRepo,
        *,
        model: str,
        backing_model: str = "",
        caps: Caps | None = None,
        now: Callable[[], float] = time.monotonic,
        cancel: Cancel | None = None,
        reader: str = "",
        on_turn: Callable[[str], None] | None = None,
    ) -> None:
        self._client = client
        self._store = store
        self._workspace = workspace
        # Who acts on the handback, from its capability profile (L9): never
        # a model's name.
        self._reader = reader or "a materially weaker model"
        # The delegate's own text for each turn, for progress (#42).
        self._on_turn = on_turn
        self._git = git_repo
        self._model = model
        self._backing_model = backing_model
        self._caps = caps or Caps()
        self._now = now
        # A stopped delegation is discarded like a failed one, not preserved
        # for resuming as an interrupted one is: the operator stopped it for
        # a reason, and the next step is to talk about that reason.
        self._cancel = cancel or Cancel()

    def run(
        self, transcript: Transcript, *, brief: str, branch: str, role: str = "helper"
    ) -> DelegationResult:
        """Run a delegation in the work item whose transcript this is,
        writing as the helper (L6, L7); its intent is owned by that item."""
        started_at = self._now()
        base_branch = self._git.current_branch()
        # Recorded before the branch exists, so a crash anywhere after this
        # leaves a durable statement of what was in flight (L8).
        intent_id = transcript.begin_intent(
            DELEGATION_INTENT,
            {
                "branch": branch,
                "base": base_branch,
                "brief": brief,
                "role": role,
                "caps": self._caps.as_payload(),
            },
            cause=f"delegating to {self._model}",
        )
        try:
            # Commit whatever the caller had in flight onto their own branch
            # first. Otherwise it would ride along onto the delegation branch
            # and be destroyed by an abort — discarding work nobody delegated.
            self._git.commit_all("Checkpoint before delegating")
            self._git.checkout_new_branch(branch)
        except GitError as exc:
            self._store.abort_intent(intent_id, cause=f"could not branch: {exc}")
            return DelegationResult("failed", f"could not branch for delegation: {exc}")

        transcript.append(
            DelegationStarted(
                brief=brief,
                role=role,
                model=self._model,
                caps=self._caps.as_payload(),
                backing_model=self._backing_model,
            )
        )
        result = self._loop(transcript, brief, branch, started_at)
        result.duration_s = self._now() - started_at
        transcript.append(
            DelegationCompleted(
                handback=result.handback.render() if result.handback else result.note,
                artifacts=result.artifacts,
                tokens=result.tokens,
                duration_s=result.duration_s,
                exit_status=result.status,
            )
        )
        if result.interrupted:
            # Leave the branch and the intent alone: this delegation is not
            # finished, it was cut off, and L8's resume rule is exactly the
            # machinery for picking it back up.
            return result
        if not result.usable:
            # Abort discards the branch: the caller must never resume on top
            # of a partially-edited tree (L9).
            self._abandon(branch, base_branch)
        # Settled either way: an aborted delegation is a *finished* one, and
        # leaving the intent pending would make the next boot think work was
        # still in flight.
        self._store.commit_intent(intent_id, cause=f"delegation {result.status}")
        return result

    def resume(self, intent: Intent) -> DelegationResult:
        """Continue a delegation the process died in the middle of (L8 boot step 5).

        The transcript is already in the canonical log, so the loop re-renders
        it and carries on. No new branch and no second ``delegation_started``:
        this is the same delegation, which is why its vendor-opaque reasoning
        may still be replayed (L7 defines a session as the delegation, not the
        process lifetime).
        """
        payload = intent.payload
        branch = str(payload["branch"])
        transcript = self._store.transcript(intent.work_item, "helper")
        started_at = self._now()
        try:
            self._git.checkout_existing_branch(branch)
        except GitError as exc:
            self._store.commit_intent(
                intent.id, cause=f"delegation branch unusable: {exc}"
            )
            return DelegationResult("failed", f"could not resume on {branch}: {exc}")

        result = self._loop(
            transcript, str(payload.get("brief", "")), branch, started_at
        )
        result.duration_s = self._now() - started_at
        transcript.append(
            DelegationCompleted(
                handback=result.handback.render() if result.handback else result.note,
                artifacts=result.artifacts,
                tokens=result.tokens,
                duration_s=result.duration_s,
                exit_status=result.status,
            )
        )
        if not result.usable:
            self._abandon(branch, str(payload.get("base", "main")))
        self._store.commit_intent(intent.id, cause=f"delegation {result.status}")
        return result

    def recover(self, notify: Callable[[str], None]) -> DelegationResult | None:
        """Sweep pending delegation intents at boot (L8 boot step 5).

        Settles what is already finished, discards what never started, resumes
        a clean cut — announcing it with a *notice*, not a question, because a
        clean cut needs no decision from the operator — and leaves an unclean
        one pending so the generic boot sweep gates on it.
        """
        for intent in self._store.pending_intents():
            if intent.kind != DELEGATION_INTENT:
                continue
            verdict = delegation_verdict(self._store, self._git, intent)
            branch = intent.payload.get("branch")
            if verdict == "settled":
                self._store.commit_intent(
                    intent.id, cause="delegation had already completed"
                )
            elif verdict == "absent":
                self._store.abort_intent(
                    intent.id, cause=f"delegation branch {branch} never created"
                )
            elif verdict == "resume":
                notify(
                    f"resuming the delegation interrupted on {branch}; "
                    "the transcript and branch are intact"
                )
                return self.resume(intent)
        return None

    def _loop(
        self, transcript: Transcript, brief: str, branch: str, started_at: float
    ) -> DelegationResult:
        system = (
            f"{SYSTEM_PROMPT.format(reader=self._reader)}\n\n"
            f"{self._workspace.describe()}"
            f"\n\nYour sub-task:\n{brief}"
        )
        touched: set[str] = set()
        did_work = False
        spent = 0
        after = _plan_scope(transcript)

        for turn in range(self._caps.turns):
            if self._cancel.requested:
                return DelegationResult(
                    "stopped",
                    f"stopped by the operator before turn {turn + 1}",
                    branch=branch,
                    artifacts=sorted(touched),
                    tokens=spent,
                )
            elapsed = self._now() - started_at
            if elapsed >= self._caps.wall_clock_s:
                return DelegationResult(
                    "capped",
                    f"wall-clock cap of {self._caps.wall_clock_s}s reached "
                    f"after {turn} turns",
                    branch=branch,
                    artifacts=sorted(touched),
                    tokens=spent,
                )
            try:
                produced = self._client.complete(
                    transcript.view(after=after),
                    tools=DELEGATE_TOOLS,
                    model=self._model,
                    system=system,
                )
            except UpstreamUnavailable as exc:
                return DelegationResult(
                    "interrupted",
                    f"stopped by something upstream, not by the work: {exc}",
                    branch=branch,
                    artifacts=sorted(touched),
                    tokens=spent,
                )
            except LLMError as exc:
                return DelegationResult(
                    "failed",
                    f"delegate call failed: {exc}",
                    branch=branch,
                    tokens=spent,
                )
            # Metered from what the provider reported, not estimated. The cap
            # is a scope limit (L9), distinct from the money cap L9 keeps on
            # the LiteLLM key — but an unbounded delegation can still exhaust
            # a daily budget and take external models away until it resets.
            usage = getattr(self._client, "last_usage", {}) or {}
            spent += int(usage.get("input", 0)) + int(usage.get("output", 0))
            if spent >= self._caps.tokens:
                return DelegationResult(
                    "capped",
                    f"token cap of {self._caps.tokens} reached ({spent} spent) "
                    f"after {turn + 1} turns",
                    branch=branch,
                    artifacts=sorted(touched),
                    tokens=spent,
                )

            calls: list[ToolCall] = []
            for event in produced:
                # Checkpoint as it arrives, never batched at handback (L9):
                # this is what makes the delegation resumable after a crash.
                transcript.append(event)
                if isinstance(event, ToolCall):
                    calls.append(event)
            said = " ".join(
                e.text.strip() for e in produced if isinstance(e, AssistantText)
            ).strip()
            if said and self._on_turn is not None:
                self._on_turn(said)

            if not calls:
                # A delegate that stops without handing back has not reported,
                # and its tree state is unknown. Treat it as a failure rather
                # than inventing a handback on its behalf.
                if any(isinstance(e, AssistantText) for e in produced):
                    # Said something but did nothing: told so, as the
                    # harness, so the next request ends on the user's side
                    # (models refuse assistant prefill).
                    transcript.as_actor("orchestrator").append(
                        Fact(
                            text="Your last reply was prose. Use your tools to "
                            "do the work, then call hand_back."
                        )
                    )
                    continue
                return DelegationResult(
                    "failed",
                    "delegate produced neither a tool call nor text",
                    branch=branch,
                    artifacts=sorted(touched),
                    tokens=spent,
                )

            for index, call in enumerate(calls):
                if self._cancel.requested:
                    skip_calls(transcript, calls[index:])
                    return DelegationResult(
                        "stopped",
                        f"stopped by the operator during turn {turn + 1}",
                        branch=branch,
                        artifacts=sorted(touched),
                        tokens=spent,
                    )
                if call.tool == "hand_back":
                    handback = _parse_handback(call.arguments)
                    if handback is not None and handback.succeeded and not did_work:
                        # Success claimed with nothing changed. The handback
                        # and the file state are the whole interface back, so
                        # an unbacked claim would hand the caller a note about
                        # work that does not exist.
                        transcript.append(
                            ToolResult(
                                call_id=call.call_id,
                                result=(
                                    "you have not changed anything yet, so the "
                                    "sub-task cannot be complete; do the work, "
                                    "or hand back with succeeded=false and say "
                                    "why"
                                ),
                                success=False,
                            )
                        )
                        continue
                    if handback is None:
                        transcript.append(
                            ToolResult(
                                call_id=call.call_id,
                                result=(
                                    "hand_back needs changed, now_true, next and "
                                    "succeeded; the delegation is still open"
                                ),
                                success=False,
                            )
                        )
                        continue
                    transcript.append(
                        ToolResult(
                            call_id=call.call_id,
                            result="handback accepted; delegation closed",
                            success=True,
                        )
                    )
                    self._commit_turn(turn)
                    return DelegationResult(
                        "handback" if handback.succeeded else "failed",
                        handback.next if handback.succeeded else handback.changed,
                        handback=handback if handback.succeeded else None,
                        branch=branch,
                        artifacts=sorted(touched),
                        tokens=spent,
                    )

                outcome = self._workspace.dispatch(call.tool, call.arguments)
                if call.tool in MUTATING_TOOLS and outcome.success:
                    did_work = True
                    path = call.arguments.get("path")
                    if isinstance(path, str):
                        touched.add(path)
                transcript.append(
                    ToolResult(
                        call_id=call.call_id,
                        result=outcome.result,
                        success=outcome.success,
                    )
                )

            # Commit each turn's work onto the delegation branch, so that
            # discarding the branch really does discard the work, and so a
            # crash leaves the branch at a known commit (L8).
            self._commit_turn(turn)

        return DelegationResult(
            "capped",
            f"turn cap of {self._caps.turns} reached without a handback",
            branch=branch,
            artifacts=sorted(touched),
            tokens=spent,
        )

    def _commit_turn(self, turn: int) -> None:
        try:
            self._git.commit_all(f"Delegation turn {turn + 1}")
        except GitError:
            # Losing a checkpoint commit does not invalidate the work; the
            # canonical log still records what happened.
            pass

    def _abandon(self, branch: str, base_branch: str) -> None:
        try:
            self._git.discard_branch(branch, back_to=base_branch)
        except GitError:
            # The branch is already recorded in the delegation events, so a
            # failed cleanup leaves evidence rather than a silent mess.
            pass


def _parse_handback(arguments: dict[str, Any]) -> Handback | None:
    """Validated locally rather than trusted: L9's rule for prompt-level
    structured output applies to the handback too."""
    required = ("changed", "now_true", "next")
    if any(not str(arguments.get(key, "")).strip() for key in required):
        return None
    succeeded = arguments.get("succeeded")
    if not isinstance(succeeded, bool):
        return None
    return Handback(
        changed=str(arguments["changed"]).strip(),
        now_true=str(arguments["now_true"]).strip(),
        next=str(arguments["next"]).strip(),
        succeeded=succeeded,
    )


def delegation_reconciler(store: StateStore, git_repo: GitRepo) -> Reconciler:
    """L8 boot step 5: decide whether an interrupted delegation may be resumed.

    The judgement is an observable check, never model self-assessment — the
    same reasoning as L9. Returns:

    - ``True``  the delegation is settled or cleanly resumable
    - ``False`` it never really started, so there is nothing to resume
    - ``None``  the cut is unclean, so boot gates to the operator

    "The branch sits at the expected commit" is approximated by the branch
    existing and carrying the delegation's commits: per-turn shas are not
    recorded in the log today, so a branch rewritten out from under us would
    not be detected. Stated rather than glossed, because the check is weaker
    than L8's wording implies.
    """

    def check(intent: Intent) -> bool | None:
        verdict = delegation_verdict(store, git_repo, intent)
        if verdict in ("resume", "settled"):
            return True
        return False if verdict == "absent" else None

    return check


def delegation_verdict(
    store: StateStore, git_repo: GitRepo, intent: Intent
) -> Literal["resume", "settled", "absent", "gate"]:
    """The L8 boot step 5 decision, separated from the Reconciler contract because
    boot must *act* on it — resuming is not the same as judging settled."""
    branch = intent.payload.get("branch")
    if not branch:
        return "gate"
    # The delegation's own work item: an intent is owned by the item it ran in.
    events = store.transcript(intent.work_item, "helper").read()
    if _completed_after_start(events):
        return "settled"  # it finished; the intent merely outlived its commit
    if not git_repo.branch_exists(str(branch)):
        return "absent"  # the crash landed before any work existed
    if not events:
        return "gate"
    last = events[-1].event
    if isinstance(last, ToolCall):
        # A call with no recorded result: its effect is unknown and cannot be
        # verified by inspection, so gate (L8).
        return "gate"
    if not isinstance(last, AssistantText | ToolResult):
        return "gate"
    return "resume"


def _plan_scope(transcript: Transcript) -> int:
    """Where the helper's view begins: at the approved plan the work is under,
    so it sees the plan and this milestone, not earlier ones (L6)."""
    latest = 0
    for stored in transcript.read():
        if isinstance(stored.event, PlanSubmitted):
            latest = stored.id
    return max(latest - 1, 0)


def _completed_after_start(events: list[StoredEvent]) -> bool:
    """True when the last delegation event in the log is a completion."""
    for stored in reversed(events):
        if isinstance(stored.event, DelegationCompleted):
            return True
        if isinstance(stored.event, DelegationStarted):
            return False
    return False