Add ask-the-author: interrogate a task's implementing agent during review

src/panopticon/client.py

@@ -273,6 +273,61 @@ def resolve_responsibility(
             JsonObj, self._json(self._http.post(f"/tasks/{task_id}/responsibilities", json=body))
         )
 
+    # -- asks (ask-the-author) ----------------------------------------------------
+
+    def lookup_task(
+        self,
+        *,
+        repo_id: str | None = None,
+        branch: str | None = None,
+        url: str | None = None,
+    ) -> JsonObj | None:
+        """Find the task working a branch (``repo_id`` + ``branch``) or a URL/PR (``url``); ``None``
+        if none matches (404). The review tool's entry point."""
+        params = {
+            k: v
+            for k, v in {"repo_id": repo_id, "branch": branch, "url": url}.items()
+            if v is not None
+        }
+        resp = self._http.get("/tasks/lookup", params=params)
+        if resp.status_code == 404:
+            return None
+        return cast(JsonObj, self._json(resp))
+
+    def create_ask(self, task_id: str, question: str, context: str = "") -> str:
+        """Post a reviewer's question to a task's agent; return the ``ask_id`` to poll with."""
+        body: JsonObj = {"question": question, "context": context}
+        created = cast(JsonObj, self._json(self._http.post(f"/tasks/{task_id}/ask", json=body)))
+        return cast(str, created["ask_id"])
+
+    def get_ask(self, task_id: str, ask_id: str) -> JsonObj:
+        """Poll one ask: ``{ask_id, status, answer, question, context}``. Raises on 410 (gone)."""
+        return cast(JsonObj, self._json(self._http.get(f"/tasks/{task_id}/ask/{ask_id}")))
+
+    def outstanding_ask(self, task_id: str) -> str | None:
+        """The task's current unanswered ask id, or ``None`` (the container Stop hook uses this)."""
+        body = cast(JsonObj, self._json(self._http.get(f"/tasks/{task_id}/ask")))
+        return cast("str | None", body.get("ask_id"))
+
+    def mark_ask_delivered(self, task_id: str, ask_id: str) -> JsonObj:
+        """Report that the session service delivered the ask to the agent."""
+        return cast(
+            JsonObj, self._json(self._http.post(f"/tasks/{task_id}/ask/{ask_id}/delivered"))
+        )
+
+    def mark_ask_gone(self, task_id: str, ask_id: str) -> JsonObj:
+        """Report that the task's config volume is gone — the ask is undeliverable (→ 410)."""
+        return cast(JsonObj, self._json(self._http.post(f"/tasks/{task_id}/ask/{ask_id}/gone")))
+
+    def record_ask_answer(self, task_id: str, ask_id: str, answer: str) -> JsonObj:
+        """Record the agent's reply to an ask (the container Stop hook extracts it from the transcript)."""
+        return cast(
+            JsonObj,
+            self._json(
+                self._http.post(f"/tasks/{task_id}/ask/{ask_id}/answer", json={"answer": answer})
+            ),
+        )
+
     # -- artifacts ----------------------------------------------------------------
 
     def list_artifacts(self, task_id: str) -> list[str]:

src/panopticon/container/agent.py

@@ -116,6 +116,7 @@ def _claude_argv(
     initial_prompt: str | None = None,
     turn: str | None = None,
     starting_model: str | None = None,
+    ask_prompt: str | None = None,
 ) -> list[str]:
     """`claude` argv, resuming the project's most recent conversation if one exists.
 
@@ -134,6 +135,11 @@ def _claude_argv(
     the agent's turn (``turn == "agent"``), :data:`INTERRUPT_PROMPT` is appended instead so the
     agent automatically picks up where it left off rather than waiting for user input.
 
+    ``ask_prompt`` (ask-the-author) takes precedence over both: when a parked/terminal task is
+    resumed to answer a reviewer's question, it's appended as the positional prompt so claude
+    processes the question as the agent's next message (the message already carries the read-only
+    guardrail; the agent answers and its Stop hook records the reply).
+
     ``starting_model`` (e.g. ``"opus"``) is passed as ``--model`` on the **first run only** — on
     resume claude uses whichever model the conversation was already using.
     """
