diff --git a/src/panopticon/client.py b/src/panopticon/client.py
index 01d6f7f4..ce4b2c3a 100644
--- a/src/panopticon/client.py
+++ b/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]:
diff --git a/src/panopticon/container/agent.py b/src/panopticon/container/agent.py
index ce10c642..5c58fc52 100644
--- a/src/panopticon/container/agent.py
+++ b/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)})
 
diff --git a/src/panopticon/container/hook.py b/src/panopticon/container/hook.py
index cc06478f..fe65c884 100644
--- a/src/panopticon/container/hook.py
+++ b/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.
diff --git a/src/panopticon/core/asking.py b/src/panopticon/core/asking.py
new file mode 100644
index 00000000..f7462b78
--- /dev/null
+++ b/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)
diff --git a/src/panopticon/core/models.py b/src/panopticon/core/models.py
index fcfa01cb..c832d698 100644
--- a/src/panopticon/core/models.py
+++ b/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
diff --git a/src/panopticon/core/store.py b/src/panopticon/core/store.py
index 8d4be3b5..5d146948 100644
--- a/src/panopticon/core/store.py
+++ b/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."""
diff --git a/src/panopticon/sessionservice/ask_worker.py b/src/panopticon/sessionservice/ask_worker.py
new file mode 100644
index 00000000..40de9eb8
--- /dev/null
+++ b/src/panopticon/sessionservice/ask_worker.py
@@ -0,0 +1,85 @@
+"""Host-side ask delivery (ask-the-author): hand a reviewer's question to a task's claude session.
+
+The sibling of :class:`~panopticon.sessionservice.provisioner.Provisioner` for the ask side. The
+task service records the *ask* (question → answer) but, being LLM-free and docker-free, never touches
+a container; the session service runs **where the container runs**, so it owns delivery:
+
+* a **live** task (its tmux session exists) → inject the message into the running agent's pane
+  (:meth:`LocalRunner.send_to_session`);
+* a **parked** task (no session — including a COMPLETE one whose author we want to interrogate) →
+  resume it via ``claude --continue`` with the question as the prompt
+  (:meth:`Spawner.spawn_for_ask`), which allows terminal tasks unlike the normal spawn gate;
+* a task whose **config volume was reaped** → mark the ask ``gone`` (the API returns 410 and the
+  review tool falls back).
+
+The agent's reply is recorded by the container's Stop hook (it has the transcript + the completion
+signal), not here. Delivery is **observed, not pushed** (like provisioning): the host daemon spots a
+task carrying an undelivered ask over its work-pull loop (the ``pending_ask_id`` the task service
+overlays on the task) and calls :meth:`deliver`. Idempotent + self-gating, so the loop can call it on
+every task each pass. LLM-free.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from panopticon.client import JsonObj, TaskServiceClient
+from panopticon.core.asking import compose_ask_message
+from panopticon.core.state import TERMINAL_LABELS
+from panopticon.sessionservice.local_runner import LocalRunner
+from panopticon.sessionservice.spawner import Spawner
+
+_log = logging.getLogger(__name__)
+
+
+class AskWorker:
+    """Delivers each task's pending ask to its agent — live-inject or ``--continue`` resume."""
+
+    def __init__(
+        self,
+        client: TaskServiceClient,
+        runner: LocalRunner,
+        spawner: Spawner,
+        *,
+        runner_id: str,
+    ) -> None:
+        self._client = client
+        self._runner = runner
+        self._spawner = spawner
+        self._runner_id = runner_id
+
+    def deliver(self, task: JsonObj) -> str | None:
+        """Deliver ``task``'s undelivered ask, if any, returning the ask id (else ``None``).
+
+        No-ops unless the task carries a ``pending_ask_id`` (self-gating, so the daemon can call this
+        on every task each pass — once delivered/gone the field clears). Only the host that **owns**
+        the task (or an unclaimed one) delivers, since the tmux session and config volume are
+        host-local; a task claimed by another runner is left for that host.
+        """
+        ask_id: str | None = task.get("pending_ask_id")
+        if not ask_id:
+            return None
+        if task.get("claimed_by") not in (None, self._runner_id):
+            return None  # another host owns it — its session/volume live there, so it delivers
+        task_id = task["id"]
+        if not self._runner.config_volume_exists(task_id):
+            # The claude session was reaped — the agent can't be resumed. Mark it gone (→ 410).
+            self._client.mark_ask_gone(task_id, ask_id)
+            _log.info("task %s: ask %s undeliverable (config volume gone)", task_id, ask_id)
+            return None
+        ask = self._client.get_ask(task_id, ask_id)
+        message = compose_ask_message(
+            ask["question"],
+            ask.get("context") or "",
+            terminal=task["state"] in TERMINAL_LABELS,
+            ask_id=ask_id,
+        )
+        if self._runner.has_session(task_id):
+            self._runner.send_to_session(task_id, message)  # inject into the live agent's pane
+            _log.info("task %s: ask %s injected into live session", task_id, ask_id)
+        elif self._spawner.spawn_for_ask(task, message) is None:
+            return None  # couldn't claim it (another host won the race) — retry next pass
+        else:
+            _log.info("task %s: ask %s delivered via --continue resume", task_id, ask_id)
+        self._client.mark_ask_delivered(task_id, ask_id)
+        return ask_id
diff --git a/src/panopticon/sessionservice/host.py b/src/panopticon/sessionservice/host.py
index 4ca1c80d..01ef05a7 100644
--- a/src/panopticon/sessionservice/host.py
+++ b/src/panopticon/sessionservice/host.py
@@ -36,6 +36,7 @@
 from panopticon.core.dirs import CLONE_CACHE_DIR, TASKS_DIR
 from panopticon.core.git import GitClones
 from panopticon.sessionservice._migration import migrate_session_dirs
+from panopticon.sessionservice.ask_worker import AskWorker
 from panopticon.sessionservice.clones import CloneCache
 from panopticon.sessionservice.executions import WorkflowExecutions
 from panopticon.sessionservice.images import ImageBuilder
