2026 10 27 live control

Live control: a person's control as a workspace state, with await_control: Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: A person who takes Control in the live view holds the browser. The agent's running call finishes first. After it:

• every surface's browser-changing call is refused with a 409 that names await_control;
• reads still answer;
• await_control returns the moment they hand back, with what they did. It runs as an MCP task, or as a capped blocking call.

Architecture:

• A plain live/control.py owns the episode per workspace (requested → held → released). It also owns a thread-safe gate keyed by Grid id, which the call hosts read after taking their turn on the session's lock.
• The grant waits for a mark in the lock's queue, and owes the session monitor a new control reason. That reason's events reach a per-reason sink, live/pages.py, which lists the pages the person reached.
• The MCP notice gains a structured control field.
• await_control is one async body, served by an async MCP tool with TaskConfig(mode="optional") and by a long-poll route.

Tech Stack: Python 3.11+, FastMCP 4.1.4 with fastmcp[tasks] (fastmcp-tasks 4.1.4, pydocket, memory://), Starlette, asyncio/anyio, WebDriver BiDi through bidi/, pytest with asyncio_mode = "auto"; Svelte 5, Vitest.

Spec: docs/superpowers/specs/2026-10-27-live-control-design.md. The issue's draft, finished on this branch rather than replaced:

• round 1 checked it against the code and added rulings 21–33;
• it corrected rulings 11 and 14, and fixed the refusal's words (24);
• nothing is open.

Design

The spec is the design. This plan settles these on top of it:

• Modules.
  • live/control.py: Controls, Episode and the async await_control body. Plain: no Selenium, no protocol imports. workspaces and the lock mark are injected.
  • live/pages.py: Pages, the control reason's monitor sink.
  • live/tunnels.py: keeps the tunnel registry and delegates control to a Controls it is given. Its notice copy moves to live/control.py.
  • mcp/waiting.py: the R6 helper.
  • faults.Controlled is the refusal.
• Who calls whom. Tunnel → Tunnels hooks → Controls. Controls → Monitor.attach_here / release, and the injected behind(grid_id). Hosts → Calls.begin(tool=) → Controls.refusal. Actions._at and screenshot → actions.control.refusal(session_id, "navigate"). Nothing below mcp/ imports live/. server.py wires actions.control and Workspaces(control=), as it wires actions.capture today.
• Order of construction in server.py:
  1. Pages;
  2. Monitor(sinks={"capture": capture, "control": pages});
  3. Controls(monitor=, pages=, behind=locks.behind);
  4. Tunnels(controls);
  5. Workspaces(..., control=controls);
  6. actions.control = controls.

Global Constraints

• Python 3.11 is the floor (CONTRIBUTING.md): nothing 3.12+.
• live/ stays plain. It imports no fastmcp, starlette, mcp, uvicorn or selenium, and nothing from mcp, http, routes, server or spec. live.control and live.pages join NO_SELENIUM in tests/test_boundaries.py. anyio is allowed, as in monitor/.
• No log line formats a Grid id, a page URL, a title or paste text (test_no_log_line_formats_a_grid_id). The page list is never logged.
• No thread ever blocks on the session's lock for the grant (spec ruling 25).
• The copy, verbatim:
  • refusal, held: {who} has control of this browser since {HH:MM} UTC. Call await_control (POST /browser/await-control), which returns when they hand back.
  • refusal, requested: {who} asked for control of this browser at {HH:MM} UTC. Call await_control (POST /browser/await-control), which returns when they hand back.
  • notice on the in-flight call (state: "requested"): {who} asked for control of this browser at {HH:MM} UTC; it is theirs once this call ends. Call await_control, which returns when they hand back.
  • notice on a read while held (state: "held"): the held refusal sentence.
  • notice, unsaid hand-back (state: "handed_back"): control of this browser was handed back at {HH:MM} UTC; the page may have changed. Call await_control for what they did, or look at the page before carrying on.
  • await_control at the cap: still in {who}'s control; call again.
  • await_control with nobody in control: nobody has control of this browser: carry on.
  • await_control after the browser ended: the browser ended while {who} had control: call open_session.
  • the viewer while requested: waiting for the agent's call to finish.
  • {who}: admin and open become a person; distinct names are joined with and.
• The values: live.await_cap 90 (ge=30); grant poll 0.1 s; TaskConfig(mode="optional", poll_interval=timedelta(seconds=1)); TasksExtension(url="memory://"); pages capped at HISTORY_CAP (100); a title read bounded by REPLY_TIMEOUT; the grant's attach bounded by ATTACH_TIMEOUT (15 s).
• Every published shape that moves is a golden. Regenerate with GOLDEN_UPDATE=1 pytest, read the diff, and commit the goldens on their own (CONTRIBUTING.md).
• The wiki is regenerated with python scripts/generate_wiki.py, committed inside docs/ with the pointer bump, and tests/test_wiki.py passes.
• Changelog: one line under ## [Unreleased].
• No new integration flow (AGENTS.md: less is more).
• Never commit kubed/selenium_flow/http/static/ or openapi.yaml. Never edit .github/.
• The checks, as CI runs them. Each task's last step runs the ones its files touch, and Task 12 runs them all:
  • Python (test.yml): ruff check . and pytest;
  • UI (ui.yml): npm --prefix ui ci, npm --prefix ui run lint, npm --prefix ui run check, npm --prefix ui test, npm --prefix ui run build;
  • quality.yml: python scripts/generate_openapi.py succeeds, and pip-audit passes with fastmcp[tasks] installed;
  • pr.yml: a new [Unreleased] entry.

Review Focus

1. A flow running when a person clicks Control, with onError: continue on its next step. A person expects the run to stop there with the refusal, not to skip to step 3 and keep driving the browser under their hands. Task 6 pins it.
2. A read that names a url, such as extract(url=…) while the browser is elsewhere, or a screenshot(width=1600). A person expects their page and window to stay as they left them. Task 6 pins both refusals, and that extract with the current URL answers.
3. The person closes the tab while Control is only requested. The agent was told to wait. A person expects the agent's await_control to return, and its next call to be allowed. They do not expect a gate left shut with nobody holding it. Task 4 pins the withdrawn episode.
4. The server stops while a grant waits behind a 900 s assert. A person expects the pod to stop promptly. Task 3 pins that the mark takes no thread. Task 4 pins that Controls.stop() cancels the pending grant and every waiter returns.
5. A Tasks client cancels await_control, or a blocking client disconnects mid-wait. A person expects nothing left behind: the episode still holds, and the next refusal still names it. Task 9 pins a cancelled wait leaving the episode and its event untouched.

---

Task 1: Take fastmcp[tasks], and pin the facts the design rests on

Files:

• Modify: pyproject.toml (dependencies: fastmcp[tasks]>=4.1.4,<5 replaces fastmcp>=4.0.0,<5)
• Create: tests/test_tasks_contract.py

Interfaces:

• Produces: the facts below, as passing tests, for Tasks 8 and 9. If a fact fails against what installs, stop and report which one. Do not design around it in place: Tasks 8 and 9 change with it.

• Step 1: Write the tests against a toy FastMCP server in the test file. It has one async tool with task=TaskConfig(mode="optional", poll_interval=timedelta(seconds=1)), one middleware recording what call_next returned, and mcp.add_extension(TasksExtension(url="memory://")):

def test_task_on_a_sync_function_is_refused(): ...          # ValueError at decoration (R6)
async def test_call_next_returns_a_create_task_result(): ...  # mode="auto": the middleware saw CreateTaskResult (R1)
async def test_a_legacy_client_runs_the_tool_inline(): ...    # mode="legacy": a ToolResult, no task
async def test_the_worker_has_no_request_but_the_headers(): ... # in the tool: get_http_request raises; get_http_headers has a header the middleware set at submission (R3)
async def test_the_worker_runs_on_the_server_loop(): ...      # an asyncio.Event made on the server's loop wakes the task
async def test_structured_content_survives_tasks_get(): ...   # result().structured_content == the tool's dict (R2)
async def test_the_poll_interval_is_one_second(): ...         # the CreateTaskResult's pollIntervalMs == 1000 (R4)
async def test_cancel_cancels_the_wait(): ...                 # tasks/cancel → the tool sees CancelledError

• Step 2: Install and run.

Run: pip install -e ".[test]" && pytest tests/test_tasks_contract.py -v Expected: PASS. A failure is a finding: report it.

• Step 3: Check the audit and the package.

Run: pip-audit (as quality.yml runs it) and ruff check . Expected: no finding on fastmcp-tasks, pydocket or burner-redis.

• Step 4: Commit build: take fastmcp[tasks], with the facts await_control rests on.

---

Task 2: Tunnel.run always closes

Files:

• Modify: kubed/selenium_flow/live/tunnel.py (run's finally)

• Test: tests/test_live_tunnel.py

• Step 1: Write the failing test.

async def test_a_second_cancellation_during_teardown_still_closes_both_sockets():
    # a reader task that ignores its first cancel and blocks;
    # cancel run(), then cancel it again while it waits for the tasks
    assert viewer.closed_with is not None and upstream.closed

• Step 2: Run it to see it fail. pytest tests/test_live_tunnel.py -v -k second_cancellation. Expected: FAIL, the viewer is not closed.

• Step 3: Implement. await asyncio.wait(tasks) goes inside try:, and await self._close(self._ending or STOPPING) goes in its finally:. Nothing else changes; _close's awaits are already bounded.

• Step 4: Run the live tests. pytest tests/test_live_tunnel.py tests/test_live_tunnels.py tests/test_admin_live.py -v && ruff check . Expected: PASS.

• Step 5: Commit fix: a live tunnel cancelled twice still closes both sockets.

---

Task 3: The refusal, its 409, the column, and the lock's mark

Files:

• Modify: kubed/selenium_flow/faults.py (Controlled)
• Modify: kubed/selenium_flow/errors.py (status_for: 409 first)
• Modify: kubed/selenium_flow/spec/builder.py ("409" in every /browser operation's responses, beside "404" at 233; not on /browser GET)
• Modify: kubed/selenium_flow/core/capabilities.py (Capability.under_control, Capability.waits; the reads set "answers"; UNDER_CONTROL view)
• Modify: kubed/selenium_flow/workspace/locks.py (Turns.mark, Turns.passed, behind)
• Test: tests/test_errors.py, tests/test_surfaces.py, tests/test_locks.py; goldens tests/golden/ (errors, OpenAPI)

Interfaces:

• Produces:

# faults.py
class Controlled(Exception):
    """A person holds this workspace's browser; a 409 (spec 2026-10-27-live-control)."""
    def __init__(self, message: str): ...      # str(exc) is the refusal sentence

# core/capabilities.py
under_control: Literal["answers", "refused"] = "refused"   # new column
waits: bool = False                                        # served by its own body on both surfaces
UNDER_CONTROL: dict[str, str]                              # name -> column, read off the table
# answers: extract, outline, screenshot, print, network_body, await_control (Task 9 adds that row)

# workspace/locks.py
class Turns:
    def mark(self) -> int        # take a ticket and abandon it at once; never waits
    def passed(self, ticket: int) -> bool   # every call that asked before `ticket` has had its turn
def behind(browser: str) -> Callable[[], bool]
    # mark this browser's queue now; the callable keeps the Hold alive and answers `passed`

• Step 1: Write the failing tests.

# test_errors.py
def test_a_controlled_browser_is_a_409():
    assert status_for(Controlled("kelly has control")) == 409

# test_surfaces.py
def test_what_answers_under_control():
    assert {n for n, v in UNDER_CONTROL.items() if v == "answers"} == {
        "extract", "outline", "screenshot", "print", "network_body"}   # await_control joins in Task 9

# test_locks.py
def test_a_mark_on_an_idle_browser_has_passed_at_once():
    assert behind("g1")()
def test_a_mark_passes_when_the_call_before_it_lets_go():
    # thread A holds driving("g1"); passed = behind("g1"); assert not passed(); A releases; assert passed()
def test_a_call_after_the_mark_does_not_wait_for_it():
    # mark while A holds; B queues after the mark; A releases; B gets its turn at once
def test_a_mark_holds_no_thread():
    # threading.active_count() unchanged across behind() while A holds

The OpenAPI golden test fails until Step 4 regenerates it.

• Step 2: Run them to see them fail. pytest tests/test_errors.py tests/test_surfaces.py tests/test_locks.py -v. Expected: FAIL (Controlled, UNDER_CONTROL and behind are undefined).

• Step 3: Implement.

  • Turns.mark takes self._next, adds it to _abandoned and runs _skip_abandoned(), all under _changed.
  • passed(t) is self._serving > t.
  • The 409's description in spec/builder.py: A person holds this workspace's browser in the live view. Call POST /browser/await-control, then retry.

• Step 4: Regenerate the goldens and run. GOLDEN_UPDATE=1 pytest, read the diff under tests/golden/, then python scripts/generate_openapi.py && pytest && ruff check .. Expected: PASS; the diff adds only the 409s.

• Step 5: Commit the code as feat: a 409 for a browser a person holds, and which capabilities answer then, then the goldens on their own.

---

Task 4: Episodes: request, grant, release

Files:

• Create: kubed/selenium_flow/live/control.py
• Modify: kubed/selenium_flow/live/tunnels.py (delegates control; TAKEN/HANDED_BACK move out)
• Modify: kubed/selenium_flow/live/tunnel.py (mode gains "requested"; the gate is fed "control" only when held; grant())
• Modify: kubed/selenium_flow/live/audit.py (asked, beside switched)
• Modify: kubed/selenium_flow/http/admin/live.py (Tunnel no longer takes may_control; paste asks controls.controlling)
• Modify: kubed/selenium_flow/server.py (Controls, then Tunnels(controls); controls.stop() in the lifespan before monitor.stop())
• Modify: tests/test_boundaries.py (live.control in NO_SELENIUM)
• Test: tests/test_live_control.py (new), tests/test_live_tunnel.py, tests/test_live_tunnels.py, tests/test_admin_live.py

Interfaces:

• Consumes: locks.behind (Task 3), faults.Controlled (Task 3).
• Produces:

# live/control.py
@dataclass(frozen=True)
class Gate:                     # what a worker thread reads; replaced whole, never mutated
    workspace: str
    who: tuple[str, ...]
    since: float
    state: Literal["requested", "held"]

@dataclass
class Episode:
    workspace: str; grid_id: str; who: list[str]; since: float
    state: Literal["requested", "held", "released"]
    granted: float | None = None; until: float | None = None
    ended: bool = False          # the browser ended under it
    followed: bool = False       # pages listened for (Task 7)
    done: asyncio.Event          # set at the release

class Controls:
    def __init__(self, *, behind: Callable[[str], Callable[[], bool]], monitor=None,
                 pages=None, clock=time.time, poll: float = 0.1): ...
    def ask(self, tunnel) -> Literal["control", "requested"]   # on the loop
    def withdraw(self, tunnel) -> None                          # Watch, or the tunnel ended
    def ended(self, grid_id: str) -> None                       # session.ended: release, ended=True
    def refusal(self, session_id: str, tool: str) -> Controlled | None   # any thread
    def controlling(self, workspace: str, who: str) -> bool
    def notice(self, workspace: str) -> dict | None   # {"state","who","since","hint"}; pops an unsaid hand-back
    async def wait(self, workspace: str, timeout: float | None) -> Waited   # Task 9 uses it
    def stop(self) -> None                            # cancel pending grants, release every episode

def said(who: Iterable[str]) -> str      # "a person" for admin/open, " and "-joined

• Step 1: Write the failing tests in tests/test_live_control.py. They use fake tunnels (workspace, who, grid_id, an async grant() that records) and a fake behind whose answer the test flips:

async def test_with_no_call_running_control_is_granted_at_once():
    assert controls.ask(t1) == "control" and controls.refusal("g1", "navigate") is not None

async def test_with_a_call_running_control_is_requested_then_granted():
    passed.value = False
    assert controls.ask(t1) == "requested"
    assert "asked for control" in str(controls.refusal("g1", "interact"))
    passed.value = True; await asyncio.sleep(0.15)
    assert t1.granted and "has control" in str(controls.refusal("g1", "interact"))

async def test_reads_and_paste_are_never_refused():
    controls.ask(t1)
    assert controls.refusal("g1", "extract") is None and controls.refusal("g1", "paste") is None

async def test_leaving_while_requested_ends_the_episode():          # ruling 23
    passed.value = False; controls.ask(t1); controls.withdraw(t1)
    assert controls.refusal("g1", "navigate") is None
    assert controls.notice("w1")["state"] == "handed_back"

async def test_control_is_held_while_any_viewer_holds_it():
    controls.ask(t1); controls.ask(t2); controls.withdraw(t1)
    assert controls.refusal("g1", "navigate") is not None
    controls.withdraw(t2)
    assert controls.refusal("g1", "navigate") is None

async def test_the_browser_ending_releases_with_ended(): ...
async def test_stop_cancels_a_pending_grant_and_sets_every_event(): ...   # Review Focus 4
def test_the_refusal_says_who_and_when():
    assert str(refusal) == ("a person has control of this browser since 15:12 UTC. Call "
                            "await_control (POST /browser/await-control), which returns when they hand back.")

In tests/test_live_tunnel.py:

async def test_a_requested_viewer_is_told_requested_and_stays_watch_at_the_gate(): ...
async def test_a_grant_sends_control_and_opens_the_gate(): ...

• Step 2: Run them to see them fail. pytest tests/test_live_control.py tests/test_live_tunnel.py -v. Expected: FAIL, no module live.control.

• Step 3: Implement.

  • Asking. ask starts or joins the episode and replaces the gate at once. It calls behind(grid_id) once per episode: already passed means _grant now; otherwise a task polls every poll seconds, then grants.
  • _grant sets held and granted, replaces the gate, and awaits each requesting tunnel's grant(). Task 7 adds the monitor's attach before it.
  • The release. withdraw of the last asker sets released, until, keeps the episode as the workspace's unsaid hand-back, clears the gate and sets done.
  • Tunnel._control on mode: control, once piping, asks self.on_ask(), which Tunnels.add wires to controls.ask. It sends what that returns. grant() sets mode = "control", sends it and audits switched.
  • Tunnels. on_mode to watch and on_end call withdraw. Tunnels.ended also calls controls.ended(event.session_id). Tunnels.notice and controlling delegate.

• Step 4: Run the live tests. pytest tests/test_live_control.py tests/test_live_tunnel.py tests/test_live_tunnels.py tests/test_admin_live.py tests/test_control_notice.py tests/test_boundaries.py -v && ruff check . Expected: PASS. test_control_notice.py is updated to the new copy in Task 8; mark its copy assertions xfail here only if they block, and remove the mark there.

• Step 5: Commit feat: Control is requested while the agent's call runs, and granted after it.

---

Task 5: The hosts refuse, after their turn

Files:

• Modify: kubed/selenium_flow/workspace/calls.py (Calls(control=); begin(..., tool: str) checks first)
• Modify: kubed/selenium_flow/workspace/workspaces.py:
  • _said always takes locks.driving;
  • Workspaces(control=) passes it to Calls;
  • open_browser and end_browser check control.refusal(record.session_id, name) before acting;
  • paste passes tool="paste", as it does.
• Modify: kubed/selenium_flow/secrets.py (perform_write: calls.begin(..., tool="write"))
• Modify: kubed/selenium_flow/flows/api.py (step_started passes entry["tool"])
• Modify: kubed/selenium_flow/flows/engine.py (_drive: a faults.Controlled stops the run like Cancelled)
• Modify: kubed/selenium_flow/server.py (Workspaces(..., control=controls))
• Test: tests/test_workspaces.py (or the file that owns act), tests/test_flow_engine.py (or the file that owns _drive), tests/test_secrets*.py

Interfaces:

• Consumes: Controls.refusal (Task 4), duck-typed: anything with refusal(session_id, tool) -> Exception | None.

• Produces: Calls.begin(workspace, session_id, *, browser="", tool: str) -> Begun, raising the refusal before monitor.calling.

• Step 1: Write the failing tests. Use a fake control whose refusal returns Controlled(...) for a Grid id, and None for reads:

def test_an_acting_call_is_refused_and_never_begun():      # monitor.calling not called; no call.finished
def test_a_read_answers_while_controlled():
def test_a_call_queued_before_the_request_is_refused_when_its_turn_comes():
    # A holds driving(g1); B enters act() and waits; the gate shuts; A releases → B raises Controlled
def test_open_session_and_end_browser_are_refused_while_controlled():
def test_a_bound_write_is_refused_before_the_page_is_read():  # actions.page never called
def test_a_flow_stops_at_a_refused_step_even_with_on_error_continue():  # Review Focus 1
    # step 2 refused, onError: continue → status "failed", step 3 never dispatched,
    # step 2's error is the refusal sentence
def test_paste_is_never_refused():

• Step 2: Run them to see them fail. pytest tests/test_workspaces.py tests/test_flow_engine.py -v -k "controlled or refused or paste". Expected: FAIL.

• Step 3: Implement to the interfaces. In _drive, the onError == "continue" test gains and not isinstance(exc, faults.Controlled).

• Step 4: Run the suite. pytest && ruff check . Expected: PASS. The lock is now held for every _said, so watch tests/test_locks.py and the capture tests in particular.

• Step 5: Commit feat: every surface is refused while a person holds the browser.

---

Task 6: Reads that would change what the person sees are refused

Files:

• Modify: kubed/selenium_flow/core/browser.py (ensure_url(driver, url, *, before_going: Callable[[], None] = lambda: None))

• Modify: kubed/selenium_flow/core/actions.py:

  • Actions.control = None, beside capture;
  • _at passes before_going that raises self.control.refusal(session_id, "navigate");
  • screenshot's capture raises refusal(session_id, "resize") before a width, height or full_page resize.

• Modify: kubed/selenium_flow/server.py (self.actions.control = controls)

• Test: tests/test_actions*.py (the file that drives extract and screenshot through ScriptedDriver)

• Step 1: Write the failing tests (Review Focus 2):

def test_extract_on_the_current_url_answers_while_controlled():
def test_extract_naming_another_url_is_refused_before_it_navigates():   # driver.get never called
def test_a_screenshot_with_a_width_is_refused_while_controlled():       # set_window_size never called
def test_a_full_page_screenshot_is_refused_while_controlled():
def test_a_plain_screenshot_answers_while_controlled():

• Step 2: Run them to see them fail. pytest tests/ -v -k "while_controlled". Expected: FAIL.

• Step 3: Implement to the signatures above. refusal returns None when nobody holds the browser, so the check costs one dict read.

• Step 4: Run. pytest && ruff check . Expected: PASS.

• Step 5: Commit feat: a read that would move the page or the window waits for the hand-back.

---

Task 7: The monitor's control reason, and the pages the person reached

Files:

• Modify: kubed/selenium_flow/monitor/monitor.py:
  • sinks: Mapping[str, Sink] replaces sink;
  • Watch.heard and Watch.top;
  • routing in _hold, _extend, _deliver, _loser, _detached and _unowe;
  • capturing() means capture's connection;
  • async attach_here(...).
• Create: kubed/selenium_flow/live/pages.py
• Modify: kubed/selenium_flow/live/control.py (_grant awaits monitor.attach_here(..., "control") first; the release calls monitor.release(grid_id, "control") and pages.stop(grid_id))
• Modify: kubed/selenium_flow/server.py:
  • Pages() before the monitor;
  • events_for={"capture": EVENTS, "control": pages_module.EVENTS};
  • sinks={"capture": capture, "control": pages};
  • pages.connection = monitor.connection.
• Modify: tests/test_boundaries.py (live.pages in NO_SELENIUM)
• Test: tests/test_monitor.py, tests/test_monitor_capture.py, tests/test_live_pages.py (new), tests/test_live_control.py

Interfaces:

• Produces:

# monitor/monitor.py
Monitor(bus, *, ..., sinks: Mapping[str, Sink] | None = None, ...)   # `sink=` goes; callers updated
async def attach_here(self, workspace: str, session_id: str, reason: str, *, browser: str = "") -> None
    # `attach` on the loop: `_attach` under asyncio.timeout(ATTACH_TIMEOUT); any failure is BidiUnavailable
def capturing(self, session_id: str) -> bool   # held, and "capture" in the watch's heard reasons
# Routing: a reason's sink hears `attached` when the hold opens with it owed (`_hold`),
# or when `_extend` adds it (top = Watch.top, again=False); then `ingest` for the
# events in events_for[reason], `replayed`, and `lost`; and `detached` at the close,
# or at `_unowe(reason)` when the connection stays for another.

# live/pages.py
EVENTS = ("browsingContext.load", "browsingContext.historyUpdated", "browsingContext.contextCreated")
class Pages:                                   # a monitor Sink
    connection: Callable[[str], Connection | None]
    def follow(self, grid_id: str) -> None     # start a list; Controls calls it before attach_here
    def stop(self, grid_id: str) -> list[dict] # the list, ended; [] if never followed
    def listed(self, grid_id: str) -> tuple[list[dict], int]   # (pages, dropped), oldest first

• Step 1: Write the failing tests.

# test_monitor_capture.py
async def test_a_control_only_hold_never_reaches_capture():
    # attach_here(..., "control"): capture's sink hears no attached/ingest/detached;
    # monitor.capturing(sid) is False; Calls(capture=capture).wanted("w1") is False
async def test_control_added_to_a_capture_hold_subscribes_its_events_on_the_same_connection():
    # one connection; session.subscribe sent once more with only browsingContext.load (contextCreated
    # and historyUpdated are capture's already); a load event reaches pages, not capture
async def test_releasing_control_keeps_capture_held_and_tells_pages_detached():
async def test_releasing_control_alone_closes_the_connection_clean():

# test_live_pages.py
async def test_pages_in_order_with_titles():          # load, historyUpdated → [{url,title},…]
async def test_a_frames_load_is_not_a_page():          # context with a parent
async def test_a_new_tab_is_followed():                # contextCreated parent None, then load
async def test_a_title_that_does_not_come_back_is_left_out():   # script.evaluate times out
async def test_a_url_with_no_origin_records_nothing(): # about:blank, data:
async def test_at_most_a_hundred_pages_with_dropped():

# test_live_control.py
async def test_the_grant_waits_for_the_pages_to_be_followed():
async def test_bidi_unavailable_still_grants_with_followed_false():

• Step 2: Run them to see them fail. pytest tests/test_monitor.py tests/test_monitor_capture.py tests/test_live_pages.py tests/test_live_control.py -v. Expected: FAIL.

• Step 3: Implement.

  • Pages.ingest appends on load and historyUpdated for a top-level context. That is Watch.top, handed in through attached, or one seen in contextCreated with parent null.
  • The title is read in a task: connection(grid_id).command("script.evaluate", {"expression": "document.title", "target": {"context": ctx}, "awaitPromise": False}) under asyncio.timeout(REPLY_TIMEOUT), and filled into the entry it was appended for.
  • No URL or title is logged.

• Step 4: Run. pytest && ruff check . Expected: PASS. The capture tests are the regression net for the routing.

• Step 5: Commit feat: the pages a person reaches in Control are followed on capture's connection.

---

Task 8: The notice is structured, and a task's result passes through

Files:

• Modify: kubed/selenium_flow/mcp/control.py (Notices(controls))
• Modify: kubed/selenium_flow/server.py (control.Notices(self.controls))
• Test: tests/test_control_notice.py

Interfaces:

• Consumes: Controls.notice(workspace) -> dict | None (Task 4).

• Step 1: Write the failing tests, through a real fastmcp.Client on the test server:

async def test_an_in_flight_call_carries_control_in_structured_content():
    # the call's action asks control mid-call; result.structured_content["control"] ==
    # {"state": "requested", "who": ["kelly"], "since": "…Z", "hint": <the requested notice>}
    # and the text block ends with the same hint
async def test_a_read_while_held_carries_state_held():
async def test_the_unsaid_hand_back_is_said_once():
async def test_a_refusal_carries_no_second_notice():   # the error text has the refusal sentence once
async def test_a_create_task_result_passes_untouched(): # R1
async def test_a_screenshot_image_gains_structured_content_with_only_control():

• Step 2: Run them to see them fail. pytest tests/test_control_notice.py -v. Expected: FAIL.

• Step 3: Implement.

  • A ToolResult gains structured_content["control"]. When there is none, it becomes {"control": …}.
  • A ToolError whose __cause__ is a faults.Controlled is re-raised as it is.
  • Any other result type, such as CreateTaskResult, is returned untouched.

• Step 4: Run. pytest tests/test_control_notice.py tests/test_surfaces.py && ruff check . Expected: PASS. If a tool's output schema refuses control in a client's validation, declare control in capabilities._page() beside page. That moves the OpenAPI golden: regenerate and commit it on its own.

• Step 5: Commit feat: an agent sees control in the structured result Claude Code reads.

---

Task 9: await_control, as a task and as a long poll

Files:

• Modify: kubed/selenium_flow/config.py (LiveSettings.await_cap: int = Field(90, ge=30, description="Seconds a blocking await_control waits before it answers \"call again\"."))
• Modify: kubed/selenium_flow/core/capabilities.py (the await_control row: route="await-control", http_method="POST", annotations=reads("Wait for a person to hand back"), under_control="answers", waits=True, response=AWAIT_RESPONSE)
• Modify: kubed/selenium_flow/live/control.py (async def await_control)
• Modify: kubed/selenium_flow/workspace/workspaces.py (handed_back)
• Create: kubed/selenium_flow/mcp/waiting.py
• Modify: kubed/selenium_flow/mcp/tools.py (the async tool with task=; publish(..., task=None))
• Modify: kubed/selenium_flow/mcp/clients.py (CARRIED = "x-selenium-flow-workspace", read only in the no-request branch; Carry middleware)
• Modify: kubed/selenium_flow/routes.py (rows with waits get their own async handler)
• Modify: kubed/selenium_flow/flows/document.py (NOT_STEPS["await_control"]: A flow cannot wait for a person: a refused step stops the run, and whoever started it calls await_control)
• Modify: kubed/selenium_flow/server.py (mcp.add_extension(TasksExtension(url="memory://")); Carry added before Notices)
• Modify: tests/test_surfaces.py (EXPECTED gains await_control; test_every_route_maps_to_a_real_action skips rows with waits; read-only pinned in both modes; UNDER_CONTROL answers gains it)
• Test: tests/test_await_control.py (new); goldens (tool lists, OpenAPI); docs/ wiki page and docs/Configuration.md

Interfaces:

• Consumes: Controls.wait, Episode (Task 4), Pages.listed / stop (Task 7), Workspaces.browser and settle.
• Produces:

# live/control.py
@dataclass(frozen=True)
class Waited:
    kind: Literal["handed_back", "still", "nobody"]
    episode: Episode | None
async def await_control(controls: Controls, workspaces, name: str, *, cap: float | None) -> dict
    # wait (cap=None: until the hand-back), then workspaces.handed_back in a thread (anyio)

# workspace/workspaces.py
def handed_back(self, name: str, episode: dict | None) -> dict
    # page (url, title) and outline({}) through `_said` on `self.browser(name)`, never reopening;
    # settle(url=None); the summary per the spec, `visited` shaped by urls (last whole minus userinfo,
    # earlier page_of); `ended: true` with no page or outline when there is no browser

# mcp/waiting.py (R6)
async def waited(ctx, work: Callable[[float | None], Awaitable[T]], *, cap: float, message: Callable[[], str]) -> T
    # cap is None in a background task (ctx.is_background_task); progress every progress.HEARTBEAT

# mcp/clients.py
class Carry(Middleware):   # on_call_tool for a tool with task support: set CARRIED in the request's
                           # scope headers from caller().name before call_next (R3, ruling 31)

• Step 1: Write the failing tests in tests/test_await_control.py, through a real fastmcp.Client on the test server and through the ASGI app:

async def test_it_returns_with_the_summary_the_moment_the_person_hands_back():
    # legacy client; the call waits; controls.withdraw → result.structured_content has
    # handed_back True, control.who ["kelly"], held_for, page, visited, outline
async def test_a_tasks_client_gets_a_task_and_the_summary_at_the_hand_back():
    # import fastmcp_tasks; Client(mode="auto"); call_tool_task → status "working";
    # hand back → result().structured_content["handed_back"] is True
async def test_the_workspace_reaches_the_worker_from_the_query():   # ?workspace=w1 only
async def test_the_blocking_call_sends_progress_and_answers_at_the_cap():
    # cap 0.2 (patched); handed_back False; message "still in a person's control; call again."
async def test_nobody_in_control_answers_at_once():
async def test_an_unsaid_hand_back_is_returned_once():               # then the next result has no notice
async def test_a_browser_that_ended_is_said_not_reopened():           # grid.open never called
async def test_a_cancelled_wait_leaves_the_episode_holding():         # Review Focus 5
async def test_the_long_poll_answers_the_same_summary():               # POST /browser/await-control
async def test_the_long_poll_answers_at_the_cap():
def test_it_is_not_a_flow_step():                                     # save_flow refuses it
def test_the_cap_is_at_least_thirty():                                # LIVE_AWAIT_CAP=10 → ConfigError

• Step 2: Run them to see them fail. pytest tests/test_await_control.py -v. Expected: FAIL.

• Step 3: Implement to the interfaces.

  • The tool is @own("await_control", task=TaskConfig(mode="optional", poll_interval=timedelta(seconds=1))), an async def await_control(ctx: Context) -> dict.
  • Its docstring is prompt. It says what control is, that a refusal names this tool, and that it returns when the person hands back, with what they did. It says that on a client without background tasks it returns after about 90 seconds saying to call again.
  • The route's handler authorises and names the caller as _answer does, but awaits the body on the loop rather than in the threadpool.

• Step 4: Regenerate and run.

Run: GOLDEN_UPDATE=1 pytest (read the diff under tests/golden/: one tool in both modes, one operation), then python scripts/generate_openapi.py && python scripts/generate_wiki.py && pytest && ruff check . Expected: PASS. docs/await_control.md exists, and docs/Configuration.md gains live.await_cap.

• Step 5: Commit the code as feat: await_control returns when a person hands the browser back, then the goldens on their own, then the wiki inside docs/ with the pointer bump.

---

Task 10: The viewer shows "requested"

Files:

• Modify: ui/src/admin/liveview.svelte.ts (LiveMode gains 'requested'; #text accepts it; viewOnly stays true until control)

• Modify: ui/src/admin/LivePane.svelte (the toggle reads waiting for the agent's call to finish while requested; Paste stays disabled)

• Test: ui/src/admin/liveview.test.ts, ui/src/admin/LivePane.test.ts

• Step 1: Write the failing tests.

it('stays view-only while control is requested', () => { /* text frame mode:requested → mode 'requested', rfb.viewOnly true */ })
it('says it is waiting for the agent', () => { /* LivePane shows "waiting for the agent's call to finish" */ })
it('grants on control', () => { /* requested then control → viewOnly false */ })

• Step 2: Run them to see them fail. npm --prefix ui test. Expected: FAIL.

• Step 3: Implement.

• Step 4: Run the UI checks. npm --prefix ui ci && npm --prefix ui run lint && npm --prefix ui run check && npm --prefix ui test && npm --prefix ui run build. Expected: PASS.

• Step 5: Commit feat: the live view says when Control waits for the agent's call.

---

Task 11: The words: skill, AGENTS.md, notes, changelog

Files:

• Modify: skills/selenium-flow/SKILL.md (a routing-table line: "a person took control, or a call was refused for it → references/WORKSPACES.md")

• Modify: skills/selenium-flow/references/WORKSPACES.md (a "When a person takes control" section)

• Modify: AGENTS.md:

  • "The live view" loses ruling 35's sentence on the notice and gains a short "A person in Control holds the browser" bullet list: the gate after the turn; the reads; the mark, not a thread; the control reason's own sink; await_control as a task and as a capped poll; Docket in memory;
  • the "Scaling" paragraph gains control and Docket as process memory.

• Create: docs/notes/await_control.notes.md (the cap, Tasks versus blocking, the gateway)

• Modify: CHANGELOG.md, under ## [Unreleased] → Added: **A person who takes Control in the live view holds the browser**: agents are refused until they hand back, and \await_control` waits for it.`

• Test: tests/test_skill*.py, tests/test_wiki.py, tests/test_readme.py

• Step 1: Write the section and the lines. Write for a model deciding what to do next: what a refusal looks like; call await_control; what the summary holds; reads still answer.

• Step 2: Regenerate and run. python scripts/generate_wiki.py && pytest tests/test_skill*.py tests/test_wiki.py tests/test_readme.py && ruff check . Expected: PASS. Every reference is linked, and no note shadows a page.

• Step 3: Commit docs: control and await_control for agents and maintainers, then the notes inside docs/ with the pointer bump.

---

Task 12: Everything, then live

• Step 1: Run every check CI runs.

ruff check . && pytest
npm --prefix ui ci && npm --prefix ui run lint && npm --prefix ui run check && npm --prefix ui test && npm --prefix ui run build
python scripts/generate_openapi.py
pip-audit

Expected: all pass, with CHANGELOG.md carrying the new [Unreleased] line.

• Step 2: Live, on the cluster (Dr K, after the image is deployed), against "Done when":
  • Claude Code through agentgateway, with Tasks: a call in flight finishes with control shown, the next acting call is refused, await_control runs in the background and returns at the hand-back with the pages;
  • the same with a client that has no Tasks, where the call returns at 90 s saying to call again;
  • pages followed with capture on and with capture off;
  • a shadow-DOM input keeps its caret, and the viewer gets 1012 when the server restarts with it open.