@@ -147,17 +153,16 @@ def _claude_argv(
     project = config_dir / "projects" / str(cwd).replace("/", "-")
     if any(project.glob("*.jsonl")):
         argv.append("--continue")
-        if turn == "agent":
-            argv.append(INTERRUPT_PROMPT)  # positional: auto-resume after container restart
+        # An ask (delivered as the resume prompt) wins over the generic INTERRUPT_PROMPT.
+        positional = ask_prompt or (INTERRUPT_PROMPT if turn == "agent" else None)
     else:
         if (
             starting_model
         ):  # first run only — on resume claude uses the conversation's existing model
             argv += ["--model", starting_model]
-        if initial_prompt:
-            argv.append(
-                initial_prompt
-            )  # positional: claude sends this as the agent's first message
+        positional = ask_prompt or initial_prompt
+    if positional:  # positional: claude sends this as the agent's next message
+        argv.append(positional)
     return argv
 
 
@@ -170,12 +175,14 @@ def _run_claude(config_dir: Path) -> None:  # pragma: no cover - real LLM; skipi
     initial_prompt = os.environ.get("PANOPTICON_INITIAL_PROMPT") or None
     turn = os.environ.get("PANOPTICON_TASK_TURN") or None
     starting_model = os.environ.get("PANOPTICON_STARTING_MODEL") or None
+    ask_prompt = os.environ.get("PANOPTICON_ASK_PROMPT") or None
     argv = _claude_argv(
         config_dir,
         Path.cwd(),
         initial_prompt=initial_prompt,
         turn=turn,
         starting_model=starting_model,
+        ask_prompt=ask_prompt,
     )
     subprocess.run(argv, env={**os.environ, "CLAUDE_CONFIG_DIR": str(config_dir)})
 

src/panopticon/container/hook.py

@@ -40,6 +40,7 @@
 
 from panopticon.client import TaskServiceClient
 from panopticon.container.pricing import cost_weighted_tokens
+from panopticon.core.asking import ask_marker
 from panopticon.core.provisioning import PROVISION_NUDGE
 
 #: A background task's ``status`` value counts as *finished* (no longer in flight) only if it's one
@@ -124,6 +125,81 @@ def _line_tokens(line: str) -> int:
     return cost_weighted_tokens(int_usage, model)
 
 
+def _message_text(line: str, *, role: str) -> str | None:
+    """The text of a transcript line if it's a message from ``role`` (``"user"``/``"assistant"``),
+    else ``None``. Content is either a plain string or a list of blocks; only ``text`` blocks count
+    (tool calls/results are skipped). Tolerant of non-JSON / unexpected shapes (returns ``None``)."""
+    try:
+        obj = json.loads(line)
+        msg = obj.get("message") or {}
+        if not isinstance(msg, dict) or msg.get("role") != role:
+            return None
+        content = msg.get("content")
+    except (ValueError, AttributeError):
+        return None
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        texts = [
+            b["text"]
+            for b in content
+            if isinstance(b, dict) and b.get("type") == "text" and isinstance(b.get("text"), str)
+        ]
+        return "\n".join(texts)
+    return None
+
+
+def extract_answer(transcript_path: str, marker: str) -> str | None:
+    """The agent's reply to an ask, read from the session transcript (ask-the-author).
+
+    Finds the **last** user message carrying ``marker`` (the ask we delivered) and concatenates the
+    assistant ``text`` that follows it, up to the next user message (the boundary of that turn).
+    Returns the joined text, or ``None`` if the marker isn't present yet or produced no text — the
+    Stop hook then simply records nothing (best-effort). Pure + tolerant, so it's unit-testable with
+    a fixture transcript, like :func:`session_tokens`."""
+    try:
+        lines = Path(transcript_path).read_text().splitlines()
+    except OSError:
+        return None
+    start: int | None = None
+    for i, line in enumerate(lines):
+        text = _message_text(line, role="user")
+        if text is not None and marker in text:
+            start = i  # keep the last match — the most recent delivery of this ask
+    if start is None:
+        return None
+    parts: list[str] = []
+    for line in lines[start + 1 :]:
+        if _message_text(line, role="user") is not None:
+            break  # a new user turn began — the reply is complete
+        assistant_text = _message_text(line, role="assistant")
+        if assistant_text:
+            parts.append(assistant_text)
+    answer = "\n".join(parts).strip()
+    return answer or None
+
+
+def _maybe_record_ask_answer(
+    client: TaskServiceClient, task_id: str, payload: dict[str, Any]
+) -> None:
+    """If a reviewer's ask is outstanding, record the agent's just-finished reply (best-effort).
+
+    The Stop hook is the completion signal: the agent has answered, so we pull its reply from the
+    transcript the payload names and record it. Any failure — no transcript, no outstanding ask, a
+    REST error, an empty extraction — is swallowed so ask handling never breaks the turn flip the
+    hook exists for. This does **not** transition the task; recording an answer is a plain fact."""
+    transcript = payload.get("transcript_path")
+    if not isinstance(transcript, str):
+        return
+    with contextlib.suppress(httpx.HTTPError):
+        ask_id = client.outstanding_ask(task_id)
+        if ask_id is None:
+            return
+        answer = extract_answer(transcript, ask_marker(ask_id))
+        if answer:
+            client.record_ask_answer(task_id, ask_id, answer)
+
+
 def _report_tokens(client: TaskServiceClient, task_id: str, payload: dict[str, Any]) -> None:
     """Best-effort: total the transcript the Stop payload names and record it.
 
@@ -167,6 +243,8 @@ def main(
         _report_tokens(client, task_id, payload)
         if _has_live_background_task(payload):
             return 0
+        # A real stop (nothing in flight): if a reviewer's ask is outstanding, record the reply.
+        _maybe_record_ask_answer(client, task_id, payload)
     client.set_turn(task_id, actor)
     # `prompt` (UserPromptSubmit): ground the agent in its current phase, and (while the task is
     # unslugged) nudge toward provisioning. claude adds this hook's stdout to its context.

src/panopticon/core/asking.py

@@ -0,0 +1,47 @@
+"""Composing the message delivered to a task's agent for an *ask* (ask-the-author).
+
+Pure and LLM-free — shared by the session service (which delivers the message) and the container
+Stop hook (which locates the agent's reply in the transcript by the same marker). Keeping it here,
+in ``core``, means the wording + marker have one definition and can be unit-tested without a runner.
+"""
+
+from __future__ import annotations
+
+#: Prefix of the marker embedded at the top of every delivered ask. Stable and greppable so the
+#: container Stop hook can find the user message that carried *this* ask and collect the assistant
+#: reply that follows it (see :func:`ask_marker`).
+ASK_MARKER_PREFIX = "panopticon-ask"
+
+
+def ask_marker(ask_id: str) -> str:
+    """The marker for one ask — embedded in the delivered message and matched in the transcript."""
+    return f"[[{ASK_MARKER_PREFIX}:{ask_id}]]"
+
+
+def compose_ask_message(question: str, context: str, *, terminal: bool, ask_id: str) -> str:
+    """The full message delivered to the agent for an ask: the marker, a framing, and the question.
+
+    ``terminal`` selects the framing. For a **terminal** task (COMPLETE/DROPPED) the guardrail is
+    strict — the agent is answering about merged/proposed work and must not modify anything or touch
+    the branch (the memo's requirement). For a task still in flight it's a lighter note that a
+    reviewer is asking a question, without forbidding changes (a reviewer's question may legitimately
+    prompt a fix mid-review). The marker is always first so the Stop hook can anchor the reply.
+    """
+    marker = ask_marker(ask_id)
+    lines = [marker, ""]
+    if terminal:
+        lines += [
+            "A reviewer is asking a question about your merged or proposed work on this task.",
+            "You are ONLY answering a question — this is not a request to change anything.",
+            "Do not modify files, run git, create commits, or use workflow tools. Answer from your",
+            "session memory and, if needed, read-only inspection of the code.",
+        ]
+    else:
+        lines += [
+            "A reviewer looking at this task is asking you a question.",
+            "Answer from your knowledge of this work — you need not change anything to reply.",
+        ]
+    lines += ["", "Reviewer's question:", question.strip()]
+    if context.strip():
+        lines += ["", "Context from the reviewer:", context.strip()]
+    return "\n".join(lines)

src/panopticon/core/models.py

@@ -35,6 +35,21 @@ class Status(str, Enum):
     FAILED = "failed"  # could not be satisfied; requires a comment
 
 
+class AskStatus(str, Enum):
+    """The lifecycle of a reviewer's *ask* (ask-the-author) — a question delivered to a task's
+    implementing agent, tracked so a review tool can poll for the answer.
+
+    ``PENDING`` → the session service delivers it → ``DELIVERED`` → the agent's Stop hook records
+    its reply → ``ANSWERED``. ``GONE`` is terminal too: the task's config volume was reaped, so the
+    session can't be resumed and the ask is undeliverable (the API surfaces it as HTTP 410).
+    """
+
+    PENDING = "pending"  # created; the session service hasn't delivered it to the container yet
+    DELIVERED = "delivered"  # handed to the agent (tmux inject / --continue resume); awaiting Stop
+    ANSWERED = "answered"  # the agent's reply was extracted from the transcript and recorded
+    GONE = "gone"  # the config volume is gone (reaped) — the session can't be resumed (→ 410)
+
+
 class LifecyclePhase(str, Enum):
     """A step the **session service** reports as it brings a task's container up (ADR 0008).
 
@@ -336,3 +351,27 @@ def outstanding_responsibilities(self) -> list[Responsibility]:
         its comment, so it never lingers here.
         """
         return [r for r in self.current_entry.responsibilities if r.status is Status.PENDING]
+
+
+@dataclass
+class Ask:
+    """A reviewer's question delivered to a task's implementing agent (ask-the-author).
+
+    The tarot review tool POSTs a question; the session service delivers it to the task's claude
+    session (tmux inject if live, ``--continue`` resume if parked) and the agent's reply is
+    extracted from the transcript after its Stop. An ask is **conversation, not a transition**: it
+    never changes the task's state or seeds responsibilities (the turn may flip agent↔user as the
+    agent answers, exactly as in any turn).
+
+    Ephemeral, like a registration or a lifecycle phase: held in the task service's memory, not the
+    store — so it never bumps the store's version and is lost on a service restart (a review-time
+    conversation; the review tool re-asks). Identity is ``id``; ``created_at`` orders a task's asks.
+    """
+
+    id: str
+    task_id: str
+    question: str
+    context: str = ""
+    status: AskStatus = AskStatus.PENDING
+    answer: str | None = None
+    created_at: str | None = None

src/panopticon/core/store.py

@@ -129,6 +129,17 @@ async def list_tasks_summary(self) -> list[Task]:
         """Return all tasks without history (cheap: tasks-table data only)."""
         return await self._list_tasks_summary()
 
+    async def find_task_by_branch(self, repo_id: str, branch: str) -> Task | None:
+        """Return the task on ``repo_id`` working the given ``branch``, or ``None`` (a thin lookup
+        over an existing column — the branch a task provisions is unique per repo). If more than one
+        matches (shouldn't happen), the most-recently-created is returned."""
+        return await self._find_task_by_branch(repo_id, branch)
+
+    async def find_task_by_url(self, url: str) -> Task | None:
+        """Return the task whose recorded ``url`` (e.g. its PR) matches, or ``None`` (most-recently
+        created if more than one)."""
+        return await self._find_task_by_url(url)
+
     async def save_task(self, task: Task) -> None:
         """Persist an updated task, enforcing consistency and append-only history."""
         validate_task_consistency(task)
@@ -172,6 +183,14 @@ async def _list_tasks(self) -> list[Task]:
     async def _list_tasks_summary(self) -> list[Task]:
         """Return all tasks with ``history=[]`` (no history loaded)."""
 
+    @abstractmethod
+    async def _find_task_by_branch(self, repo_id: str, branch: str) -> Task | None:
+        """Return the task on ``repo_id`` with the given ``branch`` (most recent if >1), or ``None``."""
+
+    @abstractmethod
+    async def _find_task_by_url(self, url: str) -> Task | None:
+        """Return the task with the given ``url`` (most recent if >1), or ``None``."""
+
     @abstractmethod
     async def _stored_history(self, task_id: str) -> list[HistoryEntry]:
         """Return the task's persisted history. Raise :class:`NotFound` if it does not exist."""