@@ -55,6 +56,7 @@ def __init__(
         client: TaskServiceClient,
         spawner: Spawner,
         provisioner: Provisioner,
+        ask_worker: AskWorker,
         *,
         sleep: Callable[[float], None] = time.sleep,
         interval: float = 2.0,
@@ -62,14 +64,15 @@ def __init__(
         self._client = client
         self._spawner = spawner
         self._provisioner = provisioner
+        self._ask_worker = ask_worker
         self._sleep = sleep
         self._interval = interval
 
     def tick(self, tasks: list[JsonObj]) -> None:
         """One pass over a task snapshot: spawn each spawnable task, provision each slugged one,
-        reconcile each claimed one's container-lifecycle status (down-detection), and heal each
-        orphan (a claimed task whose tmux session is gone → respawn). All self-gate, so re-running
-        over an unchanged snapshot is a no-op.
+        deliver each pending ask (ask-the-author), reconcile each claimed one's container-lifecycle
+        status (down-detection), and heal each orphan (a claimed task whose tmux session is gone →
+        respawn). All self-gate, so re-running over an unchanged snapshot is a no-op.
 
         A cheap REST-only **pre-pass flags every orphan ``healing`` first**, before any respawn. The
         respawn loop below is serial (each :meth:`Spawner.heal` blocks on ``docker run`` + the tmux
@@ -85,6 +88,10 @@ def tick(self, tasks: list[JsonObj]) -> None:
             try:
                 self._spawner.spawn_one(task)
                 self._provisioner.provision(task)
+                # Deliver a pending ask before heal: a parked task with a question is resumed with
+                # the ask as its prompt (spawn_for_ask), so heal then sees a live session and skips it
+                # rather than respawning it a second time with the generic INTERRUPT_PROMPT.
+                self._ask_worker.deliver(task)
                 self._spawner.reconcile(task)
                 self._spawner.heal(task)
                 self._spawner.cleanup(task)
@@ -191,7 +198,10 @@ def run_host(
         makedirs=makedirs,
     )
     provisioner = Provisioner(client, clones_root=tasks_root, git=git, executions=executions)
-    HostDaemon(client, spawner, provisioner, interval=interval, sleep=sleep).run(until=until)
+    ask_worker = AskWorker(client, runner, spawner, runner_id=runner_id)
+    HostDaemon(client, spawner, provisioner, ask_worker, interval=interval, sleep=sleep).run(
+        until=until
+    )
 
 
 def build_arg_parser() -> argparse.ArgumentParser:
diff --git a/src/panopticon/sessionservice/local_runner.py b/src/panopticon/sessionservice/local_runner.py
index 7209e76f..0a5be754 100644
--- a/src/panopticon/sessionservice/local_runner.py
+++ b/src/panopticon/sessionservice/local_runner.py
@@ -59,6 +59,13 @@ def session_name(task_id: str) -> str:
 CONFIG_MOUNT = "/home/panopticon/.claude"
 
 
+def config_volume_name(task_id: str) -> str:
+    """The Docker named volume holding a task's claude session (its transcripts). Persists across
+    respawn/recreate so the agent resumes via ``--continue``; when it's reaped an ask can't be
+    delivered (see :meth:`LocalRunner.config_volume_exists`)."""
+    return f"panopticon-config-{task_id}"
+
+
 class CommandRunner(Protocol):
     """Runs an external command and returns its stdout; ``check`` raises on non-zero exit.
 
@@ -145,6 +152,7 @@ def spawn(
         initial_prompt: str | None = None,
         turn: str | None = None,
         starting_model: str | None = None,
+        ask_prompt: str | None = None,
         progress: Callable[[LifecyclePhase], None] | None = None,
     ) -> str:
         """Spawn the task container. ``env_file`` is the task's repo's secret reference (ADR
@@ -194,6 +202,10 @@ def _report(phase: LifecyclePhase) -> None:
             env["PANOPTICON_TASK_TURN"] = turn
         if starting_model:
             env["PANOPTICON_STARTING_MODEL"] = starting_model
+        if ask_prompt:
+            # A parked/terminal task resumed to answer a reviewer's ask: the agent launcher appends
+            # this as the positional prompt on a ``--continue`` session (ask-the-author).
+            env["PANOPTICON_ASK_PROMPT"] = ask_prompt
         docker_run = [
             "docker",
             "run",
@@ -222,7 +234,7 @@ def _report(phase: LifecyclePhase) -> None:
             ]
         # Per-task config volume: persists claude's session history across respawn/recreate (the
         # transcripts live in the config dir, which is otherwise thrown away with the container).
-        docker_run += ["--volume", f"panopticon-config-{task_id}:{CONFIG_MOUNT}"]
+        docker_run += ["--volume", f"{config_volume_name(task_id)}:{CONFIG_MOUNT}"]
         for key, value in env.items():
             docker_run += ["--env", f"{key}={value}"]
         docker_run.append(
@@ -289,6 +301,30 @@ def has_session(self, task_id: str) -> bool:
         sessions = self._run(self._tmux("list-sessions", "-F", "#{session_name}"), check=False)
         return session in sessions.splitlines()
 
+    def config_volume_exists(self, task_id: str) -> bool:
+        """Whether the task's per-task config volume (its claude session) still exists on this host.
+
+        ``docker volume inspect`` the named volume; empty output (a nonzero exit) means it was reaped
+        — the agent's session can't be resumed, so an ask is undeliverable. The ask worker uses this
+        to mark the ask ``gone`` (the API then returns 410, and the review tool falls back)."""
+        name = config_volume_name(task_id)
+        out = self._run(["docker", "volume", "inspect", "--format", "{{.Name}}", name], check=False)
+        return name in out.splitlines()
+
+    def send_to_session(self, task_id: str, text: str) -> None:
+        """Deliver ``text`` to the task's **live** claude session as a submitted user message.
+
+        The first tmux input path in the repo: set a per-task paste buffer to the text (passed as an
+        argument, so multi-line content stays intact), paste it into the pane with bracketed paste
+        (``-p``, so claude receives it as one pasted block rather than executing line-by-line) and
+        delete the buffer (``-d``), then send ``Enter`` to submit. Used to inject a reviewer's ask
+        into a running agent (ask-the-author); the caller checks :meth:`has_session` first."""
+        session = session_name(task_id)
+        buffer = f"panopticon-ask-{task_id}"
+        self._run(self._tmux("set-buffer", "-b", buffer, text))
+        self._run(self._tmux("paste-buffer", "-b", buffer, "-t", session, "-p", "-d"))
+        self._run(self._tmux("send-keys", "-t", session, "Enter"))
+
     def delete_workspace_contents(self, path: str) -> None:
         """Delete all files inside ``path`` by running a throwaway root Docker container.
 
diff --git a/src/panopticon/sessionservice/spawner.py b/src/panopticon/sessionservice/spawner.py
index f0d4b579..aec01092 100644
--- a/src/panopticon/sessionservice/spawner.py
+++ b/src/panopticon/sessionservice/spawner.py
@@ -157,12 +157,17 @@ def spawn_one(self, task: JsonObj) -> str | None:
             raise
         return self._spawn(task)
 
-    def _spawn(self, task: JsonObj) -> str:
+    def _spawn(self, task: JsonObj, *, ask_prompt: str | None = None) -> str:
         """Spawn the execution backend for an **already claimed** task — the body shared by
-        :meth:`spawn_one` (after it wins the claim) and :meth:`heal` (respawning an orphan this
-        runner already holds). Routes on the workflow's ``runner_type``: a ``"shell"`` workflow runs
+        :meth:`spawn_one` (after it wins the claim), :meth:`heal` (respawning an orphan this
+        runner already holds), and :meth:`spawn_for_ask` (resuming a parked agent to answer a
+        reviewer's question). Routes on the workflow's ``runner_type``: a ``"shell"`` workflow runs
         its script in a host tmux session (no clone, no image); otherwise the Docker container path.
 
+        ``ask_prompt`` (ask-the-author) is passed to the container so the resumed agent processes the
+        reviewer's question as its ``--continue`` prompt; it is container-only (a shell task has no
+        agent to ask).
+
         Reports each phase (``CLAIMING`` → … → ``AWAITING``); a step raising is reported as
         ``FAILED`` (with the error) before re-raising, so the host daemon's per-task isolation still
         applies but the failure is visible, not silent."""
@@ -173,11 +178,32 @@ def _spawn(self, task: JsonObj) -> str:
             repo = self._client.get_repo(task["repo_id"])
             if self._executions.is_shell(task["workflow"]):
                 return self._spawn_shell(task, repo)
-            return self._spawn_container(task, repo)
+            return self._spawn_container(task, repo, ask_prompt=ask_prompt)
         except Exception as exc:
             self._report(task_id, LifecyclePhase.FAILED, detail=str(exc))
             raise
 
+    def spawn_for_ask(self, task: JsonObj, ask_prompt: str) -> str | None:
+        """Resume a **parked** task's agent to answer a reviewer's ask (ask-the-author).
+
+        Unlike :meth:`spawn_one`, this deliberately allows a **terminal** task: asking a COMPLETE
+        task's author is a first-class case (the config volume — its claude session — persists, so
+        ``--continue`` resumes it). Claims the task for this host if it's unclaimed (compare-and-set;
+        a 409 means another host owns it → returns ``None`` and the ask stays pending for that host),
+        then spawns with the ask as the resume prompt. Returns the container id, or ``None`` if it
+        couldn't claim. The container answers, its Stop hook records the reply, and (for a terminal
+        task) :meth:`cleanup` reaps it once it exits — which it won't do while it's still running."""
+        if task.get("claimed_by") not in (None, self._runner_id):
+            return None  # another host owns it — leave the ask pending; that host will deliver it
+        if not task.get("claimed_by"):
+            try:
+                self._client.claim(task["id"], self._runner_id)
+            except httpx.HTTPStatusError as exc:
+                if exc.response.status_code == 409:
+                    return None
+                raise
+        return self._spawn(task, ask_prompt=ask_prompt)
+
     def _prepare_task_dir(self, task: JsonObj, repo: JsonObj, *, clone: bool) -> str:
         """The task's working directory (``<tasks_root>/<task_id>``) — shared by both backends.
 
@@ -205,9 +231,14 @@ def _prepare_task_dir(self, task: JsonObj, repo: JsonObj, *, clone: bool) -> str
         self._makedirs(workdir)
         return workdir
 
-    def _spawn_container(self, task: JsonObj, repo: JsonObj) -> str:
+    def _spawn_container(
+        self, task: JsonObj, repo: JsonObj, *, ask_prompt: str | None = None
+    ) -> str:
         """The Docker path: clone the per-task workspace, compose base → workflow → repo, and spawn
-        the container (reports ``PREPARING`` → ``BUILDING`` → ``STARTING`` → ``AWAITING``)."""
+        the container (reports ``PREPARING`` → ``BUILDING`` → ``STARTING`` → ``AWAITING``).
+
+        ``ask_prompt`` (ask-the-author), when set, is passed to the runner so the resumed agent
+        answers a reviewer's question as its ``--continue`` prompt."""
         task_id = task["id"]
         workspace = self._prepare_task_dir(
             task, repo, clone=True
@@ -236,6 +267,7 @@ def _spawn_container(self, task: JsonObj, repo: JsonObj) -> str:
             starting_model=task.get(
                 "starting_model"
             ),  # model selection passed to claude --model on first launch
+            ask_prompt=ask_prompt,  # ask-the-author: the resumed agent's --continue prompt
             progress=lambda phase: self._report(task_id, phase),  # STARTING then AWAITING
         )
 
diff --git a/src/panopticon/taskservice/api.py b/src/panopticon/taskservice/api.py
index a4ab857d..7ba7488d 100644
--- a/src/panopticon/taskservice/api.py
+++ b/src/panopticon/taskservice/api.py
@@ -19,11 +19,13 @@
 from pydantic import BaseModel, ConfigDict, Field
 
 from panopticon.core.artifacts import ArtifactError
-from panopticon.core.models import Actor, LifecyclePhase, Repo, Status, Task
+from panopticon.core.models import Actor, Ask, AskStatus, LifecyclePhase, Repo, Status, Task
 from panopticon.core.store import AlreadyExists, NotFound, StoreError
 from panopticon.core.workflow import IllegalTransition, InvalidWorkflow, ResponsibilitiesNotMet
 from panopticon.taskservice.service import (
     AlreadyClaimed,
+    AskGone,
+    AskInProgress,
     NotAuthorized,
     TaskService,
     UnknownWorkflow,
@@ -94,6 +96,9 @@ class TaskSummaryOut(BaseModel):
     runner_host: str | None = (
         None  # hostname the claiming runner registered with (M5: remote attach)
     )
+    pending_ask_id: str | None = (
+        None  # id of an undelivered ask-the-author question; the host daemon's ask worker delivers it
+    )
 
 
 class TaskOut(BaseModel):
@@ -144,6 +149,9 @@ class TaskOut(BaseModel):
     runner_host: str | None = (
         None  # hostname the claiming runner registered with (M5: remote attach)
     )
+    pending_ask_id: str | None = (
+        None  # id of an undelivered ask-the-author question; the host daemon's ask worker delivers it
+    )
     history: list[HistoryOut]
 
 
@@ -258,6 +266,41 @@ class ProvisioningIn(BaseModel):
     clone: str
 
 
+class AskIn(BaseModel):
+    """A reviewer's question for a task's implementing agent (ask-the-author)."""
+
+    question: str
+    context: str = ""
+
+
+class AskCreatedOut(BaseModel):
+    """The id a reviewer polls with after posting an ask."""
+
+    ask_id: str
+
+
+class AskOut(BaseModel):
+    """An ask's public shape: ``status`` (pending/answered) and the ``answer`` once available. The
+    internal ``delivered`` status maps to ``pending`` on the wire — the review tool only distinguishes
+    "still working" from "answered". ``question``/``context`` are echoed for the session service."""
+
+    ask_id: str
+    status: str
+    answer: str | None = None
+    question: str
+    context: str
+
+
+class OutstandingAskOut(BaseModel):
+    """The task's current unanswered ask (or ``ask_id=None``) — the container Stop hook reads this."""
+
+    ask_id: str | None = None
+
+
+class AskAnswerIn(BaseModel):
+    answer: str
+
+
 class SkillOut(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
@@ -381,6 +424,7 @@ def _task_out(task: Task) -> TaskOut:
         out.lifecycle_detail = lifecycle.detail if lifecycle is not None else None
         if task.claimed_by is not None:
             out.runner_host = service.runner_host(task.claimed_by)
+        out.pending_ask_id = service.pending_ask_id(task.id)
         return out
 
     def _task_summary_out(task: Task) -> TaskSummaryOut:
@@ -391,6 +435,7 @@ def _task_summary_out(task: Task) -> TaskSummaryOut:
         out.lifecycle_detail = lifecycle.detail if lifecycle is not None else None
         if task.claimed_by is not None:
             out.runner_host = service.runner_host(task.claimed_by)
+        out.pending_ask_id = service.pending_ask_id(task.id)
         return out
 
     # -- error mapping: domain exceptions -> HTTP status --------------------------
@@ -415,6 +460,16 @@ async def _responsibilities(_: Request, exc: ResponsibilitiesNotMet) -> JSONResp
     async def _not_authorized(_: Request, exc: NotAuthorized) -> JSONResponse:
         return JSONResponse(status_code=403, content={"detail": str(exc)})
 
+    @app.exception_handler(AskInProgress)
+    async def _ask_in_progress(_: Request, exc: AskInProgress) -> JSONResponse:
+        return JSONResponse(status_code=409, content={"detail": str(exc)})
+
+    @app.exception_handler(AskGone)
+    async def _ask_gone(_: Request, exc: AskGone) -> JSONResponse:
+        # The task's config volume was reaped — the agent can't be resumed. The review tool has a
+        # documented fallback for this (the memo's guardrail); 410 Gone is the clear signal.
+        return JSONResponse(status_code=410, content={"detail": str(exc)})
+
     @app.exception_handler(UnknownWorkflow)
     async def _unknown_wf(_: Request, exc: UnknownWorkflow) -> JSONResponse:
         return JSONResponse(status_code=400, content={"detail": str(exc)})
@@ -543,6 +598,21 @@ async def list_tasks(
         response.headers[TASKS_VERSION_HEADER] = str(version)
         return tasks
 
+    @app.get("/tasks/lookup")
+    async def lookup_task(
+        repo_id: str | None = Query(default=None),
+        branch: str | None = Query(default=None),
+        url: str | None = Query(default=None),
+    ) -> TaskOut:
+        """Find the task working a branch (``?repo_id=&branch=``) or a PR/URL (``?url=``); 404 if none.
+        Declared before ``/tasks/{task_id}`` so ``lookup`` isn't captured as a task id. The review
+        tool (ask-the-author) uses this to resolve a task from what it's reviewing."""
+        try:
+            task = await service.lookup_task(repo_id=repo_id, branch=branch, url=url)
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+        return _task_out(task)
+
     @app.get("/tasks/{task_id}")
     async def get_task(task_id: str) -> TaskOut:
         return _task_out(await service.get_task(task_id))
@@ -659,6 +729,61 @@ async def record_provisioning(task_id: str, body: ProvisioningIn) -> TaskOut:
             raise HTTPException(status_code=400, detail=str(exc)) from exc
         return _task_out(task)
 
+    # -- asks (ask-the-author) ----------------------------------------------------
+    #
+    # The review tool posts a question, then polls for the answer; the host daemon's ask worker
+    # delivers it to the task's claude session and the container Stop hook records the reply. An ask
+    # never transitions the task — it's conversation. Delivery/gone/answer are recorded by the
+    # session service + container (same-host trust), the poll is the review tool's.
+
+    def _ask_out(ask: Ask) -> AskOut:
+        # `delivered` is an internal step; the review tool only cares pending-vs-answered.
+        status = "pending" if ask.status is AskStatus.DELIVERED else ask.status.value
+        return AskOut(
+            ask_id=ask.id,
+            status=status,
+            answer=ask.answer,
+            question=ask.question,
+            context=ask.context,
+        )
+
+    @app.post("/tasks/{task_id}/ask", status_code=201)
+    async def create_ask(task_id: str, body: AskIn) -> AskCreatedOut:
+        """Post a reviewer's question to a task's agent. 409 if an unanswered ask already exists
+        (cap: 1 per task). Returns the ``ask_id`` to poll ``GET …/ask/{ask_id}`` with."""
+        ask = await service.create_ask(task_id, body.question, body.context)
+        return AskCreatedOut(ask_id=ask.id)
+
+    @app.get("/tasks/{task_id}/ask")
+    async def outstanding_ask(task_id: str) -> OutstandingAskOut:
+        """The task's current unanswered ask id (or null) — the container Stop hook reads this to
+        know whether the reply it's about to finish should be recorded."""
+        await service.get_task(task_id)  # 404 if the task is unknown
+        ask = service.outstanding_ask(task_id)
+        return OutstandingAskOut(ask_id=ask.id if ask is not None else None)
+
+    @app.get("/tasks/{task_id}/ask/{ask_id}")
+    async def get_ask(task_id: str, ask_id: str) -> AskOut:
+        """Poll an ask: ``{status: pending|answered, answer}`` (plus the echoed question/context).
+        410 if the task's config volume was reaped (undeliverable)."""
+        return _ask_out(service.get_ask(task_id, ask_id))
+
+    @app.post("/tasks/{task_id}/ask/{ask_id}/delivered")
+    async def mark_ask_delivered(task_id: str, ask_id: str) -> AskOut:
+        """The session service reports it delivered the ask to the agent (tmux / --continue)."""
+        return _ask_out(service.mark_ask_delivered(task_id, ask_id))
+
+    @app.post("/tasks/{task_id}/ask/{ask_id}/gone")
+    async def mark_ask_gone(task_id: str, ask_id: str) -> OutstandingAskOut:
+        """The session service reports the config volume is gone — the ask is undeliverable (→ 410)."""
+        ask = service.mark_ask_gone(task_id, ask_id)
+        return OutstandingAskOut(ask_id=ask.id)
+
+    @app.post("/tasks/{task_id}/ask/{ask_id}/answer")
+    async def record_ask_answer(task_id: str, ask_id: str, body: AskAnswerIn) -> AskOut:
+        """The container Stop hook records the agent's reply, extracted from the transcript."""
+        return _ask_out(service.record_ask_answer(task_id, ask_id, body.answer))
+
     # -- artifacts ----------------------------------------------------------------
 
     @app.put("/tasks/{task_id}/artifacts/{name}", status_code=204)
diff --git a/src/panopticon/taskservice/service.py b/src/panopticon/taskservice/service.py
index bd9bf023..61ad3724 100644
--- a/src/panopticon/taskservice/service.py
+++ b/src/panopticon/taskservice/service.py
@@ -24,6 +24,8 @@
 from panopticon.core.layers import LayerStore
 from panopticon.core.models import (
     Actor,
+    Ask,
+    AskStatus,
     ContainerStatus,
     LifecyclePhase,
     Repo,
@@ -61,6 +63,14 @@ class NotAuthorized(Exception):
     non-orchestration workflow trying to create other tasks)."""
 
 
+class AskInProgress(Exception):
+    """Raised when creating an ask for a task that already has an unanswered one (cap: 1 per task)."""
+
+
+class AskGone(Exception):
+    """Raised when reading an ask whose task's config volume was reaped — undeliverable (→ 410)."""
+
+
 @dataclass
 class Registration:
     """An active container's claim that it is working on a task (liveness).
@@ -131,6 +141,10 @@ def __init__(
         self._registrations: dict[str, Registration] = {}
         self._runner_registrations: dict[str, RunnerRegistration] = {}
         self._lifecycles: dict[str, ContainerLifecycle] = {}
+        #: Ephemeral ask-the-author records (question → delivery → answer), keyed by ask id. Held in
+        #: memory like registrations/lifecycles — a review-time conversation, not stored task state,
+        #: so it never bumps the store version (ephemeral changes wake the feed via ``_notify_change``).
+        self._asks: dict[str, Ask] = {}
         # Ephemeral liveness (registrations, runner liveness, lifecycle phases) lives outside the
         # store, so it doesn't bump the store's version. But the dashboard's change-feed long-poll
         # only wakes on a version change — so a container going live or a phase advancing wouldn't
@@ -688,6 +702,120 @@ async def record_provisioning(self, task_id: str, *, branch: str, clone: str) ->
         _log.info("task %s: provisioned (branch=%s)", task_id, branch)
         return task
 
+    # -- asks (ask-the-author: a reviewer interrogates a task's agent) ---------------------
+    #
+    # Ephemeral like a registration/lifecycle: a review-time question delivered to the task's claude
+    # session and its answer, held in memory (:attr:`_asks`) — never a workflow transition, so it
+    # neither changes state nor seeds responsibilities. The **session service** does the delivery
+    # (tmux inject / ``--continue`` resume) and the container's Stop hook records the answer; the task
+    # service only tracks the record and enforces the one-unanswered-ask-per-task cap.
+
+    _UNANSWERED = frozenset({AskStatus.PENDING, AskStatus.DELIVERED})
+
+    def _task_asks(self, task_id: str) -> list[Ask]:
+        """This task's asks, oldest first (created_at is an ISO string, so lexical == chronological)."""
+        asks = [a for a in self._asks.values() if a.task_id == task_id]
+        return sorted(asks, key=lambda a: a.created_at or "")
+
+    def _get_ask(self, task_id: str, ask_id: str) -> Ask:
+        ask = self._asks.get(ask_id)
+        if ask is None or ask.task_id != task_id:
+            raise NotFound(f"ask {ask_id!r} does not exist for task {task_id!r}")
+        return ask
+
+    async def create_ask(self, task_id: str, question: str, context: str = "") -> Ask:
+        """Create a pending ask for a task (the session service delivers it). Enforces the cap of one
+        unanswered ask per task (raises :class:`AskInProgress`). Wakes the change feed so the host
+        daemon's ask worker picks it up."""
+        await self.get_task(task_id)  # ensure the task exists (raises NotFound)
+        if any(a.status in self._UNANSWERED for a in self._task_asks(task_id)):
+            raise AskInProgress(f"task {task_id!r} already has an unanswered ask")
+        ask = Ask(
+            id=self._id(),
+            task_id=task_id,
+            question=question,
+            context=context,
+            created_at=self._clock(),
+        )
+        self._asks[ask.id] = ask
+        self._notify_change()  # wake the host daemon's ask worker (it reads pending_ask_id)
+        _log.info("task %s: ask %s created", task_id, ask.id)
+        return ask
+
+    def get_ask(self, task_id: str, ask_id: str) -> Ask:
+        """The ask (raises :class:`NotFound` if unknown; :class:`AskGone` if its volume was reaped)."""
+        ask = self._get_ask(task_id, ask_id)
+        if ask.status is AskStatus.GONE:
+            raise AskGone(
+                f"ask {ask_id!r}: the task's container/volume is gone; the agent can't be resumed"
+            )
+        return ask
+
+    def outstanding_ask(self, task_id: str) -> Ask | None:
+        """The task's newest unanswered ask (pending or delivered), or ``None`` — what the container
+        Stop hook checks to know whether a reply it should record is being produced."""
+        unanswered = [a for a in self._task_asks(task_id) if a.status in self._UNANSWERED]
+        return unanswered[-1] if unanswered else None
+
+    def pending_ask_id(self, task_id: str) -> str | None:
+        """The id of the task's oldest **undelivered** ask, or ``None`` — overlaid on the task's
+        serialized form so the host daemon's ask worker can spot deliverable asks without a per-task
+        request (it clears the moment the worker marks the ask delivered or gone)."""
+        for ask in self._task_asks(task_id):
+            if ask.status is AskStatus.PENDING:
+                return ask.id
+        return None
+
+    def mark_ask_delivered(self, task_id: str, ask_id: str) -> Ask:
+        """Mark an ask delivered (the session service handed it to the agent); wakes the feed."""
+        ask = self._get_ask(task_id, ask_id)
+        ask.status = AskStatus.DELIVERED
+        self._notify_change()
+        _log.info("task %s: ask %s delivered", task_id, ask_id)
+        return ask
+
+    def mark_ask_gone(self, task_id: str, ask_id: str) -> Ask:
+        """Mark an ask undeliverable because the task's config volume was reaped (→ 410)."""
+        ask = self._get_ask(task_id, ask_id)
+        ask.status = AskStatus.GONE
+        self._notify_change()
+        _log.info("task %s: ask %s gone (volume reaped)", task_id, ask_id)
+        return ask
+
+    def record_ask_answer(self, task_id: str, ask_id: str, answer: str) -> Ask:
+        """Record the agent's reply (the container Stop hook extracts it from the transcript)."""
+        ask = self._get_ask(task_id, ask_id)
+        ask.answer = answer
+        ask.status = AskStatus.ANSWERED
+        self._notify_change()
+        _log.info("task %s: ask %s answered", task_id, ask_id)
+        return ask
+
+    async def lookup_task(
+        self, *, repo_id: str | None = None, branch: str | None = None, url: str | None = None
+    ) -> Task:
+        """Find the task matching a branch (with its repo) or a URL — the review tool's entry point.
+
+        Exactly one selector is expected: ``repo_id`` + ``branch``, or ``url``. Raises
+        :class:`ValueError` for a malformed request and :class:`NotFound` if nothing matches. Returns
+        the full task (history included), so the review tool gets the same shape as ``GET /tasks/{id}``.
+        """
+        if url is not None:
+            if repo_id is not None or branch is not None:
+                raise ValueError("pass either url, or repo_id + branch — not both")
+            found = await self._store.find_task_by_url(url)
+            if found is None:
+                raise NotFound(f"no task with url {url!r}")
+        elif repo_id is not None and branch is not None:
+            found = await self._store.find_task_by_branch(repo_id, branch)
+            if found is None:
+                raise NotFound(f"no task on repo {repo_id!r} with branch {branch!r}")
+        else:
+            raise ValueError("pass either url, or repo_id + branch")
+        return await self.get_task(
+            found.id
+        )  # re-read for full history (the lookup is history-less)
+
     # -- artifacts ----------------------------------------------------------------
 
     async def put_artifact(self, task_id: str, name: str, content: bytes) -> None:
diff --git a/src/panopticon/taskservice/store_sqlalchemy.py b/src/panopticon/taskservice/store_sqlalchemy.py
index b4a03f89..00122a89 100644
--- a/src/panopticon/taskservice/store_sqlalchemy.py
+++ b/src/panopticon/taskservice/store_sqlalchemy.py
@@ -368,6 +368,28 @@ async def _list_tasks_summary(self) -> list[Task]:
             )
             return [r.to_domain() for r in result.scalars()]
 
+    async def _find_task_by_branch(self, repo_id: str, branch: str) -> Task | None:
+        async with self._session() as s:
+            result = await s.execute(
+                select(_TaskRow)
+                .options(noload(_TaskRow.history))
+                .where(_TaskRow.repo_id == repo_id, _TaskRow.branch == branch)
+                .order_by(_TaskRow.created_at.desc(), _TaskRow.id)
+            )
+            row = result.scalars().first()
+            return row.to_domain() if row is not None else None
+
+    async def _find_task_by_url(self, url: str) -> Task | None:
+        async with self._session() as s:
+            result = await s.execute(
+                select(_TaskRow)
+                .options(noload(_TaskRow.history))
+                .where(_TaskRow.url == url)
+                .order_by(_TaskRow.created_at.desc(), _TaskRow.id)
+            )
+            row = result.scalars().first()
+            return row.to_domain() if row is not None else None
+
     async def _create_task(self, task: Task) -> None:
         async with self._session.begin() as s:
             if await s.get(_TaskRow, task.id) is not None:
diff --git a/tests/container/test_agent.py b/tests/container/test_agent.py
index 748075ee..4336e555 100644
--- a/tests/container/test_agent.py
+++ b/tests/container/test_agent.py
@@ -116,6 +116,30 @@ def test_claude_argv_omits_interrupt_prompt_on_respawn_for_user_turn(tmp_path: P
     assert argv == ["claude", "--dangerously-skip-permissions", "--continue"]
 
 
+def test_claude_argv_appends_ask_prompt_on_resume_over_interrupt(tmp_path: Path) -> None:
+    # ask-the-author: a parked task resumed to answer a reviewer's question gets the ask as its
+    # --continue prompt, winning over the generic INTERRUPT_PROMPT even on the agent's turn.
+    project = tmp_path / "projects" / "-work-repo"
+    project.mkdir(parents=True)
+    (project / "session.jsonl").write_text("{}")
+    argv = agent._claude_argv(
+        tmp_path, Path("/work/repo"), turn="agent", ask_prompt="a reviewer asks: why?"
+    )
+    assert argv == [
+        "claude",
+        "--dangerously-skip-permissions",
+        "--continue",
+        "a reviewer asks: why?",
+    ]
+    assert agent.INTERRUPT_PROMPT not in argv
+
+
+def test_claude_argv_uses_ask_prompt_on_first_run_when_no_session(tmp_path: Path) -> None:
+    # Defensive: if there's somehow no prior session, the ask still lands as the first message.
+    argv = agent._claude_argv(tmp_path, Path("/work/repo"), ask_prompt="why?")
+    assert argv == ["claude", "--dangerously-skip-permissions", "why?"]
+
+
 def test_write_mcp_config_points_claude_at_the_task_service_mcp(tmp_path: Path) -> None:
     import json
 
diff --git a/tests/container/test_hooks.py b/tests/container/test_hooks.py
index ae345f31..5aed50fe 100644
--- a/tests/container/test_hooks.py
+++ b/tests/container/test_hooks.py
@@ -74,6 +74,9 @@ def set_tokens_used(self, task_id: str, tokens_used: int) -> dict[str, object]:
         self.tokens.append((task_id, tokens_used))
         return {}
 
+    def outstanding_ask(self, task_id: str) -> str | None:
+        return None  # no ask outstanding in these tests → the stop hook records nothing
+
 
 def test_hook_flips_the_turn(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("PANOPTICON_SERVICE_URL", "http://svc")
@@ -289,3 +292,100 @@ def test_user_prompt_submit_unaffected_by_background_tasks(
     assert hook.main(["agent", "prompt"], client=client, stdin=io.StringIO(payload)) == 0  # type: ignore[arg-type]
     assert client.calls == [("t1", "agent")]
     assert "PHASE BRIEFING" in capsys.readouterr().out
+
+
+# -- ask-the-author: extract the agent's reply and record it on Stop ---------------------------
+
+from panopticon.core.asking import ask_marker  # noqa: E402
+
+
+class _AskClient:
+    """A Stop-hook client that reports one outstanding ask and records the answer + turn flip."""
+
+    def __init__(self, ask_id: str | None) -> None:
+        self._ask_id = ask_id
+        self.answers: list[tuple[str, str, str]] = []
+        self.turns: list[tuple[str, str]] = []
+
+    def set_turn(self, task_id: str, turn: str) -> dict[str, object]:
+        self.turns.append((task_id, turn))
+        return {}
+
+    def set_tokens_used(self, task_id: str, tokens_used: int) -> dict[str, object]:
+        return {}
+
+    def outstanding_ask(self, task_id: str) -> str | None:
+        return self._ask_id
+
+    def record_ask_answer(self, task_id: str, ask_id: str, answer: str) -> dict[str, object]:
+        self.answers.append((task_id, ask_id, answer))
+        return {}
+
+
+def _ask_transcript(tmp_path: Path, ask_id: str) -> Path:
+    """A transcript where a marked user message (the delivered ask) is followed by the agent's reply
+    (a text block, plus a tool_use block that must be ignored)."""
+    lines = [
+        {"type": "user", "message": {"role": "user", "content": "earlier unrelated turn"}},
+        {"type": "assistant", "message": {"role": "assistant", "content": "earlier answer"}},
+        {"type": "user", "message": {"role": "user", "content": f"{ask_marker(ask_id)}\nwhy?"}},
+        {
+            "type": "assistant",
+            "message": {
+                "role": "assistant",
+                "content": [
+                    {"type": "text", "text": "Because it is keyed by id."},
+                    {"type": "tool_use", "name": "Read", "input": {}},
+                ],
+            },
+        },
+    ]
+    path = tmp_path / "ask.jsonl"
+    path.write_text("\n".join(json.dumps(x) for x in lines))
+    return path
+
+
+def test_extract_answer_pulls_assistant_text_after_the_marker(tmp_path: Path) -> None:
+    path = _ask_transcript(tmp_path, "ask1")
+    assert hook.extract_answer(str(path), ask_marker("ask1")) == "Because it is keyed by id."
+
+
+def test_extract_answer_stops_at_the_next_user_turn(tmp_path: Path) -> None:
+    lines = [
+        {"type": "user", "message": {"role": "user", "content": f"{ask_marker('a')} q"}},
+        {"type": "assistant", "message": {"role": "assistant", "content": "the reply"}},
+        {"type": "user", "message": {"role": "user", "content": "a later question"}},
+        {"type": "assistant", "message": {"role": "assistant", "content": "unrelated"}},
+    ]
+    path = tmp_path / "t.jsonl"
+    path.write_text("\n".join(json.dumps(x) for x in lines))
+    assert hook.extract_answer(str(path), ask_marker("a")) == "the reply"
+
+
+def test_extract_answer_none_when_marker_absent(tmp_path: Path) -> None:
+    path = _ask_transcript(tmp_path, "ask1")
+    assert hook.extract_answer(str(path), ask_marker("other")) is None
+    assert hook.extract_answer("/no/such/file.jsonl", ask_marker("ask1")) is None
+
+
+def test_stop_hook_records_the_answer_when_an_ask_is_outstanding(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setenv("PANOPTICON_SERVICE_URL", "http://svc")
+    monkeypatch.setenv("PANOPTICON_TASK_ID", "t1")
+    client = _AskClient(ask_id="ask1")
+    stdin = io.StringIO(json.dumps({"transcript_path": str(_ask_transcript(tmp_path, "ask1"))}))
+    assert hook.main(["user", "stop"], client=client, stdin=stdin) == 0  # type: ignore[arg-type]
+    assert client.answers == [("t1", "ask1", "Because it is keyed by id.")]
+    assert client.turns == [("t1", "user")]  # the turn still flips as normal after answering
+
+
+def test_stop_hook_records_nothing_when_no_ask_is_outstanding(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setenv("PANOPTICON_SERVICE_URL", "http://svc")
+    monkeypatch.setenv("PANOPTICON_TASK_ID", "t1")
+    client = _AskClient(ask_id=None)
+    stdin = io.StringIO(json.dumps({"transcript_path": str(_ask_transcript(tmp_path, "ask1"))}))
+    assert hook.main(["user", "stop"], client=client, stdin=stdin) == 0  # type: ignore[arg-type]
+    assert client.answers == [] and client.turns == [("t1", "user")]
diff --git a/tests/core/test_asking.py b/tests/core/test_asking.py
new file mode 100644
index 00000000..be82b304
--- /dev/null
+++ b/tests/core/test_asking.py
@@ -0,0 +1,38 @@
+"""The pure ask-message composition (ask-the-author): marker + framing, terminal vs in-flight."""
+
+from __future__ import annotations
+
+from panopticon.core.asking import ASK_MARKER_PREFIX, ask_marker, compose_ask_message
+
+
+def test_ask_marker_embeds_the_id() -> None:
+    marker = ask_marker("abc123")
+    assert marker == f"[[{ASK_MARKER_PREFIX}:abc123]]"
+    # Distinct ids give distinct markers so the Stop hook anchors the right reply.
+    assert ask_marker("abc123") != ask_marker("def456")
+
+
+def test_compose_includes_marker_question_and_context() -> None:
+    msg = compose_ask_message("  why a dict?  ", "  see models.py  ", terminal=False, ask_id="x1")
+    assert msg.startswith(ask_marker("x1"))  # marker first, so the hook can find the reply
+    assert "why a dict?" in msg and "see models.py" in msg
+    # Whitespace around the question/context is trimmed.
+    assert "  why a dict?  " not in msg
+
+
+def test_compose_omits_context_when_blank() -> None:
+    msg = compose_ask_message("why?", "   ", terminal=False, ask_id="x1")
+    assert "Context from the reviewer" not in msg
+
+
+def test_terminal_message_carries_the_readonly_guardrail() -> None:
+    msg = compose_ask_message("why?", "", terminal=True, ask_id="x1")
+    assert "merged or proposed work" in msg
+    assert "not a request to change anything" in msg
+    assert "Do not modify files" in msg
+
+
+def test_non_terminal_message_omits_the_hard_readonly_ban() -> None:
+    msg = compose_ask_message("why?", "", terminal=False, ask_id="x1")
+    assert "Do not modify files" not in msg
+    assert "asking you a question" in msg
diff --git a/tests/sessionservice/test_ask_worker.py b/tests/sessionservice/test_ask_worker.py
new file mode 100644
index 00000000..d6a14e05
--- /dev/null
+++ b/tests/sessionservice/test_ask_worker.py
@@ -0,0 +1,161 @@
+"""The host-side ask worker (ask-the-author): delivers a task's pending question to its agent.
+
+Unit tests pin the delivery decision — inject into a live session, resume a parked/terminal one via
+``--continue``, mark a reaped-volume ask gone, and no-op an undelivered-free or other-host task — with
+fake runner/spawner and the real task service over REST (so the recorded ask status is authoritative).
+An integration test proves the headline path end to end: create a task, ask, observe delivery, then
+(standing in for the container Stop hook) record the answer and retrieve it. No Docker, no LLM.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from panopticon.client import JsonObj, TaskServiceClient
+from panopticon.core.asking import ask_marker
+from panopticon.core.models import Repo
+from panopticon.sessionservice.ask_worker import AskWorker
+from panopticon.taskservice.api import create_app
+from panopticon.taskservice.artifacts_fs import FilesystemArtifactStore
+from panopticon.taskservice.service import TaskService
+from panopticon.taskservice.store_sqlalchemy import SqlAlchemyStore
+from panopticon.workflows import Spike
+
+
+class _FakeRunner:
+    """Records live-inject calls; volume/session presence are configurable per test."""
+
+    def __init__(self, *, volume: bool = True, session: bool = False) -> None:
+        self.volume = volume
+        self.session = session
+        self.sent: list[tuple[str, str]] = []
+
+    def config_volume_exists(self, task_id: str) -> bool:
+        return self.volume
+
+    def has_session(self, task_id: str) -> bool:
+        return self.session
+
+    def send_to_session(self, task_id: str, text: str) -> None:
+        self.sent.append((task_id, text))
+
+
+class _FakeSpawner:
+    """Records spawn_for_ask (parked/terminal resume) calls; the claim result is configurable."""
+
+    def __init__(self, *, result: str | None = "panopticon-c1") -> None:
+        self.result = result
+        self.calls: list[tuple[str, str]] = []
+
+    def spawn_for_ask(self, task: JsonObj, ask_prompt: str) -> str | None:
+        self.calls.append((task["id"], ask_prompt))
+        return self.result
+
+
+@pytest.fixture
+def client(tmp_path: Path) -> Iterator[TaskServiceClient]:
+    service = TaskService(SqlAlchemyStore(), {"spike": Spike()}, FilesystemArtifactStore(tmp_path))
+    asyncio.run(service.init())
+    asyncio.run(service.create_repo(Repo(id="r1", name="acme/widgets", git_url="https://x/r1.git")))
+    with TestClient(create_app(service)) as http:
+        yield TaskServiceClient(http)
+
+
+def _task_with_ask(client: TaskServiceClient, *, question: str = "why?") -> tuple[JsonObj, str]:
+    """Create a task, post an ask, and return the task snapshot (carrying pending_ask_id) + ask id."""
+    task_id = client.create_task("r1", "spike")["id"]
+    ask_id = client.create_ask(task_id, question)
+    return client.get_task(task_id), ask_id
+
+
+def _worker(client: TaskServiceClient, runner: _FakeRunner, spawner: _FakeSpawner) -> AskWorker:
+    return AskWorker(client, runner, spawner, runner_id="local")  # type: ignore[arg-type]
+
+
+def test_deliver_injects_into_a_live_session(client: TaskServiceClient) -> None:
+    task, ask_id = _task_with_ask(client, question="why a dict?")
+    runner, spawner = _FakeRunner(volume=True, session=True), _FakeSpawner()
+    assert _worker(client, runner, spawner).deliver(task) == ask_id
+    # Injected into the live pane (not respawned), and the message carries the marker + question.
+    assert len(runner.sent) == 1 and spawner.calls == []
+    _, message = runner.sent[0]
+    assert ask_marker(ask_id) in message and "why a dict?" in message
+    # The ask is now delivered → no longer the task's pending ask.
+    assert client.get_task(task["id"])["pending_ask_id"] is None
+    assert client.get_ask(task["id"], ask_id)["status"] == "pending"  # delivered reads as pending
+
+
+def test_deliver_resumes_a_parked_task_via_continue(client: TaskServiceClient) -> None:
+    task, ask_id = _task_with_ask(client)
+    runner, spawner = _FakeRunner(volume=True, session=False), _FakeSpawner()
+    assert _worker(client, runner, spawner).deliver(task) == ask_id
+    # No live session → resumed with the ask as the --continue prompt; nothing injected.
+    assert runner.sent == [] and len(spawner.calls) == 1
+    assert spawner.calls[0][0] == task["id"] and ask_marker(ask_id) in spawner.calls[0][1]
+    assert client.get_task(task["id"])["pending_ask_id"] is None
+
+
+def test_deliver_to_a_complete_task_uses_the_readonly_guardrail(client: TaskServiceClient) -> None:
+    task_id = client.create_task("r1", "spike")["id"]
+    client.set_state(task_id, "COMPLETE")  # terminal — asking its author must still work
+    ask_id = client.create_ask(task_id, "what changed?")
+    task = client.get_task(task_id)
+    runner, spawner = _FakeRunner(volume=True, session=False), _FakeSpawner()
+
+    assert _worker(client, runner, spawner).deliver(task) == ask_id
+    # Resumed (spawn_for_ask allows terminal) with the strict read-only framing.
+    assert len(spawner.calls) == 1
+    message = spawner.calls[0][1]
+    assert "merged or proposed work" in message and "not a request to change anything" in message
+    assert client.get_task(task_id)["state"] == "COMPLETE"  # untouched
+
+
+def test_deliver_marks_gone_when_the_volume_is_reaped(client: TaskServiceClient) -> None:
+    task, ask_id = _task_with_ask(client)
+    runner, spawner = _FakeRunner(volume=False), _FakeSpawner()
+    assert _worker(client, runner, spawner).deliver(task) is None
+    assert runner.sent == [] and spawner.calls == []
+    # Marked gone → the poll returns 410.
+    assert client._http.get(f"/tasks/{task['id']}/ask/{ask_id}").status_code == 410
+
+
+def test_deliver_is_a_noop_without_a_pending_ask(client: TaskServiceClient) -> None:
+    task_id = client.create_task("r1", "spike")["id"]
+    task = client.get_task(task_id)  # no ask → pending_ask_id is None
+    runner, spawner = _FakeRunner(volume=True, session=True), _FakeSpawner()
+    assert _worker(client, runner, spawner).deliver(task) is None
+    assert runner.sent == [] and spawner.calls == []
+
+
+def test_deliver_skips_a_task_owned_by_another_host(client: TaskServiceClient) -> None:
+    task, _ = _task_with_ask(client)
+    task = {**task, "claimed_by": "other-host"}  # another runner owns it — it delivers there
+    runner, spawner = _FakeRunner(volume=True, session=True), _FakeSpawner()
+    assert _worker(client, runner, spawner).deliver(task) is None
+    assert runner.sent == [] and spawner.calls == []
+
+
+def test_deliver_leaves_ask_pending_when_it_cannot_claim(client: TaskServiceClient) -> None:
+    # A parked task the worker can't claim (another host won the race) stays pending for retry.
+    task, ask_id = _task_with_ask(client)
+    runner, spawner = _FakeRunner(volume=True, session=False), _FakeSpawner(result=None)
+    assert _worker(client, runner, spawner).deliver(task) is None
+    assert client.get_task(task["id"])["pending_ask_id"] == ask_id  # still pending
+
+
+def test_end_to_end_ask_delivery_and_answer(client: TaskServiceClient) -> None:
+    # The memo's headline: create a task, ask, observe delivery, then (as the container Stop hook
+    # would) record the reply and retrieve it.
+    task, ask_id = _task_with_ask(client, question="why memoize?")
+    runner, spawner = _FakeRunner(volume=True, session=True), _FakeSpawner()
+    _worker(client, runner, spawner).deliver(task)
+
+    # The agent answered; its Stop hook records the reply extracted from the transcript.
+    client.record_ask_answer(task["id"], ask_id, "to avoid recomputation")
+    answered = client.get_ask(task["id"], ask_id)
+    assert answered["status"] == "answered" and answered["answer"] == "to avoid recomputation"
diff --git a/tests/sessionservice/test_host.py b/tests/sessionservice/test_host.py
index 226e0f35..ad8700f2 100644
--- a/tests/sessionservice/test_host.py
+++ b/tests/sessionservice/test_host.py
@@ -49,6 +49,7 @@ def spawn(
         initial_prompt: str | None = None,
         turn: str | None = None,
         starting_model: str | None = None,
+        ask_prompt: str | None = None,
         progress: object = None,
     ) -> str:
         self.spawned.append(task_id)
@@ -67,6 +68,18 @@ def delete_workspace_contents(self, path: str) -> None:
         pass
 
 
+class _AskWorker:
+    """No-op ask worker for the tick-level tests (they use no asks). Records deliver calls so a test
+    can assert the daemon runs it each pass if it wants to."""
+
+    def __init__(self) -> None:
+        self.delivered: list[str] = []
+
+    def deliver(self, task: JsonObj) -> None:
+        self.delivered.append(task["id"])
+        return None
+
+
 class _FakeImageBuilder:
     """Stands in for ImageBuilder (no docker); always reports the base image as present."""
 
@@ -117,7 +130,7 @@ class _Provisioner:
         def provision(self, task: JsonObj) -> None:
             return None
 
-    daemon = HostDaemon(_FakeClient([]), _Spawner(), _Provisioner())  # type: ignore[arg-type]
+    daemon = HostDaemon(_FakeClient([]), _Spawner(), _Provisioner(), _AskWorker())  # type: ignore[arg-type]
     daemon.tick([{"id": "t1"}, {"id": "t2"}])
     assert seen == ["t1", "t2"]  # t1's error is logged + skipped; t2 still processed
 
@@ -146,7 +159,9 @@ class _Provisioner:
         def provision(self, task: JsonObj) -> None:
             return None
 
-    HostDaemon(_FakeClient([]), _Spawner(), _Provisioner()).tick([{"id": "t1"}, {"id": "t2"}])  # type: ignore[arg-type]
+    HostDaemon(_FakeClient([]), _Spawner(), _Provisioner(), _AskWorker()).tick(
+        [{"id": "t1"}, {"id": "t2"}]
+    )  # type: ignore[arg-type]
     assert healed == ["t1", "t2"]
 
 
@@ -176,7 +191,9 @@ class _Provisioner:
         def provision(self, task: JsonObj) -> None:
             return None
 
-    HostDaemon(_FakeClient([]), _Spawner(), _Provisioner()).tick([{"id": "t1"}, {"id": "t2"}])  # type: ignore[arg-type]
+    HostDaemon(_FakeClient([]), _Spawner(), _Provisioner(), _AskWorker()).tick(
+        [{"id": "t1"}, {"id": "t2"}]
+    )  # type: ignore[arg-type]
     assert events == ["mark:t1", "mark:t2", "heal:t1", "heal:t2"]  # all marks precede any respawn
 
 
@@ -218,7 +235,7 @@ def list_tasks_versioned(
             passes.append(len(passes))
             return [{"id": f"t{len(passes)}"}], len(passes)
 
-    daemon = HostDaemon(_FeedClient(), _Spawner(), _Provisioner())  # type: ignore[arg-type]
+    daemon = HostDaemon(_FeedClient(), _Spawner(), _Provisioner(), _AskWorker())  # type: ignore[arg-type]
     daemon.run(until=lambda: len(passes) >= 3)
     assert len(reclaims) == 1  # exactly once — on the first successful fetch
     assert reclaims[0] == [{"id": "t1"}]  # the snapshot from that first fetch
@@ -263,7 +280,7 @@ def list_tasks_versioned(
             return [{"id": f"t{len(sinces)}"}], len(sinces)  # a fresh snapshot + a bumped version
 
     spawner = _Spawner()
-    daemon = HostDaemon(_FeedClient(), spawner, _Provisioner())  # type: ignore[arg-type]
+    daemon = HostDaemon(_FeedClient(), spawner, _Provisioner(), _AskWorker())  # type: ignore[arg-type]
     daemon.run(until=lambda: len(sinces) >= 3)
     assert sinces == [0, 1, 2]  # starts at 0, then each returned version becomes the next `since`
     assert spawner.seen == ["t1", "t2", "t3"]  # ticked the snapshot returned by each wake
@@ -309,7 +326,9 @@ def list_tasks_versioned(
     def until() -> bool:
         return passes["n"] >= 3  # let it wake a few times after the failure
 
-    daemon = HostDaemon(_FlakyClient(), _Spawner(), _Provisioner(), sleep=lambda _s: None)  # type: ignore[arg-type]
+    daemon = HostDaemon(
+        _FlakyClient(), _Spawner(), _Provisioner(), _AskWorker(), sleep=lambda _s: None
+    )  # type: ignore[arg-type]
     daemon.run(until=until)
     assert passes["n"] >= 3  # did not die on the first pass's error; kept going
 
@@ -458,5 +477,7 @@ class _Provisioner:
         def provision(self, task: JsonObj) -> None:
             return None
 
-    HostDaemon(_FakeClient([]), _Spawner(), _Provisioner()).tick([{"id": "t1"}, {"id": "t2"}])  # type: ignore[arg-type]
+    HostDaemon(_FakeClient([]), _Spawner(), _Provisioner(), _AskWorker()).tick(
+        [{"id": "t1"}, {"id": "t2"}]
+    )  # type: ignore[arg-type]
     assert cleaned == ["t1", "t2"]
diff --git a/tests/sessionservice/test_local_runner.py b/tests/sessionservice/test_local_runner.py
index 2e7c4735..9fbcb593 100644
--- a/tests/sessionservice/test_local_runner.py
+++ b/tests/sessionservice/test_local_runner.py
@@ -405,3 +405,88 @@ def test_cli_preps_the_workspace_then_spawns_with_secrets_and_mount(
         tmp_path / "secrets" / "r1.env"
     )  # repo's secrets
     assert f"{tasks_root}/t1:/workspace" in docker_run  # the per-task clone mounted as /workspace
+
+
+class _OutputRecorder:
+    """A CommandRunner that records calls and returns a canned stdout (for probe commands)."""
+
+    def __init__(self, output: str = "") -> None:
+        self.output = output
+        self.calls: list[list[str]] = []
+
+    def __call__(
+        self,
+        args: Sequence[str],
+        *,
+        check: bool = True,
+        interactive: bool = False,
+        verbose: bool = False,
+    ) -> str:
+        self.calls.append(list(args))
+        return self.output
+
+
+def test_spawn_passes_ask_prompt_as_env_var() -> None:
+    # ask-the-author: a parked/terminal task resumed to answer a reviewer's question carries the
+    # question as PANOPTICON_ASK_PROMPT, which the agent launcher appends to `claude --continue`.
+    rec = _Recorder()
+    runner = LocalRunner("http://svc:8000", image="img:1", run=rec)
+    runner.spawn("t1", ask_prompt="Reviewer asks: why?")
+    docker_run = next(c for c, _ in rec.calls if c[:2] == ["docker", "run"])
+    assert "PANOPTICON_ASK_PROMPT=Reviewer asks: why?" in docker_run
+
+
+def test_spawn_omits_ask_prompt_env_when_none() -> None:
+    rec = _Recorder()
+    LocalRunner("http://svc:8000", image="img:1", run=rec).spawn("t1")
+    docker_run = next(c for c, _ in rec.calls if c[:2] == ["docker", "run"])
+    assert not any(a.startswith("PANOPTICON_ASK_PROMPT=") for a in docker_run)
+
+
+def test_config_volume_exists_probes_docker_volume() -> None:
+    # Present: `docker volume inspect` echoes the volume name → True.
+    present = _OutputRecorder(output="panopticon-config-t1\n")
+    runner = LocalRunner("http://svc:8000", run=present)
+    assert runner.config_volume_exists("t1") is True
+    assert present.calls[-1] == [
+        "docker",
+        "volume",
+        "inspect",
+        "--format",
+        "{{.Name}}",
+        "panopticon-config-t1",
+    ]
+    # Reaped: empty output (nonzero exit) → False.
+    absent = _OutputRecorder(output="")
+    assert LocalRunner("http://svc:8000", run=absent).config_volume_exists("t1") is False
+
+
+def test_send_to_session_sets_buffer_pastes_then_submits() -> None:
+    rec = _Recorder()
+    runner = LocalRunner("http://svc:8000", run=rec)
+    runner.send_to_session("t1", "hello\nworld")
+    set_buffer, paste, enter = (c for c, _ in rec.calls)
+    # The multi-line text is one argv element (set-buffer takes the data as an argument).
+    assert set_buffer == [
+        "tmux",
+        "-L",
+        "panopticon",
+        "set-buffer",
+        "-b",
+        "panopticon-ask-t1",
+        "hello\nworld",
+    ]
+    # Bracketed paste (-p, so claude receives one block) into the task's pane, deleting the buffer (-d).
+    assert paste == [
+        "tmux",
+        "-L",
+        "panopticon",
+        "paste-buffer",
+        "-b",
+        "panopticon-ask-t1",
+        "-t",
+        "panopticon-t1",
+        "-p",
+        "-d",
+    ]
+    assert enter == ["tmux", "-L", "panopticon", "send-keys", "-t", "panopticon-t1", "Enter"]
diff --git a/tests/sessionservice/test_spawner.py b/tests/sessionservice/test_spawner.py
index 381f7b85..1c54a1c6 100644
--- a/tests/sessionservice/test_spawner.py
+++ b/tests/sessionservice/test_spawner.py
@@ -49,6 +49,7 @@ def spawn(
         initial_prompt: str | None = None,
         turn: str | None = None,
         starting_model: str | None = None,
+        ask_prompt: str | None = None,
         progress: Callable[[LifecyclePhase], None] | None = None,
     ) -> str:
         self.spawned.append(
@@ -59,6 +60,7 @@ def spawn(
                 "image": image,
                 "docker_in_docker": docker_in_docker,
                 "initial_prompt": initial_prompt,
+                "ask_prompt": ask_prompt,
                 "turn": turn,
                 "starting_model": starting_model,
             }
diff --git a/tests/taskservice/test_ask.py b/tests/taskservice/test_ask.py
new file mode 100644
index 00000000..e66418bb
--- /dev/null
+++ b/tests/taskservice/test_ask.py
@@ -0,0 +1,153 @@
+"""ask-the-author over REST: the tarot review tool's contract with the task service.
+
+Covers the API the memo pins: ``GET /tasks/lookup`` (resolve a task by branch/url), ``POST
+/tasks/{id}/ask`` (post a question, capped at one unanswered per task), ``GET /tasks/{id}/ask/{id}``
+(poll for the answer), the COMPLETE-without-transition guarantee, and the dead-volume → 410 fallback.
+Delivery + answer extraction happen in the session service + container; here we drive their recorded
+outcomes over the client to prove the control-plane contract. No Docker, no LLM.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Iterator
+from pathlib import Path
+
+import httpx
+import pytest
+from fastapi.testclient import TestClient
+
+from panopticon.client import TaskServiceClient
+from panopticon.core.models import Repo
+from panopticon.taskservice.api import create_app
+from panopticon.taskservice.artifacts_fs import FilesystemArtifactStore
+from panopticon.taskservice.service import TaskService
+from panopticon.taskservice.store_sqlalchemy import SqlAlchemyStore
+from panopticon.workflows import Spike
+
+
+@pytest.fixture
+def client(tmp_path: Path) -> Iterator[TaskServiceClient]:
+    service = TaskService(SqlAlchemyStore(), {"spike": Spike()}, FilesystemArtifactStore(tmp_path))
+    asyncio.run(service.init())
+    asyncio.run(service.create_repo(Repo(id="r1", name="acme/widgets", git_url="https://x/r1.git")))
+    with TestClient(create_app(service)) as http:
+        yield TaskServiceClient(http)
+
+
+def _task(client: TaskServiceClient) -> str:
+    return client.create_task("r1", "spike")["id"]
+
+
+def test_ask_then_retrieve_answer(client: TaskServiceClient) -> None:
+    # Post a question, observe it surface as the task's pending ask (what the session service's ask
+    # worker keys on), then — standing in for the session service + container Stop hook — mark it
+    # delivered and record the answer, and confirm the poll returns it.
+    task_id = _task(client)
+    ask_id = client.create_ask(task_id, "why a dict here?", context="reviewing models.py")
+
+    # Before delivery the ask is the task's pending ask and the poll reads `pending`.
+    assert client.get_task(task_id)["pending_ask_id"] == ask_id
+    pending = client.get_ask(task_id, ask_id)
+    assert pending["status"] == "pending" and pending["answer"] is None
+    assert pending["question"] == "why a dict here?" and pending["context"] == "reviewing models.py"
+
+    # The session service delivers it → no longer pending; the poll still reads `pending` (the wire
+    # only distinguishes pending vs answered — `delivered` is internal).
+    client.mark_ask_delivered(task_id, ask_id)
+    assert client.get_task(task_id)["pending_ask_id"] is None
+    assert client.get_ask(task_id, ask_id)["status"] == "pending"
+
+    # The container Stop hook records the agent's reply → the poll reads `answered` with the text.
+    client.record_ask_answer(task_id, ask_id, "because lookups are by id")
+    answered = client.get_ask(task_id, ask_id)
+    assert answered["status"] == "answered"
+    assert answered["answer"] == "because lookups are by id"
+
+
+def test_ask_is_capped_at_one_unanswered_per_task(client: TaskServiceClient) -> None:
+    task_id = _task(client)
+    client.create_ask(task_id, "first?")
+    # A second concurrent ask is rejected while the first is unanswered (cap: 1 per task).
+    resp = client._http.post(f"/tasks/{task_id}/ask", json={"question": "second?"})
+    assert resp.status_code == 409
+    # Once the first is answered, a new ask is allowed again.
+    first = client.outstanding_ask(task_id)
+    assert first is not None
+    client.record_ask_answer(task_id, first, "yes")
+    second = client.create_ask(task_id, "second?")
+    assert second != first
+
+
+def test_ask_on_a_complete_task_answers_without_transition(client: TaskServiceClient) -> None:
+    # The headline guardrail: asking a COMPLETE task's agent must work and must NOT restart the
+    # workflow — no state change, no new history entry, no responsibilities.
+    task_id = _task(client)
+    client.set_state(task_id, "COMPLETE")  # free move to the terminal state
+    before = client.get_task(task_id)
+    assert before["state"] == "COMPLETE"
+    history_len = len(before["history"])
+
+    ask_id = client.create_ask(task_id, "what did you change in the store?")
+    client.record_ask_answer(task_id, ask_id, "added two lookup queries")
+
+    after = client.get_task(task_id)
+    assert client.get_ask(task_id, ask_id)["status"] == "answered"
+    assert after["state"] == "COMPLETE"  # still terminal — the ask was conversation, not a move
+    assert len(after["history"]) == history_len  # no transition recorded
+
+
+def test_ask_with_a_reaped_volume_returns_410(client: TaskServiceClient) -> None:
+    # If the config volume is gone (reaped), the ask can't be delivered. The session service marks
+    # it gone and the poll returns 410 — the documented signal for the review tool's fallback.
+    task_id = _task(client)
+    ask_id = client.create_ask(task_id, "still around?")
+    client.mark_ask_gone(task_id, ask_id)  # the ask worker's volume-gone outcome
+
+    resp = client._http.get(f"/tasks/{task_id}/ask/{ask_id}")
+    assert resp.status_code == 410
+    with pytest.raises(httpx.HTTPStatusError):
+        client.get_ask(task_id, ask_id)
+
+
+def test_ask_on_unknown_task_is_404(client: TaskServiceClient) -> None:
+    resp = client._http.post("/tasks/nope/ask", json={"question": "hi?"})
+    assert resp.status_code == 404
+
+
+def test_lookup_by_branch(client: TaskServiceClient) -> None:
+    task_id = _task(client)
+    client.set_slug(task_id, "fix-widget")
+    client.record_provisioning(task_id, "panopticon/fix-widget", f"/clones/{task_id}")
+
+    found = client.lookup_task(repo_id="r1", branch="panopticon/fix-widget")
+    assert found is not None and found["id"] == task_id
+    # A branch that no task holds → 404 → None.
+    assert client.lookup_task(repo_id="r1", branch="panopticon/absent") is None
+
+
+def test_lookup_by_url(client: TaskServiceClient) -> None:
+    task_id = _task(client)
+    client.set_url(task_id, "https://forge/pr/7")
+
+    found = client.lookup_task(url="https://forge/pr/7")
+    assert found is not None and found["id"] == task_id
+    assert client.lookup_task(url="https://forge/pr/999") is None
+
+
+def test_lookup_requires_a_valid_selector(client: TaskServiceClient) -> None:
+    # Neither branch nor url → 400 (a malformed request, not a miss).
+    assert client._http.get("/tasks/lookup").status_code == 400
+    # Mixing url with branch is rejected too.
+    assert (
+        client._http.get(
+            "/tasks/lookup", params={"url": "u", "repo_id": "r1", "branch": "b"}
+        ).status_code
+        == 400
+    )
+
+
+def test_lookup_is_not_shadowed_by_the_task_id_route(client: TaskServiceClient) -> None:
+    # `/tasks/lookup` must resolve to the lookup handler, not be captured as task_id="lookup".
+    resp = client._http.get("/tasks/lookup", params={"url": "https://none"})
+    assert resp.status_code == 404  # a clean "no match", not a 200 task nor a validation error