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: open_session(capture=true) keeps what the page logs and fetches in a bounded in-memory buffer read by URI, tells a call that caused a page error so, fetches one body on demand, streams everything live to the admin page — and every BiDi conversation with a browser goes through one client of our own.

Architecture: A new kernel package bidi/ (one asyncio Connection per browser with an id-routed reply table and a bounded queue and task per subscriber; a synchronous Channel facade for worker threads; Channels.open_channel(session_id) that lends the monitor's held connection or opens one for the block). The monitor holds a bidi.Connection for a session owed capture, attaches it synchronously before the first page, and feeds a capture/ package (buffers, masking, merging, alignment, TTL, the body index, the live tail). Site data drops Selenium's BiDi client for Channel. A workspace/calls.py seam at every recipe host touches the monitor, publishes call.finished and adds the hint. Resources, network_body, show views, admin endpoints and the admin tabs read the same buffer.

Tech Stack: Python 3.11+ (this plan raises the floor), asyncio, anyio, websockets>=15, cachetools, FastMCP 4.1, Starlette + sse-starlette, pytest + pytest-asyncio (asyncio_mode = "auto"); Svelte 5, vitest, @modelcontextprotocol/ext-apps.

Spec: docs/superpowers/specs/2026-10-10-capture-design.md (cited spec, its rulings ruling n). Also binding: the programme docs/superpowers/specs/2026-10-09-workspaces-and-observability-design.md (programme R n) and the monitor docs/superpowers/specs/2026-10-09-session-monitor-design.md (E2 r n), and AGENTS.md.

Commands. From the worktree root. In this sandbox a system kubed package shadows the tree, so:

export D=/tmp/claude-1000/-projects-kubed-io-selenium-flow/55f277e2-9886-4a71-8158-035ac51251f5/scratchpad/deps
export PYTHONPATH=$PWD:$D
python3 -m pytest -q            # the whole Python suite
$D/bin/ruff check .
npm --prefix ui test && npm --prefix ui run -s check && npm --prefix ui run -s lint

Goldens are rewritten with GOLDEN_UPDATE=1 python3 -m pytest tests/test_golden.py (and the test that owns the golden); a moved golden is a changed contract and is named in its commit message.

Copied from the spec, the programme and E2; every task's requirements include these.

Values (verbatim from the spec):

• Per workspace: 1 000 console entries and 1 000 requests, oldest evicted; the last 5 navigations marked (ruling 10).
• Expiry: capture.ttl seconds after the last entry, default 10 800 s (3 h), sliding on every entry; nothing is cleaned when the browser ends (ruling 10, programme R9).
• A request's detail keeps at most 32 headers of 4 kB each; a console text is capped at 4 kB (spec §4).
• Bodies stay in the browser: network_body(id) returns at most 100 kB of text, marked truncated; binary returns its type and size only (ruling 11).
• The live tail: the page keeps the last 2 000 rows; each listener has a server-side queue of 500, overflow counted and sent as {"dropped": n} (ruling 2, spec §4).
• Every BiDi subscriber: its own bounded queue of 1 000, drained by its own task, overflow dropped and counted; each command its own timeout; nothing runs on the read loop (ruling 8).
• The capture subscription: log.entryAdded, network.beforeRequestSent, network.responseCompleted, network.fetchError, browsingContext.contextCreated, browsingContext.contextDestroyed, browsingContext.navigationStarted, browsingContext.navigationCommitted, browsingContext.historyUpdated, browsingContext.fragmentNavigated (spec §2).
• The data collector: network.addDataCollector({"dataTypes": ["response"], "maxEncodedDataSize": 10000000, "collectorType": "blob"}) (spec §2).
• Masking, once, at capture, shared by buffer and tail (ruling 13): header names authorization, proxy-authorization, cookie, set-cookie in any case → •••; request.cookies[].value → •••; goog:postData dropped; every value the secrets catalogue bound in this session replaced by ••• wherever it appears in a URL, header, console text or argument; URLs keep their query string only on the newest navigation.
• Not kept, only counted per navigation: successful static requests (type stylesheet, image, font, script or media with a 2xx/3xx response) (ruling 3) and debug console entries (ruling 4). Every failure is kept.
• The hint: "page": {"errors": n, "failed_requests": n, "see": "<uri>"}, present only when a count is non-zero; counted: error console entries and uncaught exceptions; 4xx/5xx and failed document/fetch/xhr requests; not warnings, not failed images; a flow run sums its steps' (ruling 12, programme R10).
• With capture off a capture resource answers {"capture": false, "how": "open_session(capture=true)"}, not an error (spec §7).
• capture is a workspace setting (None, None, _as_flag) beside record: replayed on a silent reopen after a reap, never inherited by an explicit open_session, off by default (programme R8).
• Python ≥ 3.11 (ruling 5): requires-python = ">=3.11"; every 3.10 workaround removed.

AGENTS.md rules every task keeps:

• Every browser action is both a tool and an endpoint, one to one, mounted from CAPABILITIES; tests/test_surfaces.py holds it — add the missing half, never edit the assertion.
• Every tool declares honest MCP annotations from core/annotations.py (network_body is a read: reads(...)).
• errors.py decides every status: 400 the caller's, 404 a gone browser or thing, 503 the Grid or its BiDi route, 500 only for what we do not understand. A new status only with its responses block in spec/.
• http/auth.py decides every authorisation (the admin door, signed links over the unprefixed path with hmac.compare_digest); a new door calls it.
• Plain W3C WebDriver and WebDriver BiDi only. No CDP, ever, live tests included.
• Layering is enforced by tests/test_boundaries.py: bidi/ and capture/ are kernel — no fastmcp, starlette, mcp, uvicorn, no selenium, nothing from mcp/, http/, routes, server, spec; websockets has exactly one importer.
• Coerce at the edge (as_bool, as_int): pages="3", capture="false" and "" arrive by hand.
• Docstrings in mcp/tools.py are prompt, written for a model choosing a tool; type hints there are the schema.
• Every text an agent reads names a resource by URI (workspace://console, never a tool name standing in for one).
• CHANGELOG.md [Unreleased] only, one short line per user-visible change; internal work earns no line.
• openapi.yaml is generated and never committed; the wiki is generated (scripts/generate_wiki.py) and tests/test_wiki.py holds it; prose goes in wiki/notes/<tool>.notes.md.
• README.md stays ≤ 25 000 bytes (tests/test_readme.py; it is at 24 976 today).
• Integration tests: no new flow in tests/integration/flows/ (spec §12, AGENTS "less is more").
• No session_id in any tool, body or result; no Grid id in a log line or an error a caller reads.
• Nothing runs a cleanup loop; the only waits are bounded ones for something owed.

The spec is binding; these are the places it is silent, contradicts a standing ruling, or meets the Penpot boards. Each is Claude's, for Dr K on the PR, with what it costs if wrong.

1. network_body is POST /browser/network-body with body {"id": "3.7"}, not the spec's GET /browser/network/{id}/body. §F2.13 Q2 (Dr K's, closed): every browser action is a POST, reads included, so a request body is the tool's arguments and a flow step's args; core/capabilities.py and spec/builder.py are built on it. Cost if wrong: one row's route and method.
2. The intercept rule moves from "never" to "only the spare tab". E2's BidiSocket refuses network.addIntercept; spec §3 puts site data's intercept on the shared connection. Connection refuses nothing; site_data/spare.py is the only module that sends network.addIntercept, scoped to its tab and SPARE_PATH, removed before its listener — held by a source test. Cost: none; the hazard (an intercept outliving its answerer) is unchanged.
3. Capture attaches after a site-data restore and before the first navigation. Ruling 6 holds (nothing the page does is missed), and the restore's stand-in navigations of the main tab happen before anything is subscribed, so they are never captured. Cost: none.
4. call.finished is E2's declared CallFinished (session_id, started, duration in seconds); spec §6's "ms" is read as duration. E2 r8 is the definition both epics cite. Cost: none.
5. Capture.mark returns a counters snapshot Mark(errors, failed) and caused(workspace, mark) subtracts — the spec's seq contract, made immune to eviction. Cost: none.
6. The hint's see is workspace://console?level=error when errors > 0, else workspace://network?filter=failed. Cost: a string.
7. A call that raises carries the hint too, as one sentence on its error (exc.add_note, read by faults.message): "why didn't my action work" is the brief's first job, and a click that 500'd and then timed out has no result to carry a page object. Cost: a note nobody reads, if Dr K wants results only.
8. With capture on, Workspaces.act holds the session's lock for the whole call, so a call's start and end bound its events (ruling 12). The lock is reentrant; a call refused by its own checks then waits its turn behind another call — only with capture on. Cost: a refused call waits, only on a captured, busy session.
9. The boards draw more than the spec builds; the spec wins: no request body section (dataTypes: ["request"] is the spec's Next round), no during or initiator rows (no spec field), "up to 100 kB" not "10 000 characters", ids <nav>.<n> not 1. The UI omits those rows. Cost: Dr K may want during (cheap: capture knows the call window).
10. The admin fetches a body through POST /admin/workspaces/{key}/network/{id}/body (the network-body board has Fetch body; spec §9 names no route). Admin door, never reopens a browser. Cost: one route.
11. Clear is page-local (boards: "nothing was sent to it"): no server endpoint, the page hides what it had. Cost: none.
12. workspace://current and the admin row also carry captured — {"console", "network", "errors", "failed_requests"} since the last navigation — because the show context chips (Console 2 errors, Network 1 failed) and the summary's CAPTURE row (24 requests · 5 console entries) draw them. Counts follow the boards: console counts kept entries; network counts every request seen, static included. Cost: four integers.
13. Tail rows are the normalised, masked rows (static and debug ones included, marked "kept": false), never raw events. Cost: none; masking happens once (ruling 13).
14. at and took_ms come from each event's own timestamp (epoch ms on both browsers), not request.timings (Chrome relative, Firefox epoch); the live task confirms. Cost: a fallback to timings if one browser disagrees.
15. The capture cache holds at most 256 workspaces (WORKSPACES), least recently fed evicted first — TTLCache needs a size. Cost: a constant.
16. network_body is not a flow step (flows/document.NOT_STEPS): a request id belongs to one page load, which a saved flow cannot know. Cost: none.
17. Navigation marks come only from the top-level context present at attach (the tab the agent drives); other top-level tabs' entries are kept but mark nothing; historyUpdated and fragmentNavigated update the current mark's URL without starting a navigation. Cost: an SPA's route change is not a "page".
18. Chrome's replay is aligned only for Chrome (ruling 9); Firefox replays nothing to a later subscriber, so holding its first entries could only drop a real repeat. Cost: none.
19. A Channel failure is a 503: a closed connection or a reply timeout raises faults.BidiUnavailable; a browser error reply raises bidi.BidiError; no such network data becomes faults.NotFound (404) in network_body. Cost: none.
20. E2's deferred socket minors are the six spec §2 lists; no other list of them exists in the repo or on #61/#68, so Task 3 implements exactly those six.

The five inputs the spec implies but no ruling spells out, most likely to bite first. Each has its test in the owning task.

1. A call that fails after causing a page error — a click that triggers a 500 and then times out waiting. Expected: the error the caller reads ends with the hint sentence naming the URI (Task 7, test_a_failed_call_says_what_the_page_reported).
2. A secret the page echoes back — a bound write whose value a GET form puts in the URL, or a page that console.logs the field. Expected: ••• in the buffer, the tail and the request detail; the forms are registered before the keystrokes (Task 5 test_a_bound_value_is_masked_everywhere, Task 7 test_a_bound_write_registers_its_value_before_typing).
3. Two calls on one captured session at once — a flow step and a direct call. Expected: each hint counts only its own call's events (Task 7, test_each_call_counts_only_its_own_window).
4. A body asked for after the browser was reaped and reopened. Expected: 404 "the browser that made request 2.3 has ended", never a 400 or a body from the wrong browser (Task 8, test_a_body_from_an_ended_browser_is_404).
5. A page that floods the console (thousands of entries a second). Expected: the buffer stays at 1 000, the subscriber's overflow is counted, and a site-data command on the same connection is answered on time (Task 2 test_a_blocked_subscriber_does_not_delay_a_reply, Task 5 test_the_console_buffer_keeps_the_newest_thousand).

Files:

• Modify: pyproject.toml (requires-python, classifiers, the test extra's tomli line and its comment, [tool.ruff] target-version), Dockerfile:80-84 (the tomli install and its comment), scripts/requirements.py:25-46, tests/test_packaging.py:15-23,295-308, tests/test_prompts.py:96-101, tests/test_skill.py:35-42, tests/fakes.py:174-191, .github/workflows/test.yml (header comment and the matrix), .github/workflows/publish.yml:7,64, .github/instructions/python.instructions.md (The supported range), .github/copilot-instructions.md:142-143, CONTRIBUTING.md:17,154, AGENTS.md:50,58, CHANGELOG.md

Interfaces:

• Produces: the floor every later task may use — tomllib, except*, asyncio.TaskGroup, typing.Self, BaseException.add_note (Task 7), TimeoutError as asyncio.TimeoutError.

• Step 1: Write the failing test. In tests/test_packaging.py, replace test_the_toml_reader_works_on_the_oldest_python_the_image_can_build with:

def test_the_floor_is_3_11_and_needs_no_toml_backport():
    """tomllib is stdlib from 3.11, the floor (capture spec, ruling 5)."""
    project = tomllib.loads((REPO / "pyproject.toml").read_text())["project"]
    assert project["requires-python"] == ">=3.11"
    assert "Programming Language :: Python :: 3.10" not in project["classifiers"]
    assert not any("tomli" in dep for dep in project["optional-dependencies"]["test"])
    assert "tomli" not in (REPO / "Dockerfile").read_text()
    assert "import tomllib\n" in (REPO / "scripts" / "requirements.py").read_text()
    assert "tomli" not in (REPO / "scripts" / "requirements.py").read_text()

and change the module's import to a plain import tomllib.

• Step 2: Run it to see it fail. python3 -m pytest tests/test_packaging.py -q → FAIL on requires-python.

• Step 3: Raise the floor.

  • pyproject.toml: requires-python = ">=3.11"; drop the 3.10 classifier; drop "tomli; python_version < '3.11'" and its comment from the test extra; target-version = "py311" (keep the comment: the oldest interpreter supported).
  • Dockerfile: delete the tomli pip install line and the four comment lines above it.
  • scripts/requirements.py: import tomllib at the top, no fallback; read() loses the tomllib is None branch.
  • tests/test_prompts.py, tests/test_skill.py: import tomllib, no fallback and no 3.10 comment.
  • tests/fakes.py patch_os: drop the _NormalAccessor branch and the 3.10 paragraph of its docstring; it becomes monkeypatch.setattr(os, name, replacement) with the 3.13 listdir/scandir sentence kept.
  • .github/workflows/test.yml: matrix fromJSON('["3.11", "3.14"]') || fromJSON('["3.11", "3.12", "3.13", "3.14"]'); the header comment says 3.11 is the floor and Test (3.11) is the advisory leg; the paragraph about the 3.10 sweep staying red keeps its lesson with "the floor" in place of "3.10".
  • publish.yml: "the full 3.11 → 3.14 matrix", "the >=3.11 promise".
  • .github/instructions/python.instructions.md: the range is 3.11 → 3.14; except*, TaskGroup, typing.Self, tomllib move to fine; the X | Y paragraph stays.
  • .github/copilot-instructions.md, CONTRIBUTING.md, AGENTS.md (the loop diagram): 3.11 wherever they say 3.10; "advisory" moves to Test (3.11).
  • CHANGELOG.md [Unreleased]: - **BREAKING:** Python 3.11 or newer is required.

• Step 4: Verify. git grep -n -E "3\.10|tomli|_NormalAccessor" -- ':!docs' ':!wiki' ':!CHANGELOG.md' ':!ui' prints nothing; python3 -m pytest -q passes; $D/bin/ruff check . passes (it already does at py311 — measured).

• Step 5: Commit.

git add -A && git commit -m "Python 3.11 is the floor; every 3.10 workaround goes (#64)

Co-Authored-By: Claude Opus 5.5 <[email protected]>"

Files:

• Create: kubed/selenium_flow/bidi/__init__.py, bidi/connection.py, bidi/channel.py, tests/bidi_fakes.py, tests/test_bidi_connection.py, tests/test_bidi_channel.py
• Delete: tests/test_monitor_bidi.py (its cases move to test_bidi_connection.py, intercept case excepted — plan ruling 2)
• Modify: tests/test_boundaries.py, pyproject.toml (packages list)

monitor/bidi.py stays until Task 3 swaps the monitor onto bidi.Connection; the suite stays green between the two.

Interfaces:

• Produces (kubed.selenium_flow.bidi):

OPEN_TIMEOUT = 5.0; REPLY_TIMEOUT = 5.0; PING_INTERVAL = 20.0
MAX_MESSAGE = 16 * 2**20; QUEUE_LIMIT = 1000; LAGS_KEPT = 1000
FLUSH_GRACE = 0.0   # seconds slept between flush's round trip and its barriers; Task 13 sets 0.05 if Verify first 2 fails

class BidiError(Exception):            # the browser answered with an error
    error: str; message: str

Handler = Callable[[str, dict], Awaitable[None] | None]

class Subscriber:
    events: frozenset[str]; name: str; limit: int
    dropped: int          # events lost to a full queue
    peak: int             # deepest the queue has been
    lags: deque[float]    # seconds from the reader receiving an event to its handler starting (last LAGS_KEPT)
    def deliver(self, method: str, params: dict, at: float) -> None   # reader side; never awaits
    async def barrier(self) -> None    # returns once everything queued before it was handled
    def cancel(self) -> None

class Connection:
    def __init__(self, url: str, *, on_close: Callable[[], None] | None = None,
                 connect=None, clock: Callable[[], float] = time.monotonic): ...
    is_open: bool; dropped: bool; unsolicited: int
    subscriptions: dict[str, tuple[tuple[str, ...], tuple[str, ...] | None]]  # id -> (events, contexts)
    collectors: set[str]
    async def open(self) -> None
    async def command(self, method: str, params: dict | None = None, *, timeout: float = REPLY_TIMEOUT) -> dict
    async def subscribe(self, events: Sequence[str], contexts: Sequence[str] | None = None) -> str
    async def unsubscribe(self, subscription: str) -> None
    def listen(self, events: Iterable[str], handler: Handler, *, name: str, limit: int = QUEUE_LIMIT) -> Subscriber
    async def add_collector(self, params: dict) -> str
    async def remove_collector(self, collector: str) -> None
    async def flush(self, timeout: float = REPLY_TIMEOUT) -> None
    async def reopen(self) -> None
    async def close(self) -> None

class Channel:
    def __init__(self, connection: Connection, loop: asyncio.AbstractEventLoop): ...
    def call(self, method: str, params: dict | None = None, *, timeout: float = REPLY_TIMEOUT) -> dict
    def post(self, method: str, params: dict | None = None) -> None
    def listen(self, events: Iterable[str], handler: Callable[[str, dict], None], *,
               contexts: Sequence[str] | None = None, name: str = "channel") -> ContextManager[Subscriber]
    def flush(self, timeout: float = REPLY_TIMEOUT) -> None

class Channels:
    def __init__(self, url_for: Callable[[str], str], *,
                 held: Callable[[str], Connection | None] = lambda _s: None,
                 loop: Callable[[], asyncio.AbstractEventLoop | None] = lambda: None,
                 connect=None): ...
    def open_channel(self, session_id: str) -> ContextManager[Channel]

• Consumes: faults.BidiUnavailable (a plain module; kernel may import it).

Rules the code must keep (ruling 8, spec §1):

• The reader only parses and routes. A reply (id) resolves its future (type: error → BidiError(error, message)); an event (type: event) is handed to every subscriber whose events contain its method via deliver; an event no subscriber asked for is discarded and counted in unsolicited (Firefox fans every event out to every socket). A handler never runs on the reader.

• deliver appends (method, params, at) to a deque and wakes the subscriber's task; a full queue (limit) drops the event, dropped += 1, logged once a streak by subscriber name. peak is the deepest the queue has been.

• Each subscriber's task pops, records clock() - at in lags, calls the handler, and awaits it when it returns an awaitable. A handler that raises is logged at WARNING once a streak by exception type (never the params); the task carries on.

• barrier() enqueues a sentinel that bypasses limit and returns when the task reaches it.

• flush() = one session.status round trip, then FLUSH_GRACE seconds (0 by default), then every live subscriber's barrier(), within timeout: everything the browser sent before the reply has been handled when it returns (Verify first 2 — Task 13 checks it on the Grid).

• subscribe returns the id from {"subscription": id} (BidiError("no subscription id", …) when absent) and records events and contexts; reopen() re-subscribes by events and contexts (ids are per connection) and re-adds nothing else.

• close() never raises: unsubscribes each recorded subscription and removes each recorded collector (network.removeDataCollector({"collector": id})), each bounded by REPLY_TIMEOUT; cancels every subscriber's task; closes the socket; aborts the transport and cancels the reader even when cancelled part-way (E2's BidiSocket.close shape, kept). on_close runs only when the socket closed and we did not close it.

• Connect: websockets.asyncio.client.connect(url, open_timeout=OPEN_TIMEOUT, ping_interval=PING_INTERVAL, ping_timeout=PING_INTERVAL, max_size=MAX_MESSAGE); connect is the test seam. websockets is imported in bidi/connection.py only.

• Channel.call runs command on loop with asyncio.run_coroutine_threadsafe and blocks only the calling worker for timeout + 1 s; called on the loop's own thread it raises RuntimeError (it would deadlock). A BidiError passes through; ConnectionError or TimeoutError becomes faults.BidiUnavailable(CHANNEL_DOWN) with CHANNEL_DOWN = "the browser's BiDi channel did not answer: retry, and if it persists the Grid's /se/bidi route is down" (plan ruling 19).

• Channel.post schedules command without waiting (from the loop: loop.create_task; from a worker: run_coroutine_threadsafe); a failure is logged at debug by type and never raised. It is how a handler answers (the spare tab's provideResponse).

• Channel.listen subscribes (contexts scoped when given) and registers a local subscriber; on exit it unsubscribes by id, then cancels the subscriber — subscribe after listen, so no event is lost between the two.

• Channels.open_channel(session_id): when held(session_id) returns an open connection and loop() a running loop, yields a Channel on them and closes nothing; otherwise opens a Connection(url_for(session_id)) on loop() — or, when there is none (unit tests, a CLI), on a private loop from anyio.from_thread.start_blocking_portal() for the block — and closes it after. An open that fails raises faults.BidiUnavailable(CHANNEL_DOWN).

• Step 1: Move and extend the fake browser. tests/bidi_fakes.py: FakeBrowser from tests/test_monitor_bidi.py, unchanged in what it answers, plus: session.status → {"ready": False, "message": "already connected"}; network.addDataCollector → {"collector": "col-N"}; network.removeDataCollector → {}; network.getData → whatever self.bodies[request] holds ({"bytes": {...}}), else the error no such network data; slow (a command answered after self.delay seconds); events(n, method=...) sends n events in one burst before returning; received_params keeps each command's params. A browser fixture in tests/conftest.py yields one (it was local to the old file).

• Step 2: Write the failing tests. tests/test_bidi_connection.py — the old file's cases on Connection (reply by id, BidiError, a silent command times out with TimeoutError, subscribe records the id, close unsubscribes first, a server-side close fails waiting commands with ConnectionError and calls on_close, reopen re-subscribes by event name), and these:

async def test_an_event_reaches_only_the_subscribers_that_asked(browser):
    conn = Connection(browser.url)
    await conn.open()
    logs, nets = [], []
    conn.listen(["log.entryAdded"], lambda m, p: logs.append(p), name="logs")
    conn.listen(["network.fetchError"], lambda m, p: nets.append(p), name="nets")
    await browser.event("log.entryAdded", {"n": 1})
    await browser.event("browsingContext.load", {"n": 2})   # nobody asked
    await conn.flush()
    assert logs == [{"n": 1}] and nets == [] and conn.unsolicited == 1
    await conn.close()


async def test_a_blocked_subscriber_does_not_delay_a_reply(browser):
    conn = Connection(browser.url)
    await conn.open()
    gate = asyncio.Event()

    async def stuck(_m, _p):
        await gate.wait()

    sub = conn.listen(["log.entryAdded"], stuck, name="stuck", limit=10)
    await browser.events(50, "log.entryAdded")
    started = asyncio.get_running_loop().time()
    assert await conn.command("browsingContext.getTree") == {"contexts": []}
    assert asyncio.get_running_loop().time() - started < 0.5
    assert sub.dropped >= 39 and sub.peak == 10
    gate.set()
    await conn.close()


async def test_flush_returns_after_every_earlier_event_was_handled(browser):
    conn = Connection(browser.url)
    await conn.open()
    seen = []

    async def slow(_m, p):
        await asyncio.sleep(0.01)
        seen.append(p["i"])

    conn.listen(["log.entryAdded"], slow, name="slow")
    for i in range(5):
        await browser.event("log.entryAdded", {"i": i})
    await conn.flush()
    assert seen == [0, 1, 2, 3, 4]
    await conn.close()


async def test_a_raising_handler_is_logged_once_and_the_next_event_still_arrives(browser, caplog):
    conn = Connection(browser.url)
    await conn.open()
    got = []

    def handler(_m, p):
        if p.get("boom"):
            raise KeyError("x")
        got.append(p)

    conn.listen(["log.entryAdded"], handler, name="h")
    for params in ({"boom": 1}, {"boom": 1}, {"ok": 1}):
        await browser.event("log.entryAdded", params)
    await conn.flush()
    assert got == [{"ok": 1}]
    assert len([r for r in caplog.records if "KeyError" in r.getMessage()]) == 1
    await conn.close()


async def test_close_removes_collectors_and_subscriptions_then_closes(browser):
    conn = Connection(browser.url)
    await conn.open()
    await conn.subscribe(["log.entryAdded"])
    await conn.add_collector({"dataTypes": ["response"], "maxEncodedDataSize": 10_000_000, "collectorType": "blob"})
    await conn.close()
    assert sorted(browser.received[-2:]) == ["network.removeDataCollector", "session.unsubscribe"]
    assert not conn.is_open and not conn.dropped


async def test_lags_are_recorded_per_event(browser):
    conn = Connection(browser.url)
    await conn.open()
    sub = conn.listen(["log.entryAdded"], lambda m, p: None, name="l")
    await browser.events(3, "log.entryAdded")
    await conn.flush()
    assert len(sub.lags) == 3 and all(lag >= 0 for lag in sub.lags)
    await conn.close()

tests/test_bidi_channel.py, driving a real loop in a portal thread (anyio.from_thread.start_blocking_portal) the way production's worker threads see the monitor's loop:

def test_call_blocks_only_the_worker_and_returns_the_reply(browser_in_thread):
    url, portal, loop = browser_in_thread
    channels = Channels(lambda _s: url, loop=lambda: loop)
    with channels.open_channel("s") as ch:
        assert ch.call("browsingContext.getTree") == {"contexts": []}


def test_an_error_reply_is_a_bidi_error_and_silence_is_a_503(browser_in_thread):
    url, portal, loop = browser_in_thread
    with Channels(lambda _s: url, loop=lambda: loop).open_channel("s") as ch:
        with pytest.raises(BidiError, match="unknown command"):
            ch.call("nope.nothing")
        with pytest.raises(BidiUnavailable):
            ch.call("silence", timeout=0.1)


def test_a_held_connection_is_lent_and_not_closed(browser_in_thread):
    url, portal, loop = browser_in_thread
    held = portal.call(_opened, url)
    channels = Channels(lambda _s: "ws://unused", held=lambda _s: held, loop=lambda: loop)
    with channels.open_channel("s") as ch:
        ch.call("browsingContext.getTree")
    assert held.is_open


def test_without_a_loop_a_private_one_serves_the_block(browser_in_thread):
    url, _portal, _loop = browser_in_thread
    with Channels(lambda _s: url).open_channel("s") as ch:
        assert ch.call("browsingContext.getTree") == {"contexts": []}


def test_an_unreachable_route_is_a_503():
    with pytest.raises(BidiUnavailable), Channels(lambda _s: "ws://127.0.0.1:9/x").open_channel("s"):
        pass


def test_listen_subscribes_scoped_and_unsubscribes_on_exit(browser_in_thread):
    url, portal, loop = browser_in_thread
    got = []
    with Channels(lambda _s: url, loop=lambda: loop).open_channel("s") as ch:
        with ch.listen(["network.beforeRequestSent"], lambda m, p: got.append(p), contexts=["tab-1"]):
            portal.call(_event, "network.beforeRequestSent", {"context": "tab-1"})
            ch.flush()
    assert got == [{"context": "tab-1"}]
    assert {"events": ["network.beforeRequestSent"], "contexts": ["tab-1"]} in browser_params("session.subscribe")
    assert "session.unsubscribe" in browser_methods()


def test_post_never_raises(browser_in_thread, caplog):
    url, portal, loop = browser_in_thread
    conn = portal.call(_opened, url)
    portal.call(conn.close)
    caplog.set_level(logging.DEBUG, logger="kubed.selenium_flow.bidi")
    assert Channel(conn, loop).post("network.provideResponse", {"request": "r-1"}) is None
    portal.call(asyncio.sleep, 0.05)
    assert any("ConnectionError" in r.getMessage() for r in caplog.records)

(browser_in_thread is a fixture in tests/bidi_fakes.py that starts FakeBrowser inside a blocking portal and yields (url, portal, loop); _opened, _event, browser_params, browser_methods are its helpers — write them there, once.)

• Step 3: Run them to see them fail. python3 -m pytest tests/test_bidi_connection.py tests/test_bidi_channel.py -q → ModuleNotFoundError: kubed.selenium_flow.bidi.

• Step 4: Implement bidi/connection.py, bidi/channel.py, bidi/__init__.py to the rules above. The module docstring of connection.py carries E2's two measured facts (a held socket outlives its session; a ping proves the hub, never the browser) and ruling 8's reason (a slow subscriber on the receive path starves replies).

• Step 5: Boundaries. In tests/test_boundaries.py: *walk("bidi") in NO_PROTOCOL; "bidi.connection", "bidi.channel" in NO_SELENIUM; "bidi" in KERNEL; WEBSOCKETS_HOME = "bidi/connection.py" — and, until Task 3 deletes it, test_only_the_monitor_holds_a_socket becomes test_one_module_imports_websockets asserting websockets_importers() <= {"bidi/connection.py", "monitor/bidi.py"} with a comment that Task 3 narrows it to the first. pyproject.toml: kubed.selenium_flow.bidi in the packages list.

• Step 6: Verify. python3 -m pytest tests/test_bidi_connection.py tests/test_bidi_channel.py tests/test_boundaries.py -q passes; then the whole suite.

• Step 7: Commit. git commit -m "bidi/: one connection per browser, a queue and a task per subscriber (#64)" with the co-author line.

Files:

• Modify: kubed/selenium_flow/monitor/monitor.py, monitor/__init__.py, tests/test_monitor.py, tests/test_boundaries.py
• Delete: kubed/selenium_flow/monitor/bidi.py
• Create: tests/test_monitor_capture.py

Interfaces:

• Consumes: bidi.Connection (Task 2), faults.BidiUnavailable.
• Produces:

# monitor/monitor.py
ATTACH_TIMEOUT = 15.0
BACKOFF_MAX = 30.0          # reconnect back-off: 1, 2, 4 … 30 s
ATTACH_FAILED = ("capture needs the browser's BiDi channel and the Grid's /se/bidi route "
                 "did not answer, so this browser was ended: retry, or open without capture")

class Sink(Protocol):
    def attached(self, workspace: str, session_id: str, browser: str, top: str | None, *, again: bool) -> None: ...
    def ingest(self, workspace: str, session_id: str, method: str, params: dict) -> None: ...
    def replayed(self, workspace: str, session_id: str) -> None: ...
    def detached(self, workspace: str, session_id: str) -> None: ...

@dataclass
class Watch:  # E2's, with: connection: Connection | None; failures: int = 0; retry_at: float = 0.0;
              # subscribed: frozenset[str] = frozenset(); attached_once: bool = False
              # (`socket` is renamed `connection`)

class Monitor:
    def __init__(self, bus, *, listing, socket_url=None, events_for=None, collectors_for=None,
                 connect=None, sink: Sink | None = None, clock=time.time, tick=LISTING_TICK): ...
    def attach(self, workspace: str, session_id: str, reason: str, *, browser: str = "",
               grid_timeout: int | None = None) -> None        # any worker thread; raises BidiUnavailable
    def connection(self, session_id: str) -> Connection | None  # any thread: the held, open one
    def capturing(self, session_id: str) -> bool                # any thread
    @property
    def loop(self) -> asyncio.AbstractEventLoop | None
    # E2's opened / ended / watch / release / touch / watching / look / start / stop stay;
    # `on_bidi` and `socket_held` are gone (sink.ingest; capturing).

Behaviour (spec §2):

• Holding a connection (_hold(w), used by attach and by look for a watch owed events): connect(url, on_close=self._wake); await open(); browsingContext.getTree({"maxDepth": 0}) → top = the first context's id or None; sink.attached(workspace, session_id, browser, top, again=w.attached_once); connection.listen(wanted, deliver, name="capture") where deliver calls sink.ingest(workspace, session_id, method, params); await subscribe(wanted); await add_collector(params) for each owed reason in collectors_for; when again, await flush() then sink.replayed(...). Any failure — CancelledError included — closes the connection (shielded) and calls sink.detached, then re-raises (attach) or counts a failure (look).

• attach from a worker thread: with the monitor started, schedules _attach on the loop with asyncio.run_coroutine_threadsafe and waits ATTACH_TIMEOUT; _attach adds reason to the watch (creating it, as watch does), then _holds it now rather than at the next look, and _ensures the task. A timeout cancels the future (so _hold's cleanup runs) and, like any failure, raises BidiUnavailable(ATTACH_FAILED); the watch's reason is released. Not started → RuntimeError("the monitor is not running").

• E2's deferred minors (spec §2): (1) a reason added to a watch that holds a connection subscribes the events it adds (wanted - w.subscribed) on the next look; (2) a cancelled hold leaks nothing (above); (3) the failure-logged set is keyed by session_id, discarded on success and when the watch drops; (4) a hold that fails sets w.retry_at = now + min(BACKOFF_MAX, 2 ** w.failures) (then failures += 1) and look skips it until then; success resets both; (5) stop() iterates copies of watches and _closing; (6) a Monitor that raises never fails a caller — enforced in Task 6 at Workspaces, tested there.

• Deadline (programme R3): past w.deadline the connection is closed clean (Connection.close unsubscribes and removes the collector — ruling 7), w.connection = None, sink.detached(...); capturing reads false; a touch brings it back on the next look with again=True.

• A dropped connection (connection.dropped): as E2 (one miss, looked at now); the next hold is again=True.

• _drop (the watch ends): closes the connection in a task stop waits for and calls sink.detached.

• capturing(session_id) is bool(w and w.connection and w.connection.is_open) read without the loop (a dict get and two attribute reads).

• Step 1: Port E2's monitor tests. In tests/test_monitor.py, FakeSocket becomes FakeConnection with Connection's interface (open, command answering browsingContext.getTree with {"contexts": [{"context": "top-1"}]}, subscribe recording events, listen recording the handler, add_collector returning "col-1" and recording its params in collectors, flush, close setting closed and, when it unsubscribed and removed its collectors first, closed_clean, is_open, dropped, drop(), and emit(method, params) which runs the registered listen handler; class switches refuse (open raises ConnectionError) and hang_on_subscribe (subscribe never returns), reset per test; made lists every instance); socket_held reads become capturing; on_bidi reads become a FakeSink recording attached/ingest/replayed/detached calls. Every E2 assertion keeps its meaning.

• Step 2: Write the failing tests. tests/test_monitor_capture.py:

EVENTS = ("log.entryAdded", "network.beforeRequestSent")
COLLECTOR = {"dataTypes": ["response"], "maxEncodedDataSize": 10_000_000, "collectorType": "blob"}


def monitor_for(sink, clock):
    return Monitor(LocalBus(), listing=lambda: (1, {GID: 300}), socket_url=lambda s: f"ws://x/{s}",
                   events_for={"capture": EVENTS}, collectors_for={"capture": COLLECTOR},
                   connect=FakeConnection, sink=sink, clock=clock)


async def test_attach_holds_subscribes_and_collects_before_returning(running_monitor):
    monitor, sink = running_monitor
    await asyncio.to_thread(monitor.attach, "ws1", GID, "capture", browser="chrome")
    [conn] = FakeConnection.made
    assert conn.subscribed == [list(EVENTS)] and conn.collectors == [COLLECTOR]
    assert sink.calls[0] == ("attached", "ws1", GID, "chrome", "top-1", False)
    assert monitor.capturing(GID)


async def test_a_failed_attach_raises_503_and_holds_nothing(running_monitor):
    monitor, sink = running_monitor
    FakeConnection.refuse = True
    with pytest.raises(BidiUnavailable, match="/se/bidi"):
        await asyncio.to_thread(monitor.attach, "ws1", GID, "capture")
    assert not monitor.capturing(GID) and monitor.watching(GID) == frozenset()
    assert ("detached", "ws1", GID) in sink.calls


async def test_the_deadline_closes_clean_and_a_touch_reattaches_as_again(running_monitor, clock):
    monitor, sink = running_monitor
    await asyncio.to_thread(monitor.attach, "ws1", GID, "capture", grid_timeout=300)
    clock.now += 301
    await monitor.look()
    assert not monitor.capturing(GID) and FakeConnection.made[0].closed_clean
    monitor.touch(GID)
    await monitor.look()
    assert monitor.capturing(GID)
    assert sink.calls[-2:] == [("attached", "ws1", GID, "", "top-1", True), ("replayed", "ws1", GID)]


async def test_a_reason_added_later_subscribes_what_it_adds(running_monitor):
    monitor, _ = running_monitor
    monitor.events_for["other"] = ("browsingContext.load",)
    await asyncio.to_thread(monitor.attach, "ws1", GID, "capture")
    monitor.watch(GID, "ws1", "other")
    await monitor.look()
    assert FakeConnection.made[0].subscribed[-1] == ["browsingContext.load"]


async def test_reconnects_back_off_one_two_four_to_thirty(running_monitor, clock):
    monitor, _ = running_monitor
    monitor.watch(GID, "ws1", "capture")
    FakeConnection.refuse = True
    tries = []
    for _ in range(8):
        before = len(FakeConnection.made)
        await monitor.look()
        tries.append(len(FakeConnection.made) - before)
        clock.now += 1
    assert tries == [1, 1, 0, 1, 0, 0, 0, 1]   # retries at +1, +2, +4 …


async def test_a_cancelled_hold_leaks_nothing(running_monitor):
    monitor, sink = running_monitor
    FakeConnection.hang_on_subscribe = True
    task = asyncio.create_task(monitor._attach("ws1", GID, "capture", "", None))
    await asyncio.sleep(0.01)
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    assert FakeConnection.made[0].closed and ("detached", "ws1", GID) in sink.calls


async def test_the_sink_hears_every_event_with_its_workspace(running_monitor):
    monitor, sink = running_monitor
    await asyncio.to_thread(monitor.attach, "ws1", GID, "capture")
    await FakeConnection.made[0].emit("log.entryAdded", {"text": "hi"})
    assert ("ingest", "ws1", GID, "log.entryAdded", {"text": "hi"}) in sink.calls

(running_monitor starts the bus and monitor on the test's loop and yields (monitor, sink); clock is a settable fake; FakeConnection.refuse / hang_on_subscribe are class switches reset per test.)

• Step 3: Run them to see them fail. python3 -m pytest tests/test_monitor.py tests/test_monitor_capture.py -q → FAIL (attach, capturing, sink missing).

• Step 4: Implement the monitor changes above; delete monitor/bidi.py; monitor/__init__.py exports Sink and drops BidiSocket/BidiError (re-exported from bidi where anything imported them). The module docstring's "A socket is a channel" paragraph keeps its words with connection.

• Step 5: Narrow the boundary. test_one_module_imports_websockets asserts websockets_importers() == {"bidi/connection.py"} and is renamed back to test_only_the_bidi_client_holds_a_socket; "monitor.bidi" leaves NO_SELENIUM.

• Step 6: Verify. python3 -m pytest tests/test_monitor.py tests/test_monitor_capture.py tests/test_monitor_events.py tests/test_boundaries.py tests/test_recording_collector.py -q, then the whole suite.

• Step 7: Commit. "The monitor holds a bidi.Connection, attaches before the first page, closes clean (#64)".

Files:

• Modify: kubed/selenium_flow/site_data/spare.py, site_data/transfer.py, core/actions.py (open_session's restore, save_site_data), core/browser.py (delete Grid.bidi, BIDI_TIMEOUT, BIDI_INTERVAL, and websocket_timeout/websocket_interval in reconnect; keep options.enable_bidi = True — the Grid's /se/bidi route needs webSocketUrl), server.py (self.actions.channels = Channels(...)), tests/site_data_fakes.py, tests/fakes.py (FakeBidi → FakeChannel), tests/test_spare_tab.py, tests/test_site_data_restore.py, tests/test_site_data_save.py, tests/test_site_data_views.py, tests/test_site_data_workspaces.py, tests/test_boundaries.py

Interfaces:

• Consumes: bidi.Channel, bidi.Channels (Task 2).
• Produces:

# site_data/spare.py
def spare_tab(channel: Channel, context: str | None = None, *,
              hidden: Callable[[str], ContextManager[None]] | None = None) -> ContextManager[Callable[[str, str], object]]
# site_data/transfer.py
def capture(channel: Channel, driver, origins=(), *, hidden=None) -> dict
def restore(channel: Channel, data: dict, now: float) -> dict
# core/actions.py
Actions.channels: Channels | None        # wired by the server; a test sets a fake
Actions.hide: Callable[[str, str], ContextManager[None]] | None   # (session_id, context) -> block; Task 6 wires capture.hidden

The BiDi commands, one for one with what Selenium sent (behaviour, results and every site-data assertion unchanged — spec §3):

• capture()'s jar read failing with BidiUnavailable stays a 503 (BIDI_DOWN, today's message); a BidiError there is the browser refusing — re-raised as BidiUnavailable(BIDI_DOWN) as well, since nothing can be saved without the jar.

• spare_tab with hidden and no context wraps the tab's whole life in hidden(tab) right after browsingContext.create — the save's spare tab is invisible to capture (ruling 15). A restore needs none (plan ruling 3).

• Actions.save_site_data and open_session's restore use with self.channels.open_channel(session_id) as channel:; the save passes hidden=functools.partial(self.hide, session_id) when self.hide is set.

• server.py: self.actions.channels = Channels(self.grid.bidi_url, held=self.monitor.connection, loop=lambda: self.monitor.loop).

• Step 1: Rewrite the fake. tests/fakes.py: FakeChannel records every call in log as (method, params) (params(method) returns the last one's) and answers each command above from its state (cookies, a tab counter giving spare-1, spare-2…, value for script.evaluate, refuse for a method that answers BidiError, down for one that raises BidiUnavailable); listen(events, handler, contexts=…) logs ("session.subscribe", {"events": […], "contexts": […]}), stores handlers[event] = handler, and on exit logs ("session.unsubscribe", …); post(method, params) appends (method, params) to posts. It replaces FakeBidi. tests/site_data_fakes.py: bidi_cookie returns the BiDi dict shape; bidi_cm(fake) becomes channels_for(fake) returning an object with open_channel(session_id) yielding the fake.

• Step 2: Rewrite the spare-tab contract test (tests/test_spare_tab.py), keeping each test's assertion about order and answers:

def test_a_spare_tab_stands_on_an_origin_runs_there_and_closes():
    ch = FakeChannel(value={"theme": "dark"})
    with spare_tab(ch) as run:
        assert run("https://app.test", "localStorage.length") == {"theme": "dark"}
    assert [m for m, _ in ch.log] == [
        "browsingContext.create", "network.addIntercept", "session.subscribe",
        "browsingContext.navigate", "script.evaluate",
        "network.removeIntercept", "session.unsubscribe", "browsingContext.close",
    ]
    assert ch.params("network.addIntercept") == {
        "phases": ["beforeRequestSent"], "contexts": ["spare-1"],
        "urlPatterns": [{"type": "pattern", "pathname": SPARE_PATH}],
    }


def test_every_blocked_request_is_answered_with_the_marked_page():
    ch = FakeChannel()
    with spare_tab(ch):
        answer = ch.handlers["network.beforeRequestSent"]
        answer("network.beforeRequestSent", {"isBlocked": True, "request": {"request": "r-1"}})
        answer("network.beforeRequestSent", {"isBlocked": False, "request": {"request": "r-2"}})
    assert ch.posts == [("network.provideResponse", {
        "request": "r-1", "statusCode": 200,
        "headers": [{"name": "content-type", "value": {"type": "string", "value": "text/html"}}],
        "body": {"type": "string", "value": SPARE_PAGE},
    })]


def test_the_spare_tab_is_hidden_from_capture_for_its_whole_life():
    ch, seen = FakeChannel(), []

    @contextmanager
    def hidden(tab):
        seen.append(("in", tab)); yield; seen.append(("out", tab))

    with spare_tab(ch, hidden=hidden) as run:
        run("https://app.test", "1")
        assert seen == [("in", "spare-1")]
    assert seen == [("in", "spare-1"), ("out", "spare-1")]

and the existing tests' other cases (service worker, wrong origin, a failed teardown step still runs the rest) against FakeChannel.

• Step 3: Run the site-data tests to see them fail. python3 -m pytest tests/test_spare_tab.py tests/test_site_data_save.py tests/test_site_data_restore.py tests/test_site_data_views.py tests/test_site_data_workspaces.py -q.

• Step 4: Implement spare.py, transfer.py, the two Actions call sites, Grid deletions, and the server wiring. transfer.py no longer imports selenium or websocket.

• Step 5: Boundaries. "site_data.spare", "site_data.transfer" join NO_SELENIUM; new test:

def test_only_the_spare_tab_adds_an_intercept():
    """An intercept outlives the connection that added it and hangs every request
    it matches with nobody to answer (plan ruling 2): one module may add one."""
    found = {
        path.relative_to(PACKAGE).as_posix()
        for path in PACKAGE.rglob("*.py")
        if "network.addIntercept" in path.read_text()
    }
    assert found == {"site_data/spare.py"}

• Step 6: Verify. git grep -n -E "selenium\.webdriver\.common\.bidi|websocket import|Grid\.bidi\(|grid\.bidi\(" kubed prints nothing; the whole suite passes.

• Step 7: Commit. "Site data speaks BiDi through our Channel; Selenium's client leaves the path (#64)".

Files:

• Create: kubed/selenium_flow/capture/__init__.py, capture/masking.py, capture/merge.py, capture/align.py, capture/store.py, capture/views.py, tests/capture_events.py (event builders), tests/test_capture_masking.py, tests/test_capture_merge.py, tests/test_capture_align.py, tests/test_capture_store.py, tests/test_capture_views.py
• Modify: tests/test_boundaries.py (*walk("capture") in NO_PROTOCOL, every capture.* module in NO_SELENIUM, "capture" in KERNEL), pyproject.toml (packages)

Nothing outside capture/ changes in this task: the package is complete and tested on its own, and Task 6 plugs it in.

Interfaces:

• Consumes: nothing of ours but the standard library and cachetools. It implements Task 3's Sink.
• Produces:

# capture/__init__.py — the constants, verbatim from the spec
EVENTS = ("log.entryAdded", "network.beforeRequestSent", "network.responseCompleted",
          "network.fetchError", "browsingContext.contextCreated", "browsingContext.contextDestroyed",
          "browsingContext.navigationStarted", "browsingContext.navigationCommitted",
          "browsingContext.historyUpdated", "browsingContext.fragmentNavigated")
COLLECTOR = {"dataTypes": ["response"], "maxEncodedDataSize": 10_000_000, "collectorType": "blob"}
CONSOLE_LIMIT = 1000; REQUEST_LIMIT = 1000; NAVIGATIONS = 5; DEFAULT_TTL = 10_800
TEXT_CAP = 4096; HEADER_LIMIT = 32; HEADER_VALUE_CAP = 4096; TAIL_QUEUE = 500; BODY_CAP = 100_000
REPLAY_DEPTH = 100          # Chrome replays the last 100 per browsing context (measured)
WORKSPACES = 256            # plan ruling 15
MASK = "•••"
MASKED_HEADERS = frozenset({"authorization", "proxy-authorization", "cookie", "set-cookie"})
STATIC_TYPES = frozenset({"stylesheet", "image", "font", "script", "media"})
HINTED_TYPES = frozenset({"document", "fetch", "xhr"})
LEVELS = ("all", "error", "warn"); FILTERS = ("all", "failed", "xhr"); MAX_PAGES = NAVIGATIONS
CONSOLE_URI = "workspace://console"; NETWORK_URI = "workspace://network"
OFF = {"capture": False, "how": "open_session(capture=true)"}

# capture/masking.py
def strip_query(url: str) -> str
class Masker:
    def __init__(self, secrets: Iterable[str] = ()): ...
    def add(self, forms: Iterable[str]) -> None
    def text(self, value: str) -> str
    def url(self, value: str, *, keep_query: bool) -> str
    def headers(self, raw: list) -> list[dict]        # [{"name", "value"}]

# capture/merge.py
Key = tuple[str, float, str, str, str, str]          # (realm, timestamp, type, level, text, first frame)
def entry_key(params: dict) -> Key
def console_entry(params: dict, masker: Masker) -> dict   # {level, type, text, source: {url, line} | None, context, at}
def resource_type(request: dict, mime: str = "", navigation: bool = False) -> str
def counts_in_hint(row: dict) -> bool
def is_static_success(row: dict) -> bool

# capture/align.py
def overlap(seen: Sequence, replay: Sequence) -> int

# capture/store.py
class Mark(NamedTuple): errors: int; failed: int
class Located(NamedTuple): session_id: str; request: str; mime: str; size: int | None
class TailSubscription:
    kind: Literal["console", "network"]; dropped: int
    async def next(self) -> dict                       # a row, or {"dropped": total} when that moved
    def close(self) -> None
class Capture:                                         # implements monitor.Sink
    def __init__(self, *, ttl: int = DEFAULT_TTL, clock: Callable[[], float] = time.time): ...
    def attached(self, workspace, session_id, browser, top, *, again: bool) -> None
    def ingest(self, workspace, session_id, method, params) -> None
    def replayed(self, workspace, session_id) -> None
    def detached(self, workspace, session_id) -> None
    def discard(self, workspace: str) -> None
    def secret(self, workspace: str, forms: Iterable[str]) -> None
    def hidden(self, session_id: str, context: str) -> ContextManager[None]
    def mark(self, workspace: str) -> Mark | None
    def caused(self, workspace: str, since: Mark) -> dict          # {"errors": n, "failed_requests": n}
    def console(self, workspace: str, *, level: str = "all", pages: int = 1) -> dict | None
    def network(self, workspace: str, *, filter: str = "all", pages: int = 1) -> dict | None
    def request(self, workspace: str, id: str) -> dict | None
    def counts(self, workspace: str) -> dict | None   # {"console", "network", "errors", "failed_requests"}, since the last navigation
    def request_for(self, session_id: str, id: str) -> Located | None
    def tail(self, workspace: str, kind: str) -> TailSubscription  # on the loop only

# capture/views.py
def check_level(level) -> str; def check_filter(value) -> str; def check_pages(value) -> int   # ValueError naming the allowed values; coerce "3" -> 3

The rules (spec §4, rulings 3, 4, 9, 10, 13, 15; plan rulings 12–15, 17, 18):

• One lock: ingest runs on the loop, reads run on worker threads; every method takes self._lock (threading.Lock). The cache is cachetools.TTLCache(maxsize=WORKSPACES, ttl=ttl, timer=clock) of Buffers, and every ingest that lands re-assigns cache[workspace], which is what slides the TTL. discard deletes the entry.
• A Buffer holds: console: deque(maxlen=CONSOLE_LIMIT), requests: deque(maxlen=REQUEST_LIMIT), marks: deque(maxlen=NAVIGATIONS) of {nav, url, at}, nav: int (0 before the first navigation), per-nav counters seen, static, debug, errors, failed (dicts keyed by nav, pruned to the marked navs), running totals errors_total, failed_total (for Mark), in_flight: OrderedDict[(request, redirectCount) -> row] (capped at REQUEST_LIMIT, oldest dropped), the index {id -> Located} (dropped with its row), recent: dict[context, deque(maxlen=REPLAY_DEPTH)] of Keys, holding: dict[context, list[params]] | None, masker, session_id, browser, top, seq and a per-nav request counter.
• attached(again=False) (a new session, or the first attach): session_id, browser, top set; the masker is a fresh Masker() (secrets are the session's — ruling 13); recent and holding cleared; self._sessions[session_id] = workspace. again=True keeps all of it and, only for browser == "chrome", starts holding = {} (plan ruling 18).
• ingest: drops any event whose context (params["context"], or params["source"]["context"] for a log entry) is in self._hidden[session_id] — the spare tab is never kept, counted or tailed (ruling 15). Then by method:
  • log.entryAdded: while holding is not None, appended to holding[context]; otherwise landed: its Key goes to recent[context]; debug → debug[nav] += 1 and tailed with "kept": False; else console_entry gets seq, nav, is appended and tailed; level == "error" or type == "javascript" → errors[nav] += 1, errors_total += 1.
  • network.beforeRequestSent: a row {id: f"{nav}.{n}", seq, at: timestamp/1000, method, url, status: None, status_text: "", type, mime: "", size: None, took_ms: None, failed: False, error: None, context, nav, navigation: params.get("navigation") is not None and context == top, request_headers}; seen[nav] += 1. A row whose type is already static is held in in_flight only (it may still fail); any other row is appended to requests now (so an in-flight request is visible) and indexed.
  • network.responseCompleted: completes its row: status, status_text, mime (response.mimeType), size = response.bodySize if present else response.bytesReceived (never content.size), response_headers, took_ms = round(timestamp - begin_timestamp), type re-derived with the MIME when it was "other". If is_static_success(row): static[nav] += 1, removed from requests/index if it was there, tailed with "kept": False. Else kept (appended if it was only in flight), indexed with Located(session_id, request, mime, size), tailed. counts_in_hint(row) → failed[nav] += 1, failed_total += 1.
  • network.fetchError: failed=True, error = errorText, took_ms; always kept (every failure); hinted when its type is in HINTED_TYPES.
  • browsingContext.navigationStarted of top only: nav += 1, a mark {nav, url (masked, query kept), at}; every older row's and mark's URL goes through strip_query, and so do their referer/location/content-location header values (URLs keep their query only on the newest navigation). navigationCommitted, historyUpdated, fragmentNavigated of top update the current mark's url. contextCreated with no parent sets top when it is None. Anything of another context marks nothing (plan ruling 17).
• replayed: for each held context, j = overlap(list(recent[ctx]), [entry_key(p) for p in held]), and only held[j:] lands, in order; then holding = None. detached drops anything held (no alignment possible) and sets holding = None.
• resource_type(request, mime, navigation): navigation → "document"; destination document/iframe/frame → document; style → stylesheet; image → image; font → font; script/worker/sharedworker/serviceworker → script; audio/video/track → media; else initiatorType fetch → fetch, xmlhttprequest → xhr; else by MIME: text/css → stylesheet, image/* → image, font/*/application/font* → font, *javascript* → script, audio/*/video/* → media, text/html → document; else "other".
• is_static_success(row) = row["type"] in STATIC_TYPES and not row["failed"] and row["status"] is not None and 200 <= row["status"] < 400. counts_in_hint(row) = row["type"] in HINTED_TYPES and (row["failed"] or (row["status"] or 0) >= 400).
• console_entry: text = masker.text(params.get("text") or "")[:TEXT_CAP] (mask first, so a secret straddling the cut is still gone); source from stackTrace.callFrames[0] (url masked with keep_query=True, line = lineNumber + 1) or None; at = timestamp / 1000; args are never kept (their text is already in text).
• entry_key(params) = (source.realm or "", timestamp, type, level, text, f"{url}:{lineNumber}:{columnNumber}" of the first frame or "") — on the raw params, before masking.
• Masker.headers(raw): the first HEADER_LIMIT entries; a name in MASKED_HEADERS (case-insensitive) → MASK; else the value ({"type": "string", "value": v} → v; base64 → "(binary)") masked by text, then cut to HEADER_VALUE_CAP. request.cookies and goog:postData are never read, so they never reach a row (ruling 13).
• Masker.text replaces each secret form (longest first, empty ones ignored) with MASK; Masker.url is text then strip_query unless keep_query.
• mark returns Mark(errors_total, failed_total) or None with no buffer; caused subtracts.
• request_for(session_id, id): the workspace from self._sessions[session_id], then its index; None when either is unknown.
• Tails: tail(workspace, kind) registers a TailSubscription with an asyncio.Queue(maxsize=TAIL_QUEUE); each tailed row is {**row, "kept": bool} without headers (put_nowait, a full queue → dropped += 1); next() returns {"dropped": dropped} first whenever dropped changed since it last said, else the next row; close() unregisters. A tail exists only while a page listens: nothing is buffered for a kind nobody tails.

The views (capture/views.py, called under the lock): a window is the marks with nav > current - pages; before the first navigation it is nav 0. Newest first.

console -> {"capture": True, "pages": [{nav, url, at}...], "entries": [...], "count": len(entries),
            "total": kept in window before the level filter, "not_kept": {"debug": n}}
network -> {"capture": True, "pages": [...], "requests": [rows without headers...], "count": n,
            "total": kept in window before the filter, "not_kept": {"static": n}}
request -> {**row, "uri": f"workspace://network/{id}"}    # headers included

level: all (every kept entry), error (level == "error" or type == "javascript"), warn (level == "warn"). filter: all, failed (failed or status >= 400), xhr (type in {"xhr", "fetch"}). pages: as_int-style coercion, 1–5, else ValueError("pages is 1 to 5: how many of the last navigations to show"); a bad level or filter → ValueError naming the allowed values.

• Step 1: Event builders. tests/capture_events.py — log(text, level="info", type="console", ctx="top-1", ts=1000.0, realm="r1", frame=("https://a.test/app.js", 141, 3)), nav_started(url, ctx="top-1", ts=…), req(rid, url, *, ctx="top-1", dest="", initiator="fetch", method="GET", headers=(), nav=None, redirect=0, ts=…), done(rid, status, *, mime="application/json", body_size=10, headers=(), redirect=0, ts=…), fail(rid, error="net::ERR_FAILED", ts=…), each returning (method, params) in BiDi's shape (headers as [{"name", "value": {"type": "string", "value"}}]).

• Step 2: Write the failing tests.

tests/test_capture_masking.py:

@pytest.mark.parametrize("name", ["Authorization", "proxy-authorization", "COOKIE", "Set-Cookie"])
def test_credential_headers_are_masked_in_any_case(name):
    assert Masker().headers([{"name": name, "value": {"type": "string", "value": "s3cr3t"}}]) == [
        {"name": name, "value": MASK}]


def test_a_bound_value_is_masked_everywhere():
    m = Masker(["hunter2", "hunter%32"])
    assert m.text("pw=hunter2 and hunter%32") == f"pw={MASK} and {MASK}"
    assert m.url("https://a.test/login?pw=hunter2", keep_query=True) == f"https://a.test/login?pw={MASK}"
    assert m.headers([{"name": "x-echo", "value": {"type": "string", "value": "hunter2"}}])[0]["value"] == MASK


def test_a_secret_straddling_the_text_cap_is_still_masked():
    m = Masker(["S" * 10])
    entry = console_entry(log("x" * (TEXT_CAP - 5) + "S" * 10)[1], m)
    assert "S" not in entry["text"] and len(entry["text"]) <= TEXT_CAP


def test_headers_are_capped_at_32_of_4_kb():
    raw = [{"name": f"h{i}", "value": {"type": "string", "value": "v" * 5000}} for i in range(40)]
    kept = Masker().headers(raw)
    assert len(kept) == HEADER_LIMIT and all(len(h["value"]) == HEADER_VALUE_CAP for h in kept)


def test_strip_query_keeps_origin_and_path():
    assert strip_query("https://a.test/p/q?code=abc#frag") == "https://a.test/p/q"

tests/test_capture_merge.py:

@pytest.mark.parametrize("request, mime, navigation, kind", [
    ({"destination": "style"}, "", False, "stylesheet"),
    ({"destination": "image"}, "", False, "image"),
    ({"destination": "font"}, "", False, "font"),
    ({"destination": "script"}, "", False, "script"),
    ({"destination": "video"}, "", False, "media"),
    ({"destination": "", "initiatorType": "fetch"}, "", False, "fetch"),
    ({"destination": "", "initiatorType": "xmlhttprequest"}, "", False, "xhr"),
    ({"destination": "document"}, "", False, "document"),
    ({}, "", True, "document"),
    ({}, "text/css", False, "stylesheet"),
    ({}, "image/png", False, "image"),
    ({}, "application/json", False, "other"),
])
def test_resource_type(request, mime, navigation, kind):
    assert resource_type(request, mime, navigation) == kind


@pytest.mark.parametrize("row, static, hinted", [
    ({"type": "image", "failed": False, "status": 200}, True, False),
    ({"type": "font", "failed": False, "status": 304}, True, False),
    ({"type": "image", "failed": False, "status": 404}, False, False),   # kept, not hinted (R10)
    ({"type": "image", "failed": True, "status": None}, False, False),
    ({"type": "fetch", "failed": False, "status": 500}, False, True),
    ({"type": "xhr", "failed": True, "status": None}, False, True),
    ({"type": "document", "failed": False, "status": 404}, False, True),
    ({"type": "fetch", "failed": False, "status": 200}, False, False),
])
def test_what_is_kept_and_what_is_hinted(row, static, hinted):
    assert is_static_success(row) is static and counts_in_hint(row) is hinted


def test_the_entry_key_is_the_six_fields_of_ruling_9():
    _, p = log("boom", level="error", ts=1700.0, realm="r9", frame=("https://a.test/a.js", 4, 2))
    assert entry_key(p) == ("r9", 1700.0, "console", "error", "boom", "https://a.test/a.js:4:2")

tests/test_capture_align.py:

@pytest.mark.parametrize("seen, replay, expected", [
    ([], ["a", "b"], 0),
    (["a", "b", "c"], ["a", "b", "c"], 3),             # nothing new
    (["a", "b", "c"], ["b", "c", "d"], 2),             # one new, the oldest scrolled off
    (["x", "a", "a"], ["a", "a", "a"], 2),             # identical tuples are real entries
    (["a", "b"], ["c", "d"], 0),
])
def test_overlap_is_the_longest_prefix_of_the_replay_that_ends_what_was_seen(seen, replay, expected):
    assert overlap(seen, replay) == expected

tests/test_capture_store.py (a fake clock; feed(capture, *events) calls ingest("ws", "s1", m, p) for each):

def test_the_console_buffer_keeps_the_newest_thousand(capture):
    attach(capture)
    feed(capture, nav_started("https://a.test/"), *[log(f"m{i}", ts=1000.0 + i) for i in range(1500)])
    view = capture.console("ws", pages=5)
    assert view["total"] == CONSOLE_LIMIT and view["entries"][0]["text"] == "m1499"


def test_debug_is_counted_not_kept(capture):
    attach(capture)
    feed(capture, nav_started("https://a.test/"), log("d", level="debug"), log("i"))
    view = capture.console("ws")
    assert [e["text"] for e in view["entries"]] == ["i"] and view["not_kept"] == {"debug": 1}


def test_successful_static_is_counted_not_kept_and_a_failed_image_is_kept(capture):
    attach(capture)
    feed(capture, nav_started("https://a.test/"),
         req("r1", "https://a.test/a.css", dest="style"), done("r1", 200, mime="text/css"),
         req("r2", "https://a.test/x.png", dest="image"), done("r2", 404, mime="image/png"),
         req("r3", "https://a.test/api", initiator="fetch"), done("r3", 500))
    view = capture.network("ws")
    assert [r["url"] for r in view["requests"]] == ["https://a.test/api", "https://a.test/x.png"]
    assert view["not_kept"] == {"static": 1}
    assert capture.counts("ws") == {"console": 0, "network": 3, "errors": 0, "failed_requests": 1}


def test_requests_merge_by_id_and_redirect(capture):
    attach(capture)
    feed(capture, nav_started("https://a.test/"),
         req("r1", "https://a.test/login", dest="document", nav="n1", ts=2000.0),
         done("r1", 302, mime="text/html", ts=2010.0),
         req("r1", "https://a.test/home", dest="document", nav="n1", redirect=1, ts=2011.0),
         done("r1", 200, mime="text/html", redirect=1, ts=2050.0))
    rows = capture.network("ws")["requests"]
    assert [(r["id"], r["status"], r["took_ms"]) for r in rows] == [("1.2", 200, 39), ("1.1", 302, 10)]


def test_size_is_body_size_never_content_size(capture):
    attach(capture)
    _, completed = done("r1", 200, body_size=None)
    completed["response"].update({"bytesReceived": 777, "content": {"size": 99999}})
    feed(capture, nav_started("https://a.test/"), req("r1", "https://a.test/api"),
         ("network.responseCompleted", completed))
    assert capture.network("ws")["requests"][0]["size"] == 777


def test_only_the_newest_navigation_keeps_its_query(capture):
    attach(capture)
    feed(capture, nav_started("https://a.test/cb?code=abc"),
         req("r1", "https://a.test/api?token=t1"), done("r1", 200),
         nav_started("https://a.test/home?tab=2"),
         req("r2", "https://a.test/api?page=2"), done("r2", 200))
    rows = capture.network("ws", pages=2)["requests"]
    assert [r["url"] for r in rows] == ["https://a.test/api?page=2", "https://a.test/api"]
    assert [p["url"] for p in capture.console("ws", pages=2)["pages"]] == [
        "https://a.test/home?tab=2", "https://a.test/cb"]


def test_the_spare_tab_is_never_kept_counted_or_tailed(capture):
    attach(capture)
    with capture.hidden("s1", "spare-1"):
        feed(capture, req("r9", "https://b.test/__selenium-flow-spare__", ctx="spare-1"), done("r9", 200),
             log("from spare", ctx="spare-1"))
    assert capture.counts("ws") == {"console": 0, "network": 0, "errors": 0, "failed_requests": 0}


def test_another_tab_marks_no_navigation(capture):
    attach(capture, top="top-1")
    feed(capture, nav_started("https://a.test/"), nav_started("https://popup.test/", ctx="tab-2"))
    assert [p["url"] for p in capture.console("ws", pages=5)["pages"]] == ["https://a.test/"]


def test_chrome_replay_lands_only_the_unmatched_suffix(capture):
    attach(capture, browser="chrome")
    first = [log(f"m{i}", ts=1000.0 + i) for i in range(3)]
    feed(capture, nav_started("https://a.test/"), *first)
    capture.attached("ws", "s1", "chrome", "top-1", again=True)
    feed(capture, *first[1:], log("m3", ts=1003.0))      # Chrome's burst: m1, m2, then the new m3
    capture.replayed("ws", "s1")
    assert [e["text"] for e in capture.console("ws")["entries"]] == ["m3", "m2", "m1", "m0"]


def test_firefox_holds_nothing_on_reconnect(capture):
    attach(capture, browser="firefox")
    tick = log("tick", ts=1000.0)
    feed(capture, nav_started("https://a.test/"), tick)
    capture.attached("ws", "s1", "firefox", "top-1", again=True)
    feed(capture, tick)                                   # a real repeat, not a replay
    capture.replayed("ws", "s1")
    assert [e["text"] for e in capture.console("ws")["entries"]] == ["tick", "tick"]


def test_the_ttl_slides_on_every_entry_and_expires_after_the_last(capture, clock):
    attach(capture)
    feed(capture, nav_started("https://a.test/"), log("a"))
    clock.now += DEFAULT_TTL - 1
    feed(capture, log("b"))
    clock.now += DEFAULT_TTL - 1
    assert capture.console("ws") is not None
    clock.now += 2
    assert capture.console("ws") is None


def test_mark_and_caused_count_this_window_only(capture):
    attach(capture)
    feed(capture, nav_started("https://a.test/"), log("old", level="error"))
    since = capture.mark("ws")
    feed(capture, log("boom", type="javascript", level="error"), log("w", level="warn"),
         req("r1", "https://a.test/api"), done("r1", 503),
         req("r2", "https://a.test/i.png", dest="image"), fail("r2"))
    assert capture.caused("ws", since) == {"errors": 1, "failed_requests": 1}


def test_the_index_finds_a_request_by_session_and_id(capture):
    attach(capture)
    feed(capture, nav_started("https://a.test/"), req("r1", "https://a.test/api"), done("r1", 200))
    assert capture.request_for("s1", "1.1") == Located("s1", "r1", "application/json", 10)
    assert capture.request_for("s1", "9.9") is None and capture.request_for("other", "1.1") is None


async def test_a_tail_gets_every_row_masked_and_reports_its_drops(capture):
    attach(capture)
    capture.secret("ws", ["hunter2"])
    tail = capture.tail("ws", "console")
    feed(capture, log("pw hunter2"), *[log(f"m{i}") for i in range(TAIL_QUEUE + 5)])
    assert await tail.next() == {"dropped": 6}
    first = await tail.next()
    assert first["text"] == f"pw {MASK}" and first["kept"] is True
    tail.close()


def test_discard_forgets_the_workspace(capture):
    attach(capture)
    feed(capture, nav_started("https://a.test/"), log("a"))
    capture.discard("ws")
    assert capture.console("ws") is None and capture.mark("ws") is None

(attach(capture, browser="chrome", top="top-1") calls capture.attached("ws", "s1", browser, top, again=False).)

tests/test_capture_views.py: level/filter/pages parsing ("3" → 3; 0, 6, "x" refused with the sentence; "Error" → "error"), the error level including type == "javascript", the failed filter including a 404, xhr including fetch, pages=2 spanning two marks, and a request detail carrying its headers and uri.

• Step 3: Run them to see them fail. python3 -m pytest tests/test_capture_*.py -q → ModuleNotFoundError.

• Step 4: Implement the five modules to the rules. overlap is the one algorithm the tests do not determine on their own:

def overlap(seen: Sequence, replay: Sequence) -> int:
    """How many of ``replay``'s first entries are the last of ``seen``.

    Chrome replays the last 100 entries per browsing context to a new
    subscriber, in order (measured 2026-10-10): what it replays is the tail of
    what was already seen, then whatever arrived while nobody listened. Matched
    as a sequence, never as a set: identical tuples are real entries (ruling 9).
    """
    for j in range(min(len(seen), len(replay)), 0, -1):
        if list(replay[:j]) == list(seen[-j:]):
            return j
    return 0

• Step 5: Boundaries as listed under Files; python3 -m pytest tests/test_boundaries.py -q.

• Step 6: Verify. python3 -m pytest tests/test_capture_*.py tests/test_boundaries.py -q, then the whole suite; $D/bin/ruff check ..

• Step 7: Commit. "capture/: kept buffers, masking, merging, alignment, a TTL and the live tail (#64)".

Files:

• Modify: kubed/selenium_flow/workspace/settings.py, config.py, core/actions.py (open_session), workspace/workspaces.py (__init__, open_browser, resolve, describe, end_browser, the new captured), server.py, mcp/tools.py (open_session's parameter and docstring), core/capabilities.py (open_session's response: capture), mcp/resources.py (DESCRIPTION), spec/schemas.py (RESPONSES["current_workspace"]), tests/golden/openapi.json, tests/golden/tools-on.json, tests/golden/tools-off.json, the config goldens/tests that enumerate sections
• Create: tests/test_capture_open.py

Interfaces:

• Consumes: Monitor.attach, Monitor.capturing, Monitor.connection, Monitor.loop (Task 3); Capture (Task 5); Channels (Task 2).
• Produces:

# workspace/settings.py
SETTINGS["capture"] = (None, None, _as_flag)     # explicit only, like record
# config.py
class CaptureSettings(Section):
    ttl: int = Field(10_800, ge=60, description="Seconds captured console and network entries are kept after the last one arrived.")
Settings.capture: CaptureSettings   # description "Console and network capture, for open_session(capture=true)."
# core/actions.py
Actions.open_session(..., record=None, capture=None, video_name=None, site_data=None, *,
                     on_created=None, on_ready: Callable[[str], None] | None = None) -> dict
# workspace/workspaces.py
CAPTURE_OFF = "Capture is not set up on this server."
Workspaces.__init__(..., recordings=None, monitor=None, capture=None, channels=None)   # kept as self.capture, self.channels
Workspaces.captured(name: str, kind: Literal["console", "network", "request"], **query) -> dict

Behaviour (spec §5, programme R8, R9, R16, ruling 6, plan ruling 3):

• Actions.open_session: after the site-data restore and before driver.get(url), calls on_ready(session_id) when given; if it raises, the browser is quit (self.grid.quit, failure suppressed and logged without the id) and the exception propagates. capture true adds "capture": True to applied.

• Workspaces.open_browser: capture is excluded from the inherited settings exactly as record is; resolved capture with no monitor or no capture configured → ValueError(CAPTURE_OFF); with it, on_ready=lambda sid: self.monitor.attach(name, sid, "capture", browser=kind); without it, self.capture.discard(name) (programme R9: an explicit open without capture discards the buffers). The result's capture is bool(resolved.get("capture")), beside recording.

• Workspaces.resolve's silent reopen replays capture (it is in the stored settings) with the same on_ready; with the monitor or capture gone since, capture is dropped from the replay as record is.

• Every call into the monitor from Workspaces is guarded: opened, ended and watch failures are logged by type and never fail the caller — end_browser included (E2 minor 6). attach is the exception: its failure is the open's (a 503).

• describe adds capture (bool(settings.capture)), capturing (live and monitor.capturing(record.session_id)) and, when capture is on, captured = capture.counts(name) (or zeros) — plan ruling 12.

• captured(name, kind, **query): the setting off → dict(capture.OFF); else Capture.console|network|request(...) with "capturing" added; an empty buffer reads as an empty view with capture: true; kind == "request" with an unknown id → ValueError(f"no captured request {id!r}: workspace://network lists them").

• server.py: self.capture = Capture(ttl=settings.capture.ttl); Monitor(..., events_for={"capture": capture.EVENTS}, collectors_for={"capture": capture.COLLECTOR}, sink=self.capture); Workspaces(..., capture=self.capture, channels=self.actions.channels); self.actions.hide = self.capture.hidden; self.actions.capture = self.capture (Task 8 reads it).

• The MCP parameter (prompt, in open_session's docstring after the record paragraph):

  capture=true keeps what the page logs and fetches — workspace://console and workspace://network — and a call that caused a page error says so in its result. Off by default and not inherited: ask again for the next browser.

  Signature: capture: bool | None = None after record.

• Step 1: Write the failing tests. tests/test_capture_open.py, in two halves. Actions level — a real Actions on a stub Grid whose open() returns a stub driver recording get(url) into a shared order list, and whose quit(sid) records into quit_ids. Workspaces level — the local-fake pattern of tests/test_workspace_monitor.py (a fake Actions whose open_session(**kwargs) records its kwargs and calls on_ready(sid) when given; a fake Grid with alive), a FakeMonitor recording attach(workspace, sid, reason, **kw) (appending "attach" to order; refuse=True raises BidiUnavailable) and opened/ended/watch, and a real Capture:

def test_on_ready_runs_after_the_restore_and_before_the_first_page(stub_grid):
    actions = Actions(stub_grid)
    actions.open_session(url="https://a.test/", capture=True,
                         on_ready=lambda sid: stub_grid.order.append("attach"))
    assert stub_grid.order == ["attach", "get https://a.test/"]


def test_a_failed_on_ready_quits_the_browser_and_raises(stub_grid):
    def refuse(_sid):
        raise BidiUnavailable("no route")

    with pytest.raises(BidiUnavailable):
        Actions(stub_grid).open_session(url="https://a.test/", on_ready=refuse)
    assert stub_grid.quit_ids == [stub_grid.last_id] and "get https://a.test/" not in stub_grid.order


def test_capture_shows_in_the_applied_settings(stub_grid):
    assert Actions(stub_grid).open_session(capture=True)["settings"]["capture"] is True


def test_capture_is_explicit_only():
    assert settings.from_client({"capture": "true"}, {"x-capture": "true"}) == {}
    assert settings.resolve({"capture": "false"}, previous={"capture": True})["capture"] is False


def test_an_open_with_capture_attaches_and_remembers_it(ws):
    ws.workspaces.open_browser(caller(), url="https://a.test/", capture=True)
    assert ws.monitor.attached == [("bot", "grid0001", "capture")]
    assert ws.workspaces.store.get("bot").settings["capture"] is True


def test_a_failed_attach_fails_the_open_and_binds_nothing(ws):
    ws.monitor.refuse = True
    with pytest.raises(BidiUnavailable):
        ws.workspaces.open_browser(caller(), capture=True)
    record = ws.workspaces.store.get("bot")
    assert record is None or not record.attached


def test_capture_is_never_inherited_but_a_reap_replays_it(ws):
    ws.workspaces.open_browser(caller(), capture=True)
    ws.actions.grid.alive.clear()                               # the Grid reaped it
    ws.workspaces.resolve("bot")
    assert len(ws.monitor.attached) == 2                        # the silent reopen replayed it
    ws.workspaces.open_browser(caller())
    assert len(ws.monitor.attached) == 2                        # an explicit open did not
    assert "capture" not in ws.actions.opened[-1]


def test_an_explicit_open_without_capture_discards_the_buffers(ws):
    ws.capture.attached("bot", "old", "chrome", "top-1", again=False)
    ws.capture.ingest("bot", "old", *log("kept"))
    ws.workspaces.open_browser(caller())
    assert ws.capture.console("bot") is None


def test_capture_without_a_monitor_is_refused_before_a_browser_opens():
    bare = Workspaces(FakeActions())
    with pytest.raises(ValueError, match="Capture is not set up"):
        bare.open_browser(caller(), capture=True)
    assert bare.actions.opened == []


def test_a_raising_monitor_never_fails_end_browser(ws):
    ws.workspaces.open_browser(caller())

    def boom(*_args, **_kwargs):
        raise RuntimeError("monitor broke")

    ws.monitor.ended = boom
    ws.workspaces.end_browser(caller())
    assert not ws.workspaces.store.get("bot").attached


def test_current_reports_capture_capturing_and_counts(ws):
    ws.workspaces.open_browser(caller(), capture=True)
    ws.monitor.held.add("grid0001")
    status = ws.workspaces.describe(caller())
    assert status["capture"] is True and status["capturing"] is True
    assert status["captured"] == {"console": 0, "network": 0, "errors": 0, "failed_requests": 0}


def test_capture_off_reads_as_off_not_an_error(ws):
    ws.workspaces.open_browser(caller())
    assert ws.workspaces.captured("bot", "console") == {
        "capture": False, "how": "open_session(capture=true)"}


def test_the_ttl_is_a_setting_in_its_own_section():
    assert config.load([], {}).settings.capture.ttl == 10_800
    assert config.load([], {"CAPTURE_TTL": "600"}).settings.capture.ttl == 600
    assert config.load(["--capture-ttl", "900"], {}).settings.capture.ttl == 900

(ws bundles workspaces, actions, monitor, capture; FakeMonitor.capturing(sid) is sid in self.held; log is from tests/capture_events.py.)

• Step 2: Run them to see them fail. python3 -m pytest tests/test_capture_open.py -q.

• Step 3: Implement the changes above.

• Step 4: Goldens and schemas. RESPONSES["current_workspace"] gains capture (boolean, "whether this workspace opens its browsers with capture on"), capturing (boolean, "whether the console and network are being captured right now; false while capture is paused"), captured (object of four integers, "since the last navigation"); open_session's response row gains capture. Regenerate openapi.json, tools-on.json, tools-off.json and whatever config golden enumerates sections (GOLDEN_UPDATE=1), and read each diff: only the new fields move.

• Step 5: Verify. The whole suite and ruff.

• Step 6: Commit. "open_session(capture=true) attaches before the first page; capture.ttl (#64)" — name the moved goldens in the body.

Files:

• Create: kubed/selenium_flow/workspace/calls.py, tests/test_capture_hint.py
• Modify: workspace/workspaces.py (act, a public resolved), faults.py (PAGE_NOTE, message), secrets.py (perform_write), flows/engine.py (Hooks, Callbacks, _drive), flows/run.py (run takes bound= and step_done= and puts them in its engine.Callbacks), flows/api.py (run_one passes them through; run_for wires them), routes.py, mcp/tools.py (act, _as_image), tests/test_flowrun.py (the new hooks), tests/test_monitor_events.py (call.finished is emitted now)

Interfaces:

• Consumes: Capture.mark, Capture.caused, Capture.secret (Task 5); Monitor.touch, Monitor.connection, Monitor.loop, Monitor.bus (Task 3); bidi.Channel (Task 2); monitor.events.CallFinished (E2).
• Produces:

# workspace/calls.py
class Begun(NamedTuple):
    workspace: str; session_id: str; started: float; mark: Mark | None

class Calls:
    def __init__(self, *, monitor=None, capture=None, clock: Callable[[], float] = time.time): ...
    def wanted(self, workspace: str) -> bool                       # capture holds a buffer for it
    def begin(self, workspace: str, session_id: str) -> Begun
    def end(self, begun: Begun, *, tool: str, surface: Literal["mcp", "http", "flow"],
            browser: str, error: BaseException | None = None) -> dict | None   # the hint, or None
    def bound(self, workspace: str, forms: Iterable[str]) -> None

def hint(caused: dict) -> dict | None
def total(hints: Iterable[dict | None]) -> dict | None
# faults.py
PAGE_NOTE = "This call caused "
# workspace/workspaces.py
Workspaces.calls: Calls
Workspaces.act(caller, call, *, reshapes=False, tool: str = "", surface: str = "mcp") -> dict
Workspaces.resolved(name: str) -> tuple[str, str]      # (session_id, browser kind); run_for and act use it
# secrets.py
perform_write(catalogue, actions, workspaces, name, kwargs, *, surface: str = "mcp") -> dict
# flows/engine.py — Hooks gains
def bound(self, forms: set[str]) -> None: ...                       # right after a step's forms are known, before it acts
def step_done(self, tool: str, error: BaseException | None) -> None: ...   # after every step that acted, inside its lock

Behaviour (spec §6, ruling 12, E2 r8, plan rulings 4–8):

• hint(caused) → None when both counts are 0, else {"errors": e, "failed_requests": f, "see": "workspace://console?level=error" if e else "workspace://network?filter=failed"}. total sums the counts of the non-None ones and recomputes see the same way.

• Calls.end, never raising: monitor.touch(session_id); when begun.mark is set and monitor.connection(session_id) is open, Channel(conn, monitor.loop).flush() (any failure logged at debug and ignored); capture.caused(...) → hint; publishes CallFinished(workspace, session_id, browser, tool, surface, outcome, started, duration) through monitor.bus with outcome "ok" with no error, else "4xx"/"5xx" from errors.status_for(error); with an error and a hint, error.add_note(f"{PAGE_NOTE}{e} page error(s) and {f} failed request(s): read {see}."). Returns the hint.

• faults.message appends every note of exc.__notes__ that starts with PAGE_NOTE, after a space; other notes never reach a caller.

• Workspaces.act: resolved, kind = self.resolved(name) (resolve's body, also returning the browser kind it read off the record, so no second store read; resolve(name) returns resolved(name)[0]); when self.calls.wanted(name), the whole of begin → call → end runs inside locks.driving(resolved) (plan ruling 8), else without it; settle as today; a non-None hint goes on the result as result["page"]. A raising call goes through end(..., error=exc) and re-raises.

• perform_write: resolved, kind = workspaces.resolved(name); inside its existing locks.driving(resolved): begin first; workspaces.calls.bound(name, hidden) before actions.write; end(..., tool="write", surface=surface); the hint goes on shown["page"].

• Flows: _drive calls hooks.bound(binding.forms_of(kwargs, guarded)) when non-empty, before before_step; hooks.step_done(tool, None) after after_step, and hooks.step_done(tool, exc) in the except before landed. run_for resolves with resolved, kind = workspaces.resolved(workspace) and wires before_step → its caller's own before_step (MCP progress) first, then begun = calls.begin(workspace, resolved), step_done → hints.append(calls.end(begun, tool=tool, surface="flow", browser=kind, error=error)), bound → calls.bound(workspace, forms); the report gets report["page"] = total(hints) when that is not None.

• Surfaces: tools.act passes tool=row.name, surface="mcp"; routes._add passes tool=row.name, surface="http"; both perform_write calls pass their surface. _as_image keeps page beside file, file_error and site_data.

• Step 1: Write the failing tests. tests/test_capture_hint.py, with a real Capture, a FakeMonitor (records touch, published, serves connection()/loop as None unless a test sets them) and FakeGrid:

def test_a_call_that_caused_errors_says_so_on_its_result(captured_ws):
    workspaces, capture, _ = captured_ws

    def click(session_id):
        capture.ingest("ws", session_id, *log("Uncaught TypeError", type="javascript", level="error"))
        capture.ingest("ws", session_id, *req("r1", "https://a.test/api/orders", method="POST"))
        capture.ingest("ws", session_id, *done("r1", 500))
        return {"url": "https://a.test/", "title": "A"}

    result = workspaces.act(Caller("ws", "query"), click, tool="interact", surface="mcp")
    assert result["page"] == {"errors": 1, "failed_requests": 1, "see": "workspace://console?level=error"}


def test_no_hint_when_nothing_went_wrong_or_capture_is_off(captured_ws, plain_ws):
    workspaces, capture, _ = captured_ws
    ok = workspaces.act(Caller("ws", "query"), lambda s: {"url": "u"}, tool="extract")
    assert "page" not in ok
    assert "page" not in plain_ws.act(Caller("ws", "query"), lambda s: {"url": "u"}, tool="extract")


def test_warnings_and_failed_images_stay_silent(captured_ws):
    workspaces, capture, _ = captured_ws

    def noisy(sid):
        capture.ingest("ws", sid, *log("careful", level="warn"))
        capture.ingest("ws", sid, *req("r2", "https://a.test/x.png", dest="image"))
        capture.ingest("ws", sid, *fail("r2"))
        return {"url": "u"}

    assert "page" not in workspaces.act(Caller("ws", "query"), noisy, tool="interact")


def test_a_failed_call_says_what_the_page_reported(captured_ws):
    workspaces, capture, _ = captured_ws

    def click_then_time_out(sid):
        capture.ingest("ws", sid, *req("r1", "https://a.test/api", initiator="fetch"))
        capture.ingest("ws", sid, *done("r1", 500))
        raise TimeoutException("waited 30s for //div[@id='done']")

    with pytest.raises(TimeoutException) as raised:
        workspaces.act(Caller("ws", "query"), click_then_time_out, tool="interact")
    assert faults.message(raised.value).endswith(
        "This call caused 0 page error(s) and 1 failed request(s): read workspace://network?filter=failed.")


def test_each_call_counts_only_its_own_window(captured_ws):
    workspaces, capture, _ = captured_ws
    entered, release = threading.Event(), threading.Event()

    def slow(sid):
        entered.set(); release.wait(2)
        capture.ingest("ws", sid, *log("mine", level="error"))
        return {"url": "u"}

    results = {}
    first = threading.Thread(target=lambda: results.setdefault(
        "slow", workspaces.act(Caller("ws", "query"), slow, tool="interact")))
    first.start(); entered.wait(2)
    second = threading.Thread(target=lambda: results.setdefault(
        "quick", workspaces.act(Caller("ws", "query"), lambda s: {"url": "u"}, tool="extract")))
    second.start(); release.set(); first.join(); second.join()
    assert results["slow"]["page"]["errors"] == 1 and "page" not in results["quick"]


def test_every_call_touches_and_publishes_call_finished(captured_ws):
    workspaces, _, monitor = captured_ws
    with pytest.raises(ValueError):
        workspaces.act(Caller("ws", "query"), lambda s: (_ for _ in ()).throw(ValueError("bad")),
                       tool="write", surface="http")
    [event] = monitor.published
    assert (event.tool, event.surface, event.outcome, event.browser) == ("write", "http", "4xx", "chrome")
    assert monitor.touched == [event.session_id]


def test_a_bound_write_registers_its_value_before_typing(captured_ws, catalogue):
    workspaces, capture, _ = captured_ws
    order = []
    workspaces.calls.bound = lambda ws, forms: order.append(("bound", sorted(forms)[0]))
    actions = workspaces.actions
    actions.write = lambda sid, text, **kw: order.append(("typed", text)) or {"url": "u", "value": None}
    perform_write(catalogue, actions, workspaces, "ws",
                  {"secret": {"name": "demo", "key": "password"}, "selector": {"css": "#p"}})
    assert [step for step, _ in order] == ["bound", "typed"]


def test_a_flow_run_sums_its_steps_hints(flow_run_with_capture):
    report = flow_run_with_capture(steps=[("interact", 1, 0), ("interact", 0, 2)])  # (tool, errors, failed) per step
    assert report["page"] == {"errors": 1, "failed_requests": 2, "see": "workspace://console?level=error"}
    assert [e.surface for e in flow_run_with_capture.monitor.published] == ["flow", "flow"]

(captured_ws is a Workspaces built on the local fakes of tests/test_workspace_monitor.py, a real Capture attached for ws on its session, and a FakeMonitor recording touch and published (its bus.publish), with connection() returning None; plain_ws is the same without capture; catalogue holds demo/password usable anywhere, and the fake actions answer page() with {"url": "https://a.test/", "title": "A"}; flow_run_with_capture saves a two-step flow whose fake actions ingest the given errors and failed fetches, runs it through flows.api.run_for, and exposes the monitor.)

tests/test_flowrun.py gains: bound is called with the step's forms before the action; step_done is called once per step that acted, with the exception for a failing step, and never for a step refused before acting.

• Step 2: Run them to see them fail. python3 -m pytest tests/test_capture_hint.py tests/test_flowrun.py -q.

• Step 3: Implement workspace/calls.py, the act/resolved change, faults.message's note, perform_write, the engine hooks, run_for, the two surfaces and _as_image. Workspaces.__init__ builds self.calls = Calls(monitor=monitor, capture=capture).

• Step 4: Verify. python3 -m pytest tests/test_capture_hint.py tests/test_flowrun.py tests/test_surfaces.py tests/test_secrets.py tests/test_binding.py tests/test_monitor_events.py -q, then the whole suite (the flow-report and errors goldens must not move: no hint without capture).

• Step 5: Commit. "Every call touches the monitor, says call.finished, and hints at a page error it caused (#64)".

Files:

• Modify: kubed/selenium_flow/mcp/resources.py, core/capabilities.py (the visibility column, the network_body row), core/actions.py (network_body), mcp/tools.py (publish passes app=; the declaration), flows/document.py (NOT_STEPS), tests/test_surfaces.py (EXPECTED, the read-only set), tests/test_resources.py, goldens (openapi.json, tools-on.json, tools-off.json)
• Create: tests/test_network_body.py, tests/test_capture_resources.py

Interfaces:

• Consumes: Workspaces.captured (Task 6); Capture.request_for, Located (Task 5); Actions.channels (Task 4); faults.NotFound; bidi.BidiError.
• Produces:

# core/capabilities.py
@dataclass(frozen=True)
class Capability:  # ... existing columns ...
    visibility: tuple[str, ...] = ()       # MCP Apps audiences; () = the default
Capability(name="network_body", method="network_body", route="network-body", http_method="POST",
           annotations=reads("Fetch a response body"), response=BODY_RESPONSE,
           visibility=("model", "app"))
# core/actions.py
BODY_GONE = "the browser no longer holds this body: it keeps them while it lives, and drops the oldest past about 200 MB"
Actions.capture: Capture | None
Actions.network_body(self, session_id: str, id: str) -> dict
# mcp/resources.py
CONSOLE_DESCRIPTION, NETWORK_DESCRIPTION, REQUEST_DESCRIPTION: str

The resources (spec §7, plan ruling 12), all side-effect free (Workspaces.captured, never resolve):

Measured on FastMCP 4.1 (2026-10-10): the static URI is listed and answers a bare read; the query template answers workspace://console?level=error&pages=3; workspace://network/3.7 reaches the {id} template.

Descriptions (prompt, for a model):

• console: "What the page logged since the last navigation, newest first: errors, warnings, info, and uncaught exceptions with where they were thrown. ?level=error or ?level=warn narrows it; ?pages=2 (up to 5) reaches back over earlier pages. debug entries are counted, not kept. Present only for a workspace opened with open_session(capture=true); otherwise it says so. Reading this never opens a browser."
• network: "Requests the page made since the last navigation, newest first, each with an id: method, URL, status, type, size and time. ?filter=failed or ?filter=xhr narrows it; ?pages=2 (up to 5) reaches back. Successful stylesheets, images, fonts, scripts and media are counted, not kept. workspace://network/{id} shows one with its headers; network_body(id) fetches its response body while the browser lives. Credentials are shown as •••."
• request: "One captured request in full: its headers (authorization, cookies and any secret you typed shown as •••), status, timing and size. network_body(id) fetches its response body."
• workspace://current, one sentence added: "capture says whether this workspace's browsers capture their console and network (open_session(capture=true)); capturing is whether that is happening right now — false while it is paused, until the next call; captured counts what was kept since the last navigation."

network_body (spec §7, ruling 11, plan rulings 1, 16, 19):

• located = self.capture.request_for(session_id, id) (self.capture absent → None); None → ValueError(f"no captured request {id!r}: workspace://network lists them") (400).
• located.session_id != session_id → NotFound(f"the browser that made request {id} has ended, and its body went with it") (404) — the act's resolve may have reopened a reaped browser; the body never comes from the wrong one.
• with self.channels.open_channel(session_id) as ch: ch.call("network.getData", {"dataType": "response", "request": located.request}); a BidiError whose error == "no such network data" → NotFound(BODY_GONE).
• {"bytes": {"type": "string", "value": v}} → text: the UTF-8 bytes cut to BODY_CAP (100 000) on a character boundary (v.encode()[:BODY_CAP].decode("utf-8", "ignore")), truncated when cut; size is len(v.encode()), before the cut. {"type": "base64"} → no text; size is the decoded length. (size is what the browser handed back, not the row's transfer size.)
• Result: {"id": id, "uri": f"workspace://network/{id}", "mime": located.mime, "size": n, "binary": bool, "text": str | None, "truncated": bool}; BODY_RESPONSE in the row says exactly this.
• MCP declaration:

@action("network_body")
def network_body(id: str) -> dict:
    """Fetch the response body of one request the page made, by its id from
    workspace://network (an id looks like 3.7). Text comes back up to 100 kB,
    with truncated=true beyond that; for an image or other binary you get its
    type and size, not its bytes. The browser keeps bodies only while it lives,
    so read one soon after the request; it needs open_session(capture=true)."""

• tools.publish passes app=AppConfig(visibility=list(row.visibility)) when row.visibility is non-empty — measured: FastMCP 4.1 emits _meta.ui.visibility with no resourceUri, so it is callable by the app and listed to every client. It is not in HideMirrors' app tools.

• flows/document.NOT_STEPS["network_body"] = "A request id belongs to one page load, which a saved flow cannot know when it is written; read workspace://network and fetch a body while working one out".

• tests/test_surfaces.py: EXPECTED gains "network_body"; the read-only set becomes {"extract", "outline", "network_body"} with one line in the docstring (it reads the browser and writes nothing).

• Step 1: Write the failing tests. tests/test_network_body.py (Actions on FakeGrid, a real Capture, FakeChannel answering network.getData from a bodies map):

def body_of(value, kind="string"):
    return {"bytes": {"type": kind, "value": value}}


def test_text_comes_back_whole_under_the_cap(actions, fed):
    fed.channel.bodies["r1"] = body_of('{"error":"insert failed"}')
    assert actions.network_body("s1", "1.1") == {
        "id": "1.1", "uri": "workspace://network/1.1", "mime": "application/json",
        "size": 25, "binary": False, "text": '{"error":"insert failed"}', "truncated": False}


def test_text_over_100_kb_is_cut_on_a_character_and_marked(actions, fed):
    fed.channel.bodies["r1"] = body_of("é" * 60_000)            # 120 000 bytes
    got = actions.network_body("s1", "1.1")
    assert got["truncated"] is True and len(got["text"].encode()) <= BODY_CAP
    assert got["text"] == "é" * 50_000


def test_binary_returns_its_type_and_size_only(actions, fed):
    fed.channel.bodies["r1"] = body_of(base64.b64encode(b"\x89PNG" * 10).decode(), "base64")
    got = actions.network_body("s1", "1.1")
    assert got["binary"] is True and got["text"] is None and got["size"] == 40


def test_an_unknown_id_is_400(actions, fed):
    with pytest.raises(ValueError, match="workspace://network lists them"):
        actions.network_body("s1", "9.9")
    assert errors.status_for(ValueError()) == 400


def test_a_body_from_an_ended_browser_is_404(actions, fed):
    fed.capture.attached("ws", "s2", "chrome", "top-1", again=False)     # the reopened browser
    with pytest.raises(NotFound, match="request 1.1 has ended") as raised:
        actions.network_body("s2", "1.1")
    assert errors.status_for(raised.value) == 404
    assert fed.channel.log == []                                         # nothing asked of any browser


def test_a_body_the_browser_dropped_is_404(actions, fed):
    with pytest.raises(NotFound, match="no longer holds"):
        actions.network_body("s1", "1.1")                                # FakeChannel: no such network data

tests/test_capture_resources.py (through a real Client(server.mcp) with a named caller, as tests/test_resources.py does): the five forms are served (two listed, three templates); workspace://console?level=error&pages=2 passes both through; ?pages=9 is an error naming 1 to 5; with capture off each answers OFF; workspace://network/1.1 with an unknown id is an error naming workspace://network; reading any of them calls nothing on the Grid (FakeGrid.calls == []); workspace://current carries capture, capturing, captured.

In tests/test_surfaces.py, a new case:

async def test_network_body_is_callable_by_the_app_and_listed_to_the_model(server):
    tool = {t.name: t for t in await server.mcp.list_tools()}["network_body"]
    assert tool.meta["ui"] == {"visibility": ["model", "app"]}
    assert tool.annotations.read_only_hint is True and tool.annotations.destructive_hint is False

• Step 2: Run them to see them fail. python3 -m pytest tests/test_network_body.py tests/test_capture_resources.py tests/test_surfaces.py -q.

• Step 3: Implement the row, the action, the declaration, the resources, NOT_STEPS.

• Step 4: Goldens. Regenerate openapi.json (a new POST /browser/network-body with NetworkBodyRequest/NetworkBodyResponse and the standard 400/401/404/413/500/503), tools-on.json, tools-off.json; read the diffs. python3 scripts/generate_openapi.py and lint it as test.yml does (openapi.yaml stays uncommitted).

• Step 5: Verify. The whole suite and ruff.

• Step 6: Commit. "workspace://console, workspace://network and network_body (#64)" — goldens named.

Files:

• Modify: kubed/selenium_flow/mcp/show.py (VIEWS, NOUNS), tests/test_show_inventory.py (FORMS, concrete), tests/test_show.py, ui/src/App.svelte (VIEWS, the oncall prop), ui/src/lib/views/ContextView.svelte, ui/src/lib/types.ts, ui/src/lib/format.ts
• Create: ui/src/lib/capture.ts, ui/src/lib/views/ConsoleView.svelte, NetworkView.svelte, RequestView.svelte and a .test.ts beside each; ui/src/lib/capture.test.ts

Read first (read-only, never edited — programme R15): Penpot file Admin UI, page App · show, boards context, console, network, request, through the penpot MCP if it is connected (penpot.currentFile.pages → find the page → walk the board's text). And the kb Svelte skills: skill://svelte/_index.md and what it routes to for Svelte 5 runes and component tests. Do not edit Penpot.

Interfaces:

• Consumes: the resource shapes (Task 5 views, Task 8), network_body's result (Task 8).
• Produces:

# mcp/show.py — rows added, first match wins, query strings allowed
View("workspace://console", re.compile(r"workspace://console(\?.*)?"), "console"),
View("workspace://network/{id}", re.compile(r"workspace://network/[^/?]+"), "request"),
View("workspace://network", re.compile(r"workspace://network(\?.*)?"), "network"),
NOUNS |= {"console": "entry", "network": "request"}

// ui/src/lib/types.ts
export interface ConsoleEntry { seq: number; at: number; level: string; type: string; text: string;
  source: { url: string; line: number } | null; context: string; nav: number; kept?: boolean }
export interface RequestRow { id: string; seq: number; at: number; method: string; url: string;
  status: number | null; status_text?: string; type: string; mime?: string; size: number | null;
  took_ms: number | null; failed: boolean; error: string | null; context: string; nav: number;
  navigation: boolean; kept?: boolean }
export interface RequestDetail extends RequestRow { request_headers: Header[]; response_headers: Header[]; uri: string }
export interface Header { name: string; value: string }
export interface PageMark { nav: number; url: string; at: number }
export type CaptureOff = { capture: false; how: string }
export interface ConsoleData { capture: true; capturing?: boolean; pages: PageMark[]; entries: ConsoleEntry[]; count: number; total: number; not_kept: { debug: number } }
export interface NetworkData { capture: true; capturing?: boolean; pages: PageMark[]; requests: RequestRow[]; count: number; total: number; not_kept: { static: number } }
export interface Captured { console: number; network: number; errors: number; failed_requests: number }
export interface BodyResult { id: string; uri: string; mime: string; size: number | null; binary: boolean; text: string | null; truncated: boolean }
// ContextData gains: capture?: boolean; capturing?: boolean; captured?: Captured
// ui/src/lib/capture.ts — shared by the app views and the admin panes (Task 11)
export const kb: (bytes: number | null) => string          // 412 -> "0.4 kB", 8300 -> "8.3 kB", null -> "—"
export const clock: (at: number) => string                 // epoch s -> "12:01:43" (local, 24 h)
export const statusText: (r: RequestRow) => string         // failed -> "failed", else String(status ?? "")
export const isError: (e: ConsoleEntry) => boolean         // level error or type javascript
export const isFailed: (r: RequestRow) => boolean          // failed or status >= 400
// App.svelte: Props gains  oncall?: (name: string, args: Record<string, unknown>) => Promise<unknown>

What each view draws (boards App · show):

• ContextView: beside live/idle, ● REC (settings.record && live), ● capture (capture && capturing) and capture paused (capture && !capturing), with the classes the admin uses (pill rec, pill capture, pill capture-paused); with capture, chips Console with "{errors} errors" and Network with "{failed_requests} failed" that call onshow('workspace://console') / onshow('workspace://network'), or are inert spans titled "Open isn't available in this client" without onshow (the Site data chip's pattern).

• ConsoleView (console): title Console, the URI, chips Errors n / Warnings n (counts from the entries), rows newest first with level, text and clock(at); at most 4 rows, then "+n more · since the last navigation"; CaptureOff → "Capture is off for this workspace. Open the browser with open_session(capture=true)."

• NetworkView (network): title Network, the URI, chips Failed n / XHR · fetch n, rows with method, statusText, host+path (no scheme), took_ms ms; each row calls onshow('workspace://network/' + id); footer "{static} successful static requests not shown · since the last navigation" when not_kept.static > 0.

• RequestView (request): ← Network (onshow('workspace://network')), METHOD /path, the URI, the status; GENERAL (url, type as "{type} · {took} ms · {kb}"), REQUEST (request headers), RESPONSE (response headers); Fetch body calls oncall('network_body', { id }) and draws the text (a <pre>), "truncated at 100 kB" when truncated, "{mime} · {kb(size)} · not shown" when binary, the error text when the call fails; under the button "read from the browser while it lives". Without oncall the button is absent. No request-body section, no during row (plan ruling 9).

• App.svelte: console: ConsoleView, network: NetworkView, request: RequestView in VIEWS; oncall is (name, args) => host!.callServerTool({ name, arguments: args }) returning structuredContent, set only when getHostCapabilities()?.serverTools (as onshow is), and passed to every view.

• Step 1: Python tests. In tests/test_show_inventory.py, FORMS gains "workspace://console", "workspace://console{?level,pages}", "workspace://network", "workspace://network{?filter,pages}", "workspace://network/{id}", and concrete drops RFC 6570 query blocks before filling path ones:

def concrete(uri: str) -> str:
    """A template made into one URI it would answer: a query block is optional."""
    return re.sub(r"\{[^}]+\}", "sample", re.sub(r"\{\?[^}]+\}", "", uri))

In tests/test_show.py:

@pytest.mark.parametrize("uri, component", [
    ("workspace://console", "console"),
    ("workspace://console?level=error&pages=2", "console"),
    ("workspace://network", "network"),
    ("workspace://network?filter=failed", "network"),
    ("workspace://network/3.7", "request"),
])
def test_the_capture_uris_have_views(uri, component):
    assert show.view_for(uri) == component


async def test_capture_off_still_draws(named_server):
    result = await call_show(named_server, "workspace://console")
    assert result.structured_content["data"] == {"capture": False, "how": "open_session(capture=true)"}