diff --git a/docs/engineering/codex_dgx_worker.md b/docs/engineering/codex_dgx_worker.md
index 3e6a32def..914429b84 100644
--- a/docs/engineering/codex_dgx_worker.md
+++ b/docs/engineering/codex_dgx_worker.md
@@ -398,3 +398,86 @@ processes/results, report replay, reassignment, invitation lookup, and Git
 preparation failure. Use the Jira decision surface for current installation,
 runtime receipts and remaining publication work; a repository merge alone does
 not prove that a scheduler or public runtime was activated.
+
+## Task-linked coding activity and archives (JVNAUTOSCI-2751)
+
+The reviewed controller can register a separate run identity before execution,
+then publish complete JSONL records in bounded batches during its existing wait
+loop. The task inspector's **Coding run activity** button opens private run
+history, paged records, capture freshness/error state and the final download.
+Refresh is explicit. Historical content is inert text, never executable HTML or
+instructions to another model. Changing the browser actor or organisation closes
+the view.
+
+`task_run_archive_service` owns operational run records in `task_run_archives`
+and immutable content-addressed objects through the existing canonical blob
+store. This is a run primitive, not another task writer, workflow, scheduler or
+telemetry platform. It does not replace the current work product or existing
+task evidence. Final ZIP attachments link through the private run download;
+the ordinary organisation-visible attachment upload is deliberately not used
+for transcript bytes. A reassigned, cancelled or unavailable task receives no
+new attachment, but the original private run record retains its task backlink.
+
+The controller's trusted actor and organisation determine the producer. The
+original worker/requester audience is intersected with source-conversation
+access, then rechecked on reads, including after invitation revocation. Task
+visibility alone does not grant archive access. The HTTP routes are read-only:
+
+- `GET /api/tasks/<task>/runs?offset=0` lists up to 50 runs;
+- `GET /api/tasks/<task>/runs/<attempt>?cursor=0` reads one activity batch;
+- `GET /api/tasks/<task>/runs/<attempt>/download` returns the verified ZIP.
+
+The source cursor counts original bytes, independently of redacted/exported
+bytes. Retry identities include the run, start/end cursor and source/export
+hashes. A partial final line waits until complete while the process is running;
+an interrupted terminal tail is retained as an explicitly malformed record.
+Finalisation reconciles the complete available event stream, source segment
+hashes, exported file sizes/hashes, event count, thread identity, outcome,
+missing files, redactions and exposed-summary availability. A digest-checked
+archive reference is required before the controller reports normal completion.
+Coding results remain intact when capture fails: `archive_pending` is retried
+on the existing controller's next opportunity, with a separate pending report.
+No coding effects are repeated to repair archive storage.
+
+Only `events.jsonl`, `context.json`, `schema.json`, `result.json`, `exit.json`,
+`stderr.log`, `launch-error.json`, `deployment.json` and `deployment.log` from
+the selected run are eligible. Local originals remain untouched. Credential
+patterns and known controller environment credentials are redacted with
+counts; opaque provider state is omitted. These are **available execution
+records**, not a comprehensive private reasoning transcript. No additional
+model call or summaries setting is introduced. Provider-exposed reasoning items
+are retained when emitted; an empty exec stream does not prove that no private
+reasoning occurred. Source rollouts are not collected because this route has no
+supported bounded rollout export configured; it never searches other sessions
+or authentication files. Legacy backfill containing a transcript without a
+stable source session locator fails explicitly rather than guessing its scope.
+
+The capture interval is 15 seconds while the child runs, with at most four
+roughly 512 KiB batches per live checkpoint. Limits are 16 MiB per record/batch,
+128 MiB per companion file, and 256 MiB per run archive. Exceeding the bounded
+route leaves finalisation pending and preserves local originals; it does not
+silently truncate them. The UI displays at most 16,000 characters per record;
+the downloadable redacted record is not shortened by that display limit.
+`source_complete` is separate from a successfully stored archive: a failed or
+interrupted run can have a durable archive whose manifest identifies missing
+source files. Controller and web runtime must resolve the same private canonical
+blob storage; a local receipt alone does not prove cross-process availability.
+
+For an explicitly selected historical task/run pair, the existing operator-bound
+controller accepts `--backfill-run 'TASK_CONCEPT_ID=ATTEMPT'` (repeat for at most
+20 pairs). It checks the exact context task/worker, shares the existing worker
+lock, does not launch Codex, and keeps a retry receipt in `archive-backfills/`.
+The ordinary next tick retries only those selected pending receipts. Do not
+scan the estate, remove originals, or run backfill against an active worker.
+
+Candidate acceptance is covered by `tests/backend/test_task_run_archives.py`:
+a deterministic executable traverses the actual `launch()` loop, exposes
+activity through the canonical Flask route before exit, then verifies the final
+ZIP through that route as the requester. Other cases cover partial lines,
+large outputs, corruption, redaction, source revocation, reassignment and
+interrupted finalisation. The 1,000-event fixture performs three run-record
+writes (registration, one batch, final reference), independently of event count.
+`tests/browser/taskRunActivity.cjs` checks desktop/mobile rendering, inert source
+text, refresh, scoped download and clearing on scope change using an isolated
+browser fixture. These tests do not prove live scheduler activation, public
+OAuth, actual provider summaries, or production historical backfill.
diff --git a/scripts/codex_von_activity.py b/scripts/codex_von_activity.py
new file mode 100644
index 000000000..7c35618fe
--- /dev/null
+++ b/scripts/codex_von_activity.py
@@ -0,0 +1,378 @@
+"""Bounded, resumable capture for the existing DGX controller.
+
+Never reads authentication files or discovers sessions. Only named companion
+files in one controller-owned run directory are eligible. Local originals stay
+untouched. Opaque reasoning is removed, never interpreted or decrypted.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import os
+import re
+import zipfile
+from collections import Counter
+from pathlib import Path
+
+FILES = (
+    "context.json",
+    "schema.json",
+    "result.json",
+    "exit.json",
+    "stderr.log",
+    "launch-error.json",
+    "deployment.json",
+    "deployment.log",
+)
+REQUIRED_FILES = {"context.json", "events.jsonl", "exit.json", "result.json"}
+BATCH_TARGET = 512 * 1024
+MAX_SOURCE_FILE = 128 * 1024 * 1024
+
+
+def service():
+    from src.backend.services import task_run_archive_service
+
+    return task_run_archive_service
+
+
+def redact(value, counts):
+    from src.backend.services.turn_failure_capsule_service import _SECRET_PATTERNS
+
+    if isinstance(value, dict):
+        result = {}
+        for key, item in value.items():
+            if key in {"encrypted_content", "redacted_thinking"}:
+                counts["opaque_reasoning_omitted"] += 1
+                result[key] = "[opaque provider state omitted]"
+            elif re.fullmatch(
+                r"(?i)(?:api[_-]?key|access[_-]?token|refresh[_-]?token|id[_-]?token|password|passwd|secret|client[_-]?secret|cookie|authorization)",
+                key,
+            ):
+                counts["credential_field"] += 1
+                result[key] = "[redacted]"
+            else:
+                result[key] = redact(item, counts)
+        return result
+    if isinstance(value, list):
+        return [redact(item, counts) for item in value]
+    if isinstance(value, str):
+        # Exact controller-provided credentials, without reading credential
+        # files or copying their values into checkpoints or manifests.
+        for name, secret in os.environ.items():
+            if (
+                re.search(r"(?i)(secret|token|password|api_key|mongo_uri)", name)
+                and len(secret) >= 8
+            ):
+                count = value.count(secret)
+                if count:
+                    value = value.replace(secret, "[redacted]")
+                    counts["known_credential"] += count
+        # Existing credential patterns only; preserve source paths/trace frames.
+        for pattern, replacement in _SECRET_PATTERNS[:7]:
+            value, count = pattern.subn(replacement, value)
+            counts["credential_pattern"] += count
+        return value
+    return value
+
+
+def encode(value):
+    return (json.dumps(value, ensure_ascii=False, sort_keys=True) + "\n").encode()
+
+
+def sanitise_lines(raw):
+    counts = Counter()
+    output = []
+    thread_id = None
+    summaries = 0
+    invalid = 0
+    # split on LF only; other Unicode separators belong to the event content.
+    lines = raw.split(b"\n")
+    if lines[-1] == b"":
+        lines.pop()
+    for line in lines:
+        try:
+            value = json.loads(line)
+        except (ValueError, UnicodeDecodeError):
+            invalid += 1
+            value = {
+                "type": "capture.invalid_record",
+                "source_text": line.decode("utf-8", errors="replace"),
+            }
+        if isinstance(value, dict):
+            if value.get("type") == "thread.started":
+                thread_id = value.get("thread_id")
+            item = value.get("item", {})
+            if (
+                isinstance(item, dict)
+                and item.get("type") == "reasoning"
+                and (item.get("text") or item.get("summary"))
+            ):
+                summaries += 1
+        output.append(encode(redact(value, counts)))
+    return b"".join(output), {
+        "redactions": dict(counts),
+        "event_count": len(lines),
+        "invalid_records": invalid,
+        "exposed_summary_records": summaries,
+        "thread_id": thread_id,
+    }
+
+
+def read_batch(path, start, *, terminal=False):
+    if not path.exists():
+        return b""
+    if path.is_symlink():
+        raise ValueError("Capture source must not be a symlink")
+    with path.open("rb") as stream:
+        stream.seek(start)
+        raw = stream.read(BATCH_TARGET)
+        if not raw:
+            return b""
+        if not raw.endswith(b"\n"):
+            rest = stream.readline(service().MAX_BATCH_BYTES - len(raw) + 1)
+            raw += rest
+        if len(raw) > service().MAX_BATCH_BYTES:
+            raise ValueError(
+                "Single activity record exceeds capture bound; source retained locally"
+            )
+        if not terminal and not raw.endswith(b"\n"):
+            end = raw.rfind(b"\n")
+            return raw[: end + 1] if end >= 0 else b""
+        return raw
+
+
+def _context(state):
+    path = Path(state["run_dir"]) / "context.json"
+    if path.is_symlink() or path.stat().st_size > MAX_SOURCE_FILE:
+        raise ValueError("Invalid context capture source")
+    context = json.loads(path.read_text())
+    if context.get("task_id") != state["task_id"]:
+        raise ValueError("Run source does not match selected task")
+    return context
+
+
+def checkpoint(config, state, *, terminal=False):
+    api = service()
+    run_dir = Path(state["run_dir"])
+    if run_dir.is_symlink() or run_dir.name != state["attempt"]:
+        raise ValueError("Run directory does not match attempt identity")
+    context = _context(state)
+    conversation = context.get("conversation", {})
+    if conversation.get("available") and not conversation.get("session_id"):
+        # Legacy transcript pages have no stable common locator. Do not guess
+        # or broaden their audience in historical backfill.
+        raise ValueError(
+            "Available source conversation needs its exact session identity"
+        )
+    provenance = state.get("capture_provenance")
+    if not provenance:
+        provenance = {
+            "started_at": state.get("started_at"),
+            "execution_settings": context.get("execution_settings", {}),
+            "worker_id": context.get("worker_identity"),
+            "capture_worker_script_sha256": api.digest(
+                Path(__file__).with_name("codex_von_worker.py").read_bytes()
+            ),
+            "capture_script_sha256": api.digest(Path(__file__).read_bytes()),
+            "source": "codex exec --json and explicitly named run companions",
+            "original_worker_revision": state.get(
+                "worker_revision", "not recorded by original launcher"
+            ),
+            "backfill": bool(state.get("backfill")),
+        }
+        state["capture_provenance"] = provenance
+    row = api.register_run(
+        task_id=state["task_id"],
+        run_id=state["attempt"],
+        provenance=provenance,
+        source_session_id=(
+            conversation.get("session_id") if conversation.get("available") else None
+        ),
+    )
+    source = run_dir / "events.jsonl"
+    if source.exists() and source.stat().st_size < row["cursor"]:
+        raise ValueError("Local activity source shrank below published cursor")
+    # At most four batches per live checkpoint; terminal catches up completely.
+    batches = 0
+    while terminal or batches < 4:
+        raw = read_batch(source, row["cursor"], terminal=terminal)
+        if not raw:
+            break
+        data, info = sanitise_lines(raw)
+        row = api.append_batch(
+            state["attempt"],
+            start=row["cursor"],
+            end=row["cursor"] + len(raw),
+            data=data,
+            source_sha256=api.digest(raw),
+            event_count=info["event_count"],
+            thread_id=info["thread_id"],
+        )
+        batches += 1
+    state["capture"] = {
+        "status": row["capture_status"],
+        "cursor": row["cursor"],
+        "activity_url": row["activity_url"],
+        "updated_at": row["updated_at"],
+    }
+    if not terminal:
+        return row
+    if row["capture_status"] == "complete":
+        # Verify the durable bytes on recovery, not just a persisted status.
+        api.download_archive(state["attempt"], task_id=state["task_id"])
+        state["capture"]["archive_url"] = row["archive_url"]
+        state["capture"]["source_complete"] = row["manifest"]["source_complete"]
+        return row
+    return finalise(state, row)
+
+
+def finalise(state, row):
+    api = service()
+    run_dir = Path(state["run_dir"])
+    entries = []
+    payloads = {}
+    redactions = Counter()
+    event_info = Counter()
+    source_total = 0
+    for name in ("events.jsonl", *FILES):
+        path = run_dir / name
+        if not path.exists():
+            continue
+        if path.is_symlink() or path.stat().st_size > MAX_SOURCE_FILE:
+            raise ValueError(
+                "Source file exceeds bounded archive route or is a symlink"
+            )
+        raw = path.read_bytes()
+        source_total += len(raw)
+        if source_total > api.MAX_ARCHIVE_BYTES:
+            raise ValueError("Run exceeds bounded archive route; originals retained")
+        if name == "events.jsonl":
+            if len(raw) != row["cursor"]:
+                raise ValueError("Source changed during finalisation")
+            for segment in row["capture_segments"]:
+                if (
+                    api.digest(raw[segment["start"] : segment["end"]])
+                    != segment["source_sha256"]
+                ):
+                    raise ValueError(
+                        "Published source segment changed before finalisation"
+                    )
+            data, info = sanitise_lines(raw)
+            redactions.update(info["redactions"])
+            event_info.update(
+                {
+                    key: info[key]
+                    for key in (
+                        "event_count",
+                        "invalid_records",
+                        "exposed_summary_records",
+                    )
+                }
+            )
+        else:
+            counts = Counter()
+            try:
+                value = (
+                    json.loads(raw) if name.endswith(".json") else raw.decode("utf-8")
+                )
+            except (ValueError, UnicodeDecodeError):
+                value = raw.decode("utf-8", errors="replace")
+                counts["invalid_source_encoding_or_json"] += 1
+            cleaned = redact(value, counts)
+            data = encode(cleaned) if name.endswith(".json") else cleaned.encode()
+            redactions.update(counts)
+        payloads[name] = data
+        entries.append(
+            {
+                "name": name,
+                "source_bytes": len(raw),
+                "source_sha256": api.digest(raw),
+                "bytes": len(data),
+                "sha256": api.digest(data),
+            }
+        )
+    # An absent event file is an explicit incomplete source, not an omitted ZIP member.
+    if "events.jsonl" not in payloads:
+        payloads["events.jsonl"] = b""
+        entries.append(
+            {
+                "name": "events.jsonl",
+                "source_bytes": 0,
+                "source_sha256": api.digest(b""),
+                "bytes": 0,
+                "sha256": api.digest(b""),
+            }
+        )
+    missing = sorted(
+        REQUIRED_FILES - {p.name for p in run_dir.iterdir() if p.is_file()}
+    )
+    manifest = {
+        "schema_version": "von_task_run_archive.v1",
+        "run_id": state["attempt"],
+        "task_id": state["task_id"],
+        "thread_id": row["thread_id"],
+        "provenance": row["provenance"],
+        "finished_at": state.get("finished_at") or api.now(),
+        "outcome": state.get("result", {}).get("status", "interrupted"),
+        "source_segments": row["capture_segments"],
+        "final_cursor": row["cursor"],
+        "event_count": event_info["event_count"],
+        "files": entries,
+        "missing_files": missing,
+        "source_complete": not missing
+        and not event_info["invalid_records"]
+        and not redactions["invalid_source_encoding_or_json"],
+        "invalid_records": event_info["invalid_records"],
+        "redactions": dict(redactions),
+        "reasoning_summaries": {
+            "exposed_records": event_info["exposed_summary_records"],
+            "availability": (
+                "exposed"
+                if event_info["exposed_summary_records"]
+                else "not exposed in captured exec stream"
+            ),
+        },
+        "limitations": [
+            "Only provider-exposed exec activity and listed companion files are captured.",
+            "No hidden chain of thought is extracted, decrypted or claimed.",
+            "Source rollout not collected: no supported bounded export route is configured; no session discovery is performed.",
+            "Credential patterns and known controller credentials are redacted; opaque provider state is omitted.",
+            "The readable view is limited; archived redacted records retain available command outputs without display truncation.",
+        ],
+    }
+    buffer = io.BytesIO()
+    with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as package:
+        for name, data in payloads.items():
+            package.writestr(name, data)
+        package.writestr("manifest.json", encode(manifest))
+    verified = api.finalise_run(
+        state["attempt"], archive=buffer.getvalue(), manifest=manifest
+    )
+    state["capture"] = {
+        "status": "complete",
+        "cursor": verified["cursor"],
+        "archive_url": verified["archive_url"],
+        "activity_url": verified["activity_url"],
+        "source_complete": manifest["source_complete"],
+        "updated_at": verified["updated_at"],
+    }
+    return verified
+
+
+def try_checkpoint(config, state, *, terminal=False):
+    try:
+        checkpoint(config, state, terminal=terminal)
+        return True
+    except Exception as exc:  # noqa: BLE001
+        # Archive failures must preserve coding progress.
+        # Retain a safe error class, never a storage URL/credential-bearing exception.
+        state["capture"] = state.get("capture", {}) | {
+            "status": "pending",
+            "error_type": type(exc).__name__,
+        }
+        try:
+            service().capture_failure(state["attempt"], type(exc).__name__)
+        except Exception as report_exc:  # noqa: BLE001 - retain safe diagnostic only
+            state["capture"]["failure_report_error_type"] = type(report_exc).__name__
+        return False
diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py
index d8f69aa18..3fb0dc892 100755
--- a/scripts/codex_von_worker.py
+++ b/scripts/codex_von_worker.py
@@ -19,9 +19,11 @@
 import sys
 import uuid
 from contextlib import ExitStack
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from pathlib import Path
 
+ACTIVITY_CHECKPOINT_SECONDS = 15
+
 ACTIVE = {"pending", "in_progress", "blocked"}
 RESULT_SCHEMA = {
     "type": "object",
@@ -127,7 +129,7 @@ def authorised_task(task, config):
 def runtime_evidence(backend_root, api):
     root = Path(backend_root).resolve()
     return {
-        "observed_at": datetime.now(timezone.utc).isoformat(),
+        "observed_at": datetime.now(UTC).isoformat(),
         "pid": os.getpid(),
         "commit": subprocess.check_output(
             ["git", "-C", str(root), "rev-parse", "HEAD"], text=True
@@ -384,7 +386,7 @@ def conversation(self, task):
             pages.append(page)
             cursor = page.get("next_cursor")
             if not cursor:
-                return {"available": True, "pages": pages}
+                return {"available": True, "session_id": session_id, "pages": pages}
 
     def send(self, task, key, content):
         c = self.config
@@ -443,6 +445,11 @@ def finish(self, state):
                 f"reasoning: {settings['reasoning_effort']} "
                 f"({settings['reasoning_effort_source']})."
             )
+        capture = state.get("capture", {})
+        if capture.get("status") == "complete":
+            content += f"\nVerified execution archive: {capture['archive_url']}"
+            if not capture.get("source_complete", True):
+                content += "\nThe manifest records incomplete source capture."
         # Keep the delivery intent stable across a crash after a task transition.
         # A changed/cancelled task is left untouched; the run evidence still goes
         # to the original authorised delegator.
@@ -460,12 +467,20 @@ def finish(self, state):
             state["report_task"], state["attempt"] + ":result", state["report_content"]
         )
         if scope_valid:
+            evidence_addition = (
+                f"{result['evidence']}\nVon result: {message_id}\n"
+                f"Execution archive: {capture.get('archive_url', 'unavailable')}\n"
+                f"DGX worktree: {state.get('worktree', 'not started')}"
+            )
+            evidence = task.get("evidence") or ""
+            if evidence_addition not in evidence:
+                evidence = "\n".join(filter(None, [evidence, evidence_addition]))
             # Idempotent reconciliation through the existing task evidence field.
             self.tasks.update_task_fields(
                 state["task_id"],
                 fields={
                     "progress_signal": result["summary"],
-                    "evidence": f"{result['evidence']}\nVon result: {message_id}\nDGX worktree: {state.get('worktree', 'not started')}",
+                    "evidence": evidence,
                     "next_checkpoint": result["question"]
                     or "Result available in Von messages.",
                 },
@@ -493,6 +508,44 @@ def command(args, **kwargs):
     ).stdout.strip()
 
 
+def activity_module():
+    try:
+        from . import codex_von_activity
+    except ImportError:
+        try:
+            import codex_von_activity
+        except ImportError:
+            from scripts import codex_von_activity
+    return codex_von_activity
+
+
+def archive_checkpoint(config, state, *, terminal=False):
+    return activity_module().try_checkpoint(config, state, terminal=terminal)
+
+
+def finish_captured(config, api, state, state_path):
+    if not archive_checkpoint(config, state, terminal=True):
+        state["phase"] = "archive_pending"
+        write_json(state_path, state)
+        if not state.get("archive_pending_reported"):
+            api.send(
+                {
+                    "task_concept_id": state["task_id"],
+                    "title": state.get("title", state["task_id"]),
+                },
+                state["attempt"] + ":archive-pending",
+                f"Coding outcome: {state['result']['status']}. {state['result']['summary']}\n"
+                "Execution archive finalisation is pending; the existing controller will retry. "
+                "The task is not being marked completed. Local originals are retained.",
+            )
+            state["archive_pending_reported"] = True
+        write_json(state_path, state)
+        return
+    state["phase"] = "reporting"
+    api.finish(state)
+    write_json(state_path, state)
+
+
 def launch(config, state, inputs, conversation, state_path, lock_fd):
     root = Path(config["state_root"])
     task_key = fingerprint(state["task_id"])[:16]
@@ -518,7 +571,16 @@ def launch(config, state, inputs, conversation, state_path, lock_fd):
         worktree=str(worktree),
         run_dir=str(run_dir),
         input_hash=fingerprint(inputs),
+        started_at=datetime.now(UTC).isoformat(),
+        worker_revision=config.get("runtime_evidence", {}).get("commit"),
     )
+    for key in (
+        "capture",
+        "capture_provenance",
+        "archive_pending_reported",
+        "finished_at",
+    ):
+        state.pop(key, None)
     state.pop("result", None)
     state.pop("report_task", None)
     state.pop("report_content", None)
@@ -530,6 +592,9 @@ def launch(config, state, inputs, conversation, state_path, lock_fd):
     context["execution_settings"] = settings
     write_json(context_path, context)
     write_json(state_path, state)
+    # Capture failure must not prevent useful bounded coding progress.
+    archive_checkpoint(config, state)
+    write_json(state_path, state)
     if not state.get("checkout_prepared"):
         worktree.parent.mkdir(parents=True, exist_ok=True)
         origin = command(
@@ -621,22 +686,34 @@ def launch(config, state, inputs, conversation, state_path, lock_fd):
         ]
     else:
         args[2:2] = ["--sandbox", "workspace-write"]
-    with (
-        (run_dir / "events.jsonl").open("w") as events,
+    with (  # noqa: SIM117 - streams must outlive the process context
+        (run_dir / "events.jsonl").open("a") as events,
         (run_dir / "stderr.log").open("w") as errors,
     ):
         # The inherited lock keeps another poll out even if this controller exits.
-        completed = subprocess.run(
+        with subprocess.Popen(
             args,
-            check=False,
-            input=prompt,
+            stdin=subprocess.PIPE,
             text=True,
             env=environment,
             stdout=events,
             stderr=errors,
             pass_fds=(lock_fd,),
-        )
-    write_json(run_dir / "exit.json", {"returncode": completed.returncode})
+        ) as process:
+            process.stdin.write(prompt)
+            process.stdin.close()
+            while True:
+                try:
+                    process.wait(timeout=ACTIVITY_CHECKPOINT_SECONDS)
+                    break
+                except subprocess.TimeoutExpired:
+                    archive_checkpoint(config, state)
+                    write_json(state_path, state)
+    state["finished_at"] = datetime.now(UTC).isoformat()
+    write_json(
+        run_dir / "exit.json",
+        {"returncode": process.returncode, "finished_at": state["finished_at"]},
+    )
     recover_result(state)
     write_json(state_path, state)
 
@@ -757,19 +834,62 @@ def inbox_module():
     return codex_von_inbox
 
 
+def backfill_run(config, api, selection):
+    """Operator-selected pairs only; no session discovery or task rerun."""
+    task_id, separator, attempt = selection.partition("=")
+    if (
+        not separator
+        or len(attempt) != 32
+        or any(c not in "0123456789abcdef" for c in attempt)
+    ):
+        raise ValueError("Backfill requires TASK_CONCEPT_ID=32_HEX_ATTEMPT")
+    task = api.task(task_id)
+    if not authorised_task(task, config) or not api.native_writer(task):
+        raise PermissionError("Backfill task is outside this worker assignment")
+    run_dir = Path(config["state_root"]) / "runs" / attempt
+    context = json.loads((run_dir / "context.json").read_text())
+    if (
+        context.get("task_id") != task_id
+        or context.get("worker_identity") != config["agent_id"]
+    ):
+        raise ValueError("Backfill source does not match selected task/worker")
+    path = Path(config["state_root"]) / "archive-backfills" / (attempt + ".json")
+    state = (
+        json.loads(path.read_text())
+        if path.exists()
+        else {
+            "task_id": task_id,
+            "attempt": attempt,
+            "run_dir": str(run_dir),
+            "backfill": True,
+            "execution_settings": context.get("execution_settings", {}),
+        }
+    )
+    recover_result(state)
+    archive_checkpoint(config, state, terminal=True)
+    write_json(path, state)
+    return {"task_id": task_id, "attempt": attempt, "capture": state["capture"]}
+
+
 def tick(config, api, lock_fd):
     states_dir = Path(config["state_root"]) / "tasks"
     states_dir.mkdir(parents=True, exist_ok=True)
+    for path in sorted(
+        (Path(config["state_root"]) / "archive-backfills").glob("*.json")
+    ):
+        backfill = json.loads(path.read_text())
+        if backfill.get("capture", {}).get("status") != "complete":
+            archive_checkpoint(config, backfill, terminal=True)
+            write_json(path, backfill)
     # Reconcile an interrupted report before selecting any new work.
     for path in sorted(states_dir.glob("*.json")):
         state = json.loads(path.read_text())
         if state["phase"] == "running":
             recover_result(state)
             write_json(path, state)
-        if state["phase"] == "reporting":
+        if state["phase"] in {"reporting", "archive_pending"}:
             apply_deployment(config, api, state, path, lock_fd)
-            api.finish(state)
-            write_json(path, state)
+            finish_captured(config, api, state, path)
     if config.get("inbox_enabled") and inbox_module().tick(config, api, lock_fd):
         return
     for task in api.pending():
@@ -782,6 +902,8 @@ def tick(config, api, lock_fd):
             if path.exists()
             else {"task_id": task_id, "phase": "new"}
         )
+        if state["phase"] == "archive_pending":
+            continue
         state["title"] = task["title"]
         inputs = api.inputs(task)
         if (
@@ -820,14 +942,13 @@ def tick(config, api, lock_fd):
                 ] = "Correct this task's model or reasoning setting, then requeue it."
             write_json(path, state)
         apply_deployment(config, api, state, path, lock_fd)
-        api.finish(state)
-        write_json(path, state)
+        finish_captured(config, api, state, path)
         print(
             json.dumps(
                 {
                     "outcome": state["phase"],
                     "task_id": task_id,
-                    "message_id": state["message_id"],
+                    "message_id": state.get("message_id"),
                 }
             ),
             flush=True,
@@ -847,7 +968,15 @@ def main():
     parser.add_argument(
         "--check-task", help="Read back one assigned task without running work"
     )
+    parser.add_argument(
+        "--backfill-run",
+        action="append",
+        default=[],
+        help="Explicit TASK_CONCEPT_ID=ATTEMPT pair; at most 20; never executes coding",
+    )
     options = parser.parse_args()
+    if len(options.backfill_run) > 20:
+        parser.error("Backfill is bounded to 20 selected pairs")
     os.umask(0o077)
     config = json.loads(Path(options.config).read_text())
     resolve_execution_settings(config, {})
@@ -894,11 +1023,16 @@ def main():
                 )
             api = Von(config)
             evidence = runtime_evidence(backend_root, api)
+            config["runtime_evidence"] = evidence
             if options.check_task:
                 evidence["task_check"] = check_task(config, api, options.check_task)
                 print(json.dumps(evidence), flush=True)
                 return
             write_json(state_root / "runtime.json", evidence)
+            if options.backfill_run:
+                for selection in options.backfill_run:
+                    print(json.dumps(backfill_run(config, api, selection)), flush=True)
+                return
             tick(config, api, lock.fileno())
 
 
diff --git a/src/backend/server/routes/task_routes.py b/src/backend/server/routes/task_routes.py
index 5ad8c15c5..bad0e8175 100644
--- a/src/backend/server/routes/task_routes.py
+++ b/src/backend/server/routes/task_routes.py
@@ -1274,3 +1274,49 @@ def remove_task_link_route(task_concept_id: str) -> ResponseReturnValue:
 def remove_task_link_post_route(task_concept_id: str) -> ResponseReturnValue:
     """Remove a typed link between tasks via POST payload."""
     return remove_task_link_route(task_concept_id)
+
+@task_bp.route("/<task_concept_id>/runs", methods=["GET"])
+@task_bp.route("/<task_concept_id>/runs/<run_id>", methods=["GET"])
+@task_bp.route("/<task_concept_id>/runs/<run_id>/download", methods=["GET"])
+def task_run_archive_route(task_concept_id: str, run_id: str | None = None):
+    """Read private worker activity through the trusted browser actor context."""
+    from flask import Response
+    from ...services import task_run_archive_service as archives
+
+    try:
+        # Reject client identity headers even in a permissive local profile.
+        if not _get_current_user_concept_id() or not _get_current_org_concept_id():
+            return jsonify({"error": "Trusted actor and organisation required"}), 401
+        if run_id is None:
+            result = archives.list_runs(
+                task_concept_id,
+                offset=max(0, int(request.args.get("offset", 0))),
+                limit=50,
+            )
+        elif request.path.endswith("/download"):
+            data = archives.download_archive(run_id, task_id=task_concept_id)
+            return Response(
+                data,
+                mimetype="application/zip",
+                headers={
+                    "Content-Disposition": f'attachment; filename="coding-run-{run_id}.zip"',
+                    "Cache-Control": "private, no-store",
+                    "X-Content-Type-Options": "nosniff",
+                },
+            )
+        else:
+            result = archives.read_activity(
+                run_id,
+                task_id=task_concept_id,
+                cursor=max(0, int(request.args.get("cursor", 0))),
+            )
+        response = jsonify(result)
+        response.headers["Cache-Control"] = "private, no-store"
+        return response
+    except PermissionError:
+        return jsonify({"error": "Run unavailable in this actor scope"}), 403
+    except ValueError:
+        return jsonify({"error": "Invalid activity cursor"}), 400
+    except Exception:
+        logger.warning("Task run archive read failed", exc_info=False)
+        return jsonify({"error": "Run capture storage unavailable; retry later"}), 503
diff --git a/src/backend/services/task_run_archive_service.py b/src/backend/services/task_run_archive_service.py
new file mode 100644
index 000000000..6874420a0
--- /dev/null
+++ b/src/backend/services/task_run_archive_service.py
@@ -0,0 +1,399 @@
+"""Private coding-run records and immutable activity blobs (JVNAUTOSCI-2751).
+
+Operational records, not task semantics. Only the trusted controller writes;
+HTTP exposes actor-bound reads. Task visibility never grants archive access.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import re
+import zipfile
+from datetime import UTC, datetime
+from urllib.parse import quote
+
+from ..db.mongo_client import get_db
+from ..security.access_control import (
+    LEGACY_IDENTITY_HEADER_ACTOR_SOURCE,
+    get_effective_organisation_concept_id,
+    get_effective_user_concept_id_with_source,
+)
+from .blob_store import get_blob_store_for_backend_from_env, get_blob_store_from_env
+from .organisation_membership_service import resolve_user_organisation_membership
+
+MAX_BATCH_BYTES = 16 * 1024 * 1024
+MAX_ARCHIVE_BYTES = 256 * 1024 * 1024
+SCHEMA = "von_task_run.v1"
+
+
+def digest(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def now():
+    return datetime.now(UTC).isoformat()
+
+
+def collection():
+    db = get_db()
+    if db is None:
+        raise RuntimeError("Task run archive storage unavailable")
+    return db["task_run_archives"]
+
+
+def actor_scope():
+    actor, source = get_effective_user_concept_id_with_source()
+    org = get_effective_organisation_concept_id()
+    if (
+        not actor
+        or not org
+        or source == LEGACY_IDENTITY_HEADER_ACTOR_SOURCE
+        or not resolve_user_organisation_membership(actor, org)
+    ):
+        raise PermissionError("Task run access requires a trusted organisation actor")
+    return actor, org
+
+
+def _run(run_id, *, write=False):
+    actor, org = actor_scope()
+    row = collection().find_one(
+        {"_id": run_id, "organisation_id": org, "audience": actor}
+    )
+    if not row or (write and row["worker_id"] != actor):
+        raise PermissionError("Task run unavailable in this actor scope")
+    # Source access is checked again, so revoked conversation access does not
+    # leave a permanent second copy accessible via the task.
+    if row.get("source_session_id"):
+        from .shared_conversation_service import (
+            get_accepted_invite_for_user_session,
+            resolve_conversation_owner,
+        )
+
+        sid = row["source_session_id"]
+        if resolve_conversation_owner(session_id=sid) != actor:
+            invite = get_accepted_invite_for_user_session(
+                user_concept_id=actor, session_id=sid
+            )
+            if not invite or invite.get("organisation_concept_id") != org:
+                raise PermissionError(
+                    "Source conversation access is no longer available"
+                )
+    return row
+
+
+def register_run(*, task_id, run_id, provenance, source_session_id=None):
+    from .task_management_service import get_task
+
+    actor, org = actor_scope()
+    if not re.fullmatch(r"[a-f0-9]{32}", run_id):
+        raise ValueError("Invalid run identity")
+    existing = collection().find_one({"_id": run_id})
+    if existing:
+        row = _run(run_id, write=True)
+        if (
+            row["task_id"] != task_id
+            or row.get("source_session_id") != source_session_id
+        ):
+            raise ValueError("Run identity already bound to another source")
+        return public_record(row)
+    task = get_task(task_id)
+    if (
+        task.get("assignee_concept_id") != actor
+        or task.get("organisation_concept_id") != org
+    ):
+        raise PermissionError("Run must belong to the assigned worker and organisation")
+    owner = task.get("created_by_concept_id")
+    audience = [actor]
+    if owner and resolve_user_organisation_membership(owner, org):
+        audience.append(owner)
+    if source_session_id:
+        from .shared_conversation_service import (
+            get_accepted_invite_for_user_session,
+            resolve_conversation_owner,
+        )
+
+        source_owner = resolve_conversation_owner(session_id=source_session_id)
+        audience = [
+            user
+            for user in audience
+            if user == source_owner
+            or (
+                (
+                    get_accepted_invite_for_user_session(
+                        user_concept_id=user, session_id=source_session_id
+                    )
+                    or {}
+                ).get("organisation_concept_id")
+                == org
+            )
+        ]
+        if actor not in audience:
+            raise PermissionError("Worker cannot archive this source conversation")
+    row = {
+        "_id": run_id,
+        "schema_version": SCHEMA,
+        "task_id": task_id,
+        "worker_id": actor,
+        "requester_id": owner,
+        "organisation_id": org,
+        "audience": audience,
+        "source_session_id": source_session_id,
+        "provenance": provenance,
+        "started_at": provenance.get("started_at") or now(),
+        "updated_at": now(),
+        "capture_status": "running",
+        "cursor": 0,
+        "event_count": 0,
+        "batches": [],
+        "thread_id": None,
+    }
+    collection().update_one({"_id": run_id}, {"$setOnInsert": row}, upsert=True)
+    stored = _run(run_id, write=True)
+    if stored["task_id"] != task_id or stored["provenance"] != provenance:
+        raise ValueError("Conflicting run registration")
+    return public_record(stored)
+
+
+def public_record(row):
+    return {k: v for k, v in row.items() if k not in {"_id", "audience", "batches"}} | {
+        "capture_segments": [
+            {k: b[k] for k in ("start", "end", "source_sha256")} for b in row["batches"]
+        ],
+        "run_id": row["_id"],
+        "activity_url": run_url(row["task_id"], row["_id"]),
+    }
+
+
+def run_url(task_id, run_id):
+    return f"/api/tasks/{quote(task_id, safe='')}/runs/{run_id}"
+
+
+def list_runs(task_id, *, offset=0, limit=50):
+    actor, org = actor_scope()
+    rows = (
+        collection()
+        .find({"task_id": task_id, "organisation_id": org, "audience": actor})
+        .sort("started_at", -1)
+        .skip(max(0, offset))
+        .limit(min(100, max(1, limit)))
+    )
+    visible = []
+    for row in rows:
+        try:
+            visible.append(public_record(_run(row["_id"])))
+        except PermissionError:
+            continue
+    return {
+        "runs": visible,
+        "offset": offset,
+        "next_offset": offset + len(visible) if len(visible) == limit else None,
+    }
+
+
+def _put(data, key):
+    store = get_blob_store_from_env()
+    ref = store.put_bytes(key, data, content_type="application/octet-stream")
+    receipt = {
+        "backend": ref.backend,
+        "key": ref.key,
+        "sha256": digest(data),
+        "bytes": len(data),
+    }
+    _get(receipt)
+    return receipt
+
+
+def _get(receipt):
+    data = get_blob_store_for_backend_from_env(receipt["backend"]).get_bytes(
+        receipt["key"]
+    )
+    if len(data) != receipt["bytes"] or digest(data) != receipt["sha256"]:
+        raise RuntimeError("Archive hash/byte read-back mismatch")
+    return data
+
+
+def append_batch(
+    run_id, *, start, end, data, source_sha256, event_count, thread_id=None
+):
+    row = _run(run_id, write=True)
+    if not isinstance(data, bytes) or not data or len(data) > MAX_BATCH_BYTES:
+        raise ValueError("Invalid activity batch size")
+    if end <= start or event_count < 1:
+        raise ValueError("Invalid activity cursor")
+    old = next((b for b in row["batches"] if b["start"] == start), None)
+    if old:
+        if (old["end"], old["sha256"], old["source_sha256"], old["event_count"]) != (
+            end,
+            digest(data),
+            source_sha256,
+            event_count,
+        ):
+            raise ValueError("Conflicting activity retry")
+        _get(old)
+        return public_record(row)
+    if row["capture_status"] == "complete" or row["cursor"] != start:
+        raise ValueError("Activity cursor is not contiguous")
+    receipt = _put(data, f"task-runs/{run_id}/activity/{start}-{digest(data)}")
+    batch = receipt | {
+        "start": start,
+        "end": end,
+        "source_sha256": source_sha256,
+        "event_count": event_count,
+    }
+    updated = collection().update_one(
+        {"_id": run_id, "cursor": start, "capture_status": {"$ne": "complete"}},
+        {
+            "$push": {"batches": batch},
+            "$set": {
+                "cursor": end,
+                "updated_at": now(),
+                "capture_status": "running",
+                "thread_id": thread_id or row["thread_id"],
+            },
+            "$inc": {"event_count": event_count},
+        },
+    )
+    if not updated.modified_count:
+        return append_batch(
+            run_id,
+            start=start,
+            end=end,
+            data=data,
+            source_sha256=source_sha256,
+            event_count=event_count,
+            thread_id=thread_id,
+        )
+    return public_record(_run(run_id))
+
+
+def capture_failure(run_id, error_type):
+    _run(run_id, write=True)
+    collection().update_one(
+        {"_id": run_id, "capture_status": {"$ne": "complete"}},
+        {
+            "$set": {
+                "capture_status": "pending",
+                "capture_error": error_type,
+                "updated_at": now(),
+            }
+        },
+    )
+
+
+def read_activity(run_id, *, task_id=None, cursor=0):
+    row = _run(run_id)
+    if task_id is not None and row["task_id"] != task_id:
+        raise PermissionError("Run does not belong to this task")
+    batch = next((b for b in row["batches"] if b["start"] == cursor), None)
+    records = []
+    if batch:
+        for line in _get(batch).decode("utf-8").splitlines():
+            value = json.loads(line)
+            rendered = json.dumps(value, ensure_ascii=False, indent=2)
+            records.append(
+                {"text": rendered[:16000], "display_truncated": len(rendered) > 16000}
+            )
+    return {
+        "run": public_record(row),
+        "records": records,
+        "next_cursor": batch["end"] if batch else cursor,
+        "has_more": bool(batch and batch["end"] < row["cursor"]),
+        "projection": "Up to 16000 characters per record; download retains full redacted records.",
+    }
+
+
+def finalise_run(run_id, *, archive, manifest):
+    from .task_management_service import (
+        TaskNotFoundError,
+        add_task_attachment,
+        get_task,
+    )
+
+    row = _run(run_id, write=True)
+    if row["capture_status"] == "complete":
+        _get(row["archive"])
+        return public_record(row)
+    if len(archive) > MAX_ARCHIVE_BYTES:
+        raise ValueError("Archive exceeds bounded upload size")
+    if (
+        manifest.get("run_id") != run_id
+        or manifest.get("task_id") != row["task_id"]
+        or manifest.get("final_cursor") != row["cursor"]
+        or manifest.get("event_count") != row["event_count"]
+    ):
+        raise ValueError("Final archive does not reconcile captured activity")
+    expected_segments = [
+        {k: b[k] for k in ("start", "end", "source_sha256")} for b in row["batches"]
+    ]
+    if manifest.get("source_segments") != expected_segments:
+        raise ValueError("Final source hashes differ from captured segments")
+    with zipfile.ZipFile(io.BytesIO(archive)) as package:
+        embedded = json.loads(package.read("manifest.json"))
+        if embedded != manifest:
+            raise ValueError("Manifest differs from archive")
+        names = [entry["name"] for entry in manifest["files"]]
+        if len(set(names)) != len(names) or set(package.namelist()) != set(
+            names + ["manifest.json"]
+        ):
+            raise ValueError("Archive file inventory mismatch")
+        if sum(item.file_size for item in package.infolist()) > MAX_ARCHIVE_BYTES:
+            raise ValueError("Uncompressed archive exceeds bounded read route")
+        for entry in manifest["files"]:
+            data = package.read(entry["name"])
+            if len(data) != entry["bytes"] or digest(data) != entry["sha256"]:
+                raise ValueError("Archive file hash/byte mismatch")
+        events = package.read("events.jsonl")
+        if events != b"".join(_get(batch) for batch in row["batches"]):
+            raise ValueError("Archive differs from published activity")
+    receipt = _put(archive, f"task-runs/{run_id}/archive/{digest(archive)}.zip")
+    url = run_url(row["task_id"], run_id) + "/download"
+    # Attachment metadata reveals no source content. Download rechecks the run
+    # audience; never use org-wide file-copy visibility for private transcripts.
+    try:
+        task = get_task(row["task_id"])
+    except TaskNotFoundError:
+        task = {}
+    attachment = {}
+    if (
+        task.get("assignee_concept_id") == row["worker_id"]
+        and task.get("created_by_concept_id") == row["requester_id"]
+        and task.get("organisation_concept_id") == row["organisation_id"]
+        and task.get("status") != "cancelled"
+    ):
+        attachment = add_task_attachment(
+            row["task_id"],
+            filename=f"coding-run-{run_id}.zip",
+            uri=url,
+            media_type="application/zip",
+            size_bytes=len(archive),
+            added_by_concept_id=row["worker_id"],
+            source={"source_system": "von_task_run", "external_id": run_id},
+        )
+    collection().update_one(
+        {"_id": run_id, "cursor": row["cursor"]},
+        {
+            "$set": {
+                "archive": receipt,
+                "archive_url": url,
+                "manifest": manifest,
+                "attachment_id": attachment.get("attachment_id"),
+                "attachment_status": (
+                    "linked" if attachment else "task_changed_or_unavailable"
+                ),
+                "capture_status": "complete",
+                "updated_at": now(),
+                "finished_at": manifest["finished_at"],
+                "capture_error": None,
+            }
+        },
+    )
+    return public_record(_run(run_id))
+
+
+def download_archive(run_id, *, task_id):
+    row = _run(run_id)
+    if row["task_id"] != task_id or row["capture_status"] != "complete":
+        raise PermissionError("Completed archive unavailable")
+    return _get(row["archive"])
diff --git a/src/frontend/web/von_interface/static/js/components/taskPanel.js b/src/frontend/web/von_interface/static/js/components/taskPanel.js
index fa0c8108a..a826a531e 100644
--- a/src/frontend/web/von_interface/static/js/components/taskPanel.js
+++ b/src/frontend/web/von_interface/static/js/components/taskPanel.js
@@ -5,6 +5,7 @@
  * Tasks are stored as Vontology concepts and accessed via REST API.
  */
 
+import { openTaskRunActivity } from './taskRunActivity.js';
 import { deleteJson, getJson, patchJson, postJson, getUserContext, ensureUniqueWindowSessionId, WINDOW_SESSION_HEADER } from '../apiService.js';
 import { activateTab } from '../tabNavigation.js';
 import { showToast } from '../utils/toast.js';
@@ -2899,6 +2900,7 @@ function renderTaskDetailsPanel(task, detailState) {
 
             <div class="task-detail-section">
                 <h4>Attachments</h4>
+                <button class="task-run-activity-btn" data-task-id="${escapeHtml(taskId)}">Coding run activity</button>
                 <ul class="task-detail-list">
                     ${renderTaskAttachmentRows(detailState.attachments)}
                 </ul>
@@ -3205,6 +3207,7 @@ function renderTaskInspector(task, detailState) {
 
             <div class="task-detail-section">
                 <h4>Attachments</h4>
+                <button class="task-run-activity-btn" data-task-id="${escapeHtml(taskId)}">Coding run activity</button>
                 <ul class="task-detail-list">
                     ${renderTaskAttachmentRows(detailState?.attachments)}
                 </ul>
@@ -3663,6 +3666,13 @@ function attachTaskEventListeners() {
     if (roots.length === 0) return;
 
     roots.forEach((root) => {
+        root.querySelectorAll('.task-run-activity-btn').forEach((button) => {
+            button.addEventListener('click', (event) => {
+                event.preventDefault();
+                event.stopPropagation();
+                void openTaskRunActivity(button.dataset.taskId);
+            });
+        });
         root.querySelectorAll('.task-execute-with-von-btn').forEach((button) => {
             button.addEventListener('click', async (event) => {
                 event.preventDefault();
diff --git a/src/frontend/web/von_interface/static/js/components/taskRunActivity.js b/src/frontend/web/von_interface/static/js/components/taskRunActivity.js
new file mode 100644
index 000000000..f8990366b
--- /dev/null
+++ b/src/frontend/web/von_interface/static/js/components/taskRunActivity.js
@@ -0,0 +1,107 @@
+/** Private, inert coding-run records. No markdown/HTML execution of history. */
+import { getJson, getUserContext, ensureUniqueWindowSessionId, WINDOW_SESSION_HEADER } from '../apiService.js';
+
+export async function openTaskRunActivity(taskId) {
+    const scope = JSON.stringify(getUserContext());
+    const dialog = document.createElement('dialog');
+    dialog.style.cssText = 'width:min(900px,92vw);max-height:85vh;overflow:auto;box-sizing:border-box';
+    const heading = document.createElement('h3');
+    heading.textContent = 'Coding run activity';
+    const close = document.createElement('button');
+    close.textContent = 'Close';
+    close.addEventListener('click', () => dialog.close());
+    const refresh = document.createElement('button');
+    refresh.textContent = 'Refresh runs';
+    const status = document.createElement('p');
+    status.setAttribute('role', 'status');
+    const runs = document.createElement('div');
+    const activity = document.createElement('div');
+    dialog.append(heading, close, refresh, status, runs, activity);
+    const clearForScopeChange = () => dialog.close();
+    document.addEventListener('orgSwitched', clearForScopeChange);
+    document.addEventListener('authStatusChanged', clearForScopeChange);
+    dialog.addEventListener('close', () => {
+        document.removeEventListener('orgSwitched', clearForScopeChange);
+        document.removeEventListener('authStatusChanged', clearForScopeChange);
+        dialog.remove();
+    }, { once: true });
+    document.body.append(dialog);
+    dialog.showModal();
+    const base = `/api/tasks/${encodeURIComponent(taskId)}/runs`;
+    let generation = 0;
+    function current(requestGeneration) {
+        if (scope !== JSON.stringify(getUserContext())) dialog.close();
+        return dialog.isConnected && requestGeneration === generation;
+    }
+    function button(label, action, parent) {
+        const element = document.createElement('button');
+        element.textContent = label;
+        element.addEventListener('click', action);
+        parent.append(element);
+    }
+    async function showRun(runId, cursor = 0) {
+        const requestGeneration = ++generation;
+        status.textContent = 'Loading captured activity…';
+        try {
+            const result = await getJson(`${base}/${encodeURIComponent(runId)}?cursor=${cursor}`);
+            if (!current(requestGeneration)) return;
+            activity.replaceChildren();
+            const run = result.run;
+            status.textContent = `${run.capture_status} · Last captured ${run.updated_at} · ${run.event_count} records. ${run.capture_error ? `Capture needs retry (${run.capture_error}).` : ''}`;
+            const info = document.createElement('p');
+            info.textContent = `Run ${runId} · ${run.provenance?.execution_settings?.model || 'Model not recorded'} · reasoning effort ${run.provenance?.execution_settings?.reasoning_effort || 'not recorded'}. ${result.projection} Reasoning summaries: ${run.manifest?.reasoning_summaries?.availability || 'availability recorded at finalisation'}.`;
+            activity.append(info);
+            if (run.manifest?.source_complete === false) {
+                const warning = document.createElement('p');
+                warning.textContent = 'Archive stored; source capture is incomplete. See the manifest for missing or malformed records.';
+                activity.append(warning);
+            }
+            if (run.archive_url) button('Download archive', async () => {
+                try {
+                    const response = await fetch(run.archive_url, { headers: {
+                        [WINDOW_SESSION_HEADER]: await ensureUniqueWindowSessionId()
+                    }});
+                    if (!response.ok) throw new Error('Download unavailable');
+                    const blob = await response.blob();
+                    if (!current(requestGeneration)) return;
+                    const url = URL.createObjectURL(blob);
+                    const link = document.createElement('a');
+                    link.href = url;
+                    link.download = `coding-run-${runId}.zip`;
+                    link.click();
+                    setTimeout(() => URL.revokeObjectURL(url), 1000);
+                } catch {
+                    if (current(requestGeneration)) status.textContent = 'Archive download unavailable; refresh and retry.';
+                }
+            }, activity);
+            button('Refresh this page', () => showRun(runId, cursor), activity);
+            if (cursor) button('First records', () => showRun(runId), activity);
+            for (const record of result.records) {
+                const pre = document.createElement('pre');
+                pre.style.cssText = 'white-space:pre-wrap;overflow-wrap:anywhere;max-width:100%';
+                pre.textContent = record.text + (record.display_truncated ? '\n[Display shortened; full redacted output in archive]' : '');
+                activity.append(pre);
+            }
+            if (result.next_cursor > cursor) button(result.has_more ? 'Next records' : 'Check for newer records', () => showRun(runId, result.next_cursor), activity);
+        } catch {
+            if (current(requestGeneration)) status.textContent = 'Activity unavailable in this scope or capture storage is offline. Refresh to retry.';
+        }
+    }
+    async function loadRuns(offset = 0) {
+        const requestGeneration = ++generation;
+        status.textContent = 'Loading runs…';
+        try {
+            const result = await getJson(`${base}?offset=${offset}`);
+            if (!current(requestGeneration)) return;
+            runs.replaceChildren();
+            activity.replaceChildren();
+            status.textContent = result.runs.length ? 'Select a run. Refresh to see the latest capture.' : 'No runs visible in this scope. Capture may not yet be registered.';
+            for (const run of result.runs) button(`${run.started_at} · ${run.capture_status} · ${run.run_id}`, () => showRun(run.run_id), runs);
+            if (result.next_offset !== null) button('Older runs', () => loadRuns(result.next_offset), runs);
+        } catch {
+            if (current(requestGeneration)) status.textContent = 'Run history unavailable. Refresh to retry.';
+        }
+    }
+    refresh.addEventListener('click', () => loadRuns());
+    await loadRuns();
+}
diff --git a/tests/backend/test_codex_von_worker.py b/tests/backend/test_codex_von_worker.py
index bd40457f9..8ae841c47 100644
--- a/tests/backend/test_codex_von_worker.py
+++ b/tests/backend/test_codex_von_worker.py
@@ -18,6 +18,13 @@
 spec.loader.exec_module(worker)
 
 
+@pytest.fixture(autouse=True)
+def isolated_archive_boundary(monkeypatch):
+    # Existing controller tests isolate publication; canonical archive capture
+    # and the real launch path are exercised in test_task_run_archives.py.
+    monkeypatch.setattr(worker, "archive_checkpoint", lambda *a, **kw: True)
+
+
 @pytest.fixture
 def config(tmp_path):
     return {
diff --git a/tests/backend/test_task_run_archives.py b/tests/backend/test_task_run_archives.py
new file mode 100644
index 000000000..7fc066c06
--- /dev/null
+++ b/tests/backend/test_task_run_archives.py
@@ -0,0 +1,529 @@
+"""Exact controller fixture, durable byte read-back, recovery and private reads."""
+
+import io
+import json
+import zipfile
+from pathlib import Path
+
+import mongomock
+import pytest
+from flask import Flask
+
+from scripts import codex_von_activity as capture
+from scripts import codex_von_worker as worker
+from src.backend.services import task_run_archive_service as archives
+
+
+@pytest.fixture
+def fixture(tmp_path, monkeypatch):
+    from src.backend.services import task_management_service as tasks
+    from src.backend.services.blob_store import LocalBlobStore
+
+    db = mongomock.MongoClient().db
+    monkeypatch.setattr(archives, "get_db", lambda: db)
+    actor = {"id": "#V#worker", "org": "#V#org", "source": "trusted_local_operator"}
+    monkeypatch.setattr(
+        archives,
+        "get_effective_user_concept_id_with_source",
+        lambda: (actor["id"], actor["source"]),
+    )
+    monkeypatch.setattr(
+        archives, "get_effective_organisation_concept_id", lambda: actor["org"]
+    )
+    monkeypatch.setattr(
+        archives,
+        "resolve_user_organisation_membership",
+        lambda user, org: org == "#V#org",
+    )
+    task = {
+        "task_concept_id": "#V#task",
+        "assignee_concept_id": "#V#worker",
+        "created_by_concept_id": "#V#owner",
+        "organisation_concept_id": "#V#org",
+    }
+    monkeypatch.setattr(tasks, "get_task", lambda _: task)
+    store = LocalBlobStore(tmp_path / "blobs")
+    monkeypatch.setattr(archives, "get_blob_store_from_env", lambda: store)
+    monkeypatch.setattr(
+        archives, "get_blob_store_for_backend_from_env", lambda _: store
+    )
+    attachments = {}
+
+    def add_attachment(task_id, **fields):
+        key = fields["source"]["external_id"]
+        attachments.setdefault(
+            key, {"attachment_id": key, "task_id": task_id, **fields}
+        )
+        return attachments[key]
+
+    monkeypatch.setattr(tasks, "add_task_attachment", add_attachment)
+    run_id = "a" * 32
+    run_dir = tmp_path / "runs" / run_id
+    run_dir.mkdir(parents=True)
+    context = {
+        "task_id": "#V#task",
+        "worker_identity": "#V#worker",
+        "conversation": {"available": False},
+        "execution_settings": {
+            "model": "gpt-6-astra",
+            "reasoning_effort": "high",
+            "model_source": "task",
+            "reasoning_effort_source": "task",
+        },
+    }
+    (run_dir / "context.json").write_text(json.dumps(context))
+    (run_dir / "result.json").write_text('{"status":"completed"}')
+    (run_dir / "exit.json").write_text('{"returncode":0}')
+    state = {
+        "task_id": "#V#task",
+        "attempt": run_id,
+        "run_dir": str(run_dir),
+        "result": {"status": "completed"},
+    }
+    return state, actor, attachments, db, store
+
+
+def test_partial_line_large_output_and_restart_archive(fixture):
+    state, actor, attachments, _db, _store = fixture
+    path = Path(state["run_dir"]) / "events.jsonl"
+    first = b'{"type":"thread.started","thread_id":"fixture-thread"}\n'
+    path.write_bytes(first + b'{"type":"item.completed","item":')
+    capture.checkpoint({}, state)
+    live = archives.read_activity(state["attempt"])
+    assert live["next_cursor"] == len(first)
+    assert live["run"]["capture_status"] == "running"
+    assert live["run"]["thread_id"] == "fixture-thread"
+    large = {
+        "type": "command_execution",
+        "command": "fixture",
+        "aggregated_output": "x" * 1200000,
+        "exit_code": 0,
+    }
+    with path.open("ab") as stream:
+        stream.write(json.dumps(large).encode() + b"}\n")
+        stream.write(
+            b'{"type":"item.completed","item":{"type":"reasoning","summary":"Available summary"}}\n'
+        )
+    # Drop all local capture metadata: service cursor is authoritative.
+    state.pop("capture")
+    capture.checkpoint({}, state, terminal=True)
+    assert state["capture"]["status"] == "complete"
+    data = archives.download_archive(state["attempt"], task_id=state["task_id"])
+    with zipfile.ZipFile(io.BytesIO(data)) as package:
+        manifest = json.loads(package.read("manifest.json"))
+        assert manifest["final_cursor"] == path.stat().st_size
+        assert manifest["event_count"] == 3
+        assert manifest["reasoning_summaries"]["exposed_records"] == 1
+        assert manifest["source_complete"]
+        assert len(package.read("events.jsonl")) > 1200000
+    capture.checkpoint({}, state, terminal=True)
+    assert len(attachments) == 1
+    actor["id"] = "#V#owner"
+    assert archives.download_archive(state["attempt"], task_id=state["task_id"]) == data
+    actor["id"] = "#V#unrelated"
+    assert archives.list_runs(state["task_id"])["runs"] == []
+    with pytest.raises(PermissionError):
+        archives.download_archive(state["attempt"], task_id=state["task_id"])
+
+
+def test_credentials_opaque_state_and_inert_hostile_history(fixture, monkeypatch):
+    state, *_ = fixture
+    monkeypatch.setenv("CONTROLLER_ACCESS_TOKEN", "actual-sentinel-credential")
+    event = {
+        "type": "item.completed",
+        "item": {
+            "type": "reasoning",
+            "summary": [],
+            "encrypted_content": "opaque-test-only",
+        },
+        "password": "secret-in-json",
+        "tool_output": "<script>alert(1)</script> actual-sentinel-credential ghp_abcdefghijklmnop",
+    }
+    path = Path(state["run_dir"]) / "events.jsonl"
+    path.write_text(json.dumps(event) + "\n")
+    capture.checkpoint({}, state, terminal=True)
+    data = archives.download_archive(state["attempt"], task_id=state["task_id"])
+    with zipfile.ZipFile(io.BytesIO(data)) as package:
+        events = package.read("events.jsonl").decode()
+        assert all(
+            secret not in events
+            for secret in [
+                "secret-in-json",
+                "opaque-test-only",
+                "actual-sentinel-credential",
+                "ghp_abcdefghijklmnop",
+            ]
+        )
+        assert "<script>" in events  # inert text, not lost source content
+        manifest = json.loads(package.read("manifest.json"))
+        assert manifest["redactions"]["opaque_reasoning_omitted"] == 1
+        assert manifest["reasoning_summaries"]["exposed_records"] == 0
+    assert "actual-sentinel-credential" in path.read_text()
+
+
+def test_failed_finalisation_retries_without_duplicate_attachment(fixture, monkeypatch):
+    state, _, attachments, db, _ = fixture
+    (Path(state["run_dir"]) / "events.jsonl").write_bytes(
+        b'{"type":"turn.completed"}\n'
+    )
+    original = archives.collection
+    fail_once = {"yes": True}
+
+    class FailFinalUpdate:
+        def __getattr__(self, name):
+            return getattr(original(), name)
+
+        def update_one(self, query, update, **kwargs):
+            if (
+                update.get("$set", {}).get("capture_status") == "complete"
+                and fail_once["yes"]
+            ):
+                fail_once["yes"] = False
+                raise OSError("simulated database outage after attachment")
+            return original().update_one(query, update, **kwargs)
+
+    monkeypatch.setattr(archives, "collection", lambda: FailFinalUpdate())
+    assert not capture.try_checkpoint({}, state, terminal=True)
+    assert state["capture"]["status"] == "pending"
+    assert len(attachments) == 1
+    assert capture.try_checkpoint({}, state, terminal=True)
+    assert len(attachments) == 1
+    assert db.task_run_archives.find_one()["capture_status"] == "complete"
+
+
+def test_changed_source_and_mismatched_task_never_complete(fixture):
+    state, *_ = fixture
+    path = Path(state["run_dir"]) / "events.jsonl"
+    path.write_bytes(b'{"type":"turn.started"}\n')
+    capture.checkpoint({}, state)
+    path.write_bytes(b'{"type":"turn.changed"}\n')
+    assert not capture.try_checkpoint({}, state, terminal=True)
+    assert state["capture"]["status"] == "pending"
+    with pytest.raises(PermissionError):
+        archives.read_activity(state["attempt"], task_id="#V#other")
+    with pytest.raises(ValueError):
+        archives.register_run(
+            task_id="#V#other", run_id=state["attempt"], provenance={}
+        )
+
+
+def test_interrupted_tail_and_missing_exit_are_accounted(fixture):
+    state, *_ = fixture
+    (Path(state["run_dir"]) / "exit.json").unlink()
+    (Path(state["run_dir"]) / "events.jsonl").write_bytes(
+        b'{"type":"turn.started"}\n{"broken":'
+    )
+    capture.checkpoint({}, state, terminal=True)
+    result = archives.read_activity(state["attempt"])["run"]["manifest"]
+    assert not result["source_complete"]
+    assert result["missing_files"] == ["exit.json"]
+    assert result["invalid_records"] == 1
+    assert result["event_count"] == 2
+    state.pop("capture")
+    capture.checkpoint({}, state, terminal=True)
+    assert state["capture"]["source_complete"] is False
+
+
+def test_source_conversation_revocation_and_org_scope(fixture, monkeypatch):
+    from src.backend.services import shared_conversation_service as shares
+
+    state, actor, *_ = fixture
+    context_path = Path(state["run_dir"]) / "context.json"
+    context = json.loads(context_path.read_text())
+    context["conversation"] = {"available": True, "session_id": "private-source"}
+    context_path.write_text(json.dumps(context))
+    monkeypatch.setattr(shares, "resolve_conversation_owner", lambda **kw: "#V#owner")
+    invited = {"yes": True}
+    monkeypatch.setattr(
+        shares,
+        "get_accepted_invite_for_user_session",
+        lambda **kw: {"organisation_concept_id": "#V#org"} if invited["yes"] else None,
+    )
+    (Path(state["run_dir"]) / "events.jsonl").write_bytes(b"{}\n")
+    capture.checkpoint({}, state, terminal=True)
+    invited["yes"] = False
+    with pytest.raises(PermissionError):
+        archives.read_activity(state["attempt"])
+    actor["id"] = "#V#owner"
+    assert archives.read_activity(state["attempt"])["records"]
+    actor["org"] = "#V#other"
+    with pytest.raises(PermissionError):
+        archives.read_activity(state["attempt"])
+
+
+def test_new_read_routes_deny_forged_actor_and_download_scope(fixture, monkeypatch):
+    from src.backend.server.routes import task_routes as routes
+
+    state, actor, *_ = fixture
+    (Path(state["run_dir"]) / "events.jsonl").write_bytes(b"{}\n")
+    capture.checkpoint({}, state, terminal=True)
+    monkeypatch.setattr(routes, "_get_current_user_concept_id", lambda: actor["id"])
+    monkeypatch.setattr(routes, "_get_current_org_concept_id", lambda: actor["org"])
+    app = Flask(__name__)
+    app.register_blueprint(routes.task_bp, url_prefix="/api/tasks")
+    client = app.test_client()
+    base = "/api/tasks/%23V%23task/runs/" + state["attempt"]
+    assert client.get(base).status_code == 200
+    assert (
+        client.get(base + "/download").headers["Cache-Control"] == "private, no-store"
+    )
+    actor["id"] = "#V#unrelated"
+    assert (
+        client.get(
+            base + "/download", headers={"X-User-Concept-ID": "#V#owner"}
+        ).status_code
+        == 403
+    )
+    actor["id"] = "#V#owner"
+    actor["source"] = archives.LEGACY_IDENTITY_HEADER_ACTOR_SOURCE
+    assert client.get(base).status_code == 403
+
+
+def test_hash_readback_detects_storage_corruption(fixture):
+    state, _, _, db, store = fixture
+    (Path(state["run_dir"]) / "events.jsonl").write_bytes(b"{}\n")
+    capture.checkpoint({}, state, terminal=True)
+    receipt = db.task_run_archives.find_one()["archive"]
+    store.put_bytes(receipt["key"], b"corrupt")
+    with pytest.raises(RuntimeError):
+        archives.download_archive(state["attempt"], task_id=state["task_id"])
+
+
+def test_exact_worker_launch_publishes_before_exit_and_downloads_after(
+    fixture, tmp_path, monkeypatch
+):
+    import sys
+
+    from src.backend.server.routes import task_routes as routes
+
+    _, actor, attachments, _db, _ = fixture
+    monkeypatch.setattr(worker, "ACTIVITY_CHECKPOINT_SECONDS", 0.05)
+    executable = tmp_path / "exec-fixture"
+    executable.write_text(f"#!{sys.executable}\n" + """import json, sys, time
+from pathlib import Path
+sys.stdin.read()
+result = Path(sys.argv[sys.argv.index('--output-last-message') + 1])
+print(json.dumps({'type':'thread.started','thread_id':'exact-worker-fixture'}), flush=True)
+print(json.dumps({'type':'item.completed','item':{'type':'command_execution','command':'fixture-test','aggregated_output':'Passed','exit_code':0}}), flush=True)
+for _ in range(500):
+    if (result.parent / 'capture-seen').exists():
+        break
+    time.sleep(0.01)
+else:
+    raise RuntimeError('No live canonical activity was observed')
+result.write_text(json.dumps({'status':'completed','summary':'Fixture complete','evidence':'Live capture observed','question':'','deploy_commit':''}))
+print(json.dumps({'type':'turn.completed','usage':{'input_tokens':0,'output_tokens':0}}), flush=True)
+""")
+    executable.chmod(0o700)
+    checkout = tmp_path / "checkout"
+    checkout.mkdir()
+    state = {"task_id": "#V#task", "checkout_prepared": True, "worktree": str(checkout)}
+    config = {
+        "state_root": str(tmp_path),
+        "agent_id": "#V#worker",
+        "codex_command": str(executable),
+    }
+    monkeypatch.setattr(routes, "_get_current_user_concept_id", lambda: actor["id"])
+    monkeypatch.setattr(routes, "_get_current_org_concept_id", lambda: actor["org"])
+    app = Flask(__name__)
+    app.register_blueprint(routes.task_bp, url_prefix="/api/tasks")
+    client = app.test_client()
+    original = worker.archive_checkpoint
+    before_exit = []
+
+    def observe(config, state, **kwargs):
+        success = original(config, state, **kwargs)
+        response = client.get("/api/tasks/%23V%23task/runs/" + state["attempt"])
+        body = response.get_json()
+        if response.status_code == 200 and body["records"]:
+            assert not (Path(state["run_dir"]) / "exit.json").exists()
+            before_exit.append(body["run"]["event_count"])
+            (Path(state["run_dir"]) / "capture-seen").touch()
+        return success
+
+    monkeypatch.setattr(worker, "archive_checkpoint", observe)
+    state_path = tmp_path / "checkpoint.json"
+    with (tmp_path / "worker.lock").open("w") as lock:
+        worker.launch(
+            config,
+            state,
+            {"requested_model": "gpt-6-astra", "requested_reasoning_effort": "high"},
+            {"available": False},
+            state_path,
+            lock.fileno(),
+        )
+    assert before_exit and before_exit[0] == 2
+    assert state["result"]["status"] == "completed"
+    capture.checkpoint(config, state, terminal=True)
+    actor["id"] = "#V#owner"
+    response = client.get(state["capture"]["archive_url"])
+    assert response.status_code == 200
+    with zipfile.ZipFile(io.BytesIO(response.data)) as package:
+        manifest = json.loads(package.read("manifest.json"))
+        assert manifest["event_count"] == 3
+        assert manifest["thread_id"] == "exact-worker-fixture"
+        assert manifest["source_complete"]
+        assert (
+            manifest["provenance"]["execution_settings"]["reasoning_effort_source"]
+            == "task"
+        )
+    assert len(attachments) == 1
+
+
+def test_pending_capture_keeps_coding_outcome_and_retries_reporting(
+    tmp_path, monkeypatch
+):
+    from types import SimpleNamespace
+
+    state = {
+        "task_id": "#V#task",
+        "attempt": "b" * 32,
+        "result": {
+            "status": "completed",
+            "summary": "Code finished",
+            "evidence": "Tests passed",
+            "question": "",
+        },
+    }
+    sent, finished = [], []
+    api = SimpleNamespace(
+        send=lambda *args: sent.append(args),
+        finish=lambda state: finished.append(state["result"]),
+    )
+    monkeypatch.setattr(worker, "archive_checkpoint", lambda *a, **kw: False)
+    path = tmp_path / "state.json"
+    worker.finish_captured({}, api, state, path)
+    worker.finish_captured({}, api, state, path)
+    assert state["phase"] == "archive_pending"
+    assert state["result"]["status"] == "completed"
+    assert len(sent) == 1 and not finished
+    monkeypatch.setattr(worker, "archive_checkpoint", lambda *a, **kw: True)
+    worker.finish_captured({}, api, state, path)
+    assert len(finished) == 1
+
+
+def test_bounded_backfill_is_idempotent_and_never_launches(fixture, monkeypatch):
+    from types import SimpleNamespace
+
+    state, _, attachments, *_ = fixture
+    run_dir = Path(state["run_dir"])
+    (run_dir / "events.jsonl").write_bytes(b"{}\n")
+    (run_dir / "result.json").write_text(
+        json.dumps(
+            {
+                "status": "completed",
+                "summary": "Done",
+                "evidence": "Passed",
+                "question": "",
+            }
+        )
+    )
+    config = {
+        "state_root": str(run_dir.parents[1]),
+        "agent_id": "#V#worker",
+        "delegator_id": "#V#owner",
+        "organisation_id": "#V#org",
+    }
+    task = {
+        "task_concept_id": "#V#task",
+        "assignee_concept_id": "#V#worker",
+        "created_by_concept_id": "#V#owner",
+        "organisation_concept_id": "#V#org",
+    }
+    api = SimpleNamespace(task=lambda _: task, native_writer=lambda _: True)
+    monkeypatch.setattr(
+        worker, "launch", lambda *a: pytest.fail("Backfill must not launch")
+    )
+    first = worker.backfill_run(config, api, state["task_id"] + "=" + state["attempt"])
+    second = worker.backfill_run(config, api, state["task_id"] + "=" + state["attempt"])
+    assert first["capture"]["archive_url"] == second["capture"]["archive_url"]
+    assert len(attachments) == 1
+    with pytest.raises(ValueError):
+        worker.backfill_run(config, api, "#V#other=" + state["attempt"])
+
+
+def test_reassigned_task_keeps_private_run_without_new_attachment(fixture, monkeypatch):
+    from src.backend.services import task_management_service as tasks
+
+    state, _, attachments, *_ = fixture
+    (Path(state["run_dir"]) / "events.jsonl").write_bytes(b"{}\n")
+    capture.checkpoint({}, state)
+    monkeypatch.setattr(
+        tasks, "get_task", lambda _: {"assignee_concept_id": "#V#replacement"}
+    )
+    capture.checkpoint({}, state, terminal=True)
+    assert not attachments
+    run = archives.read_activity(state["attempt"])["run"]
+    assert run["capture_status"] == "complete"
+    assert run["attachment_status"] == "task_changed_or_unavailable"
+    assert archives.download_archive(state["attempt"], task_id=state["task_id"])
+
+
+def test_batching_does_not_write_per_event(fixture, monkeypatch):
+    state, _, _, db, _ = fixture
+    (Path(state["run_dir"]) / "events.jsonl").write_bytes(
+        b'{"type":"item.completed"}\n' * 1000
+    )
+    coll = db.task_run_archives
+    writes = []
+    original = coll.update_one
+
+    def counted(*args, **kwargs):
+        writes.append(args)
+        return original(*args, **kwargs)
+
+    monkeypatch.setattr(coll, "update_one", counted)
+    capture.checkpoint({}, state, terminal=True)
+    row = coll.find_one()
+    assert row["event_count"] == 1000
+    assert len(row["batches"]) == 1
+    assert len(writes) == 3  # registration, one batch, verified final reference
+
+
+def test_conflicting_batch_retry_is_rejected_without_advancing(fixture):
+    state, *_ = fixture
+    (Path(state["run_dir"]) / "events.jsonl").write_bytes(b"{}\n")
+    capture.checkpoint({}, state)
+    with pytest.raises(ValueError, match="Conflicting"):
+        archives.append_batch(
+            state["attempt"],
+            start=0,
+            end=3,
+            data=b'{"changed":true}\n',
+            source_sha256=archives.digest(b"{}\n"),
+            event_count=1,
+        )
+    assert archives.read_activity(state["attempt"])["run"]["cursor"] == 3
+
+
+def test_real_session_and_trusted_controller_context_resolve_access(
+    fixture, monkeypatch
+):
+    from src.backend.security import access_control as access
+    from src.backend.server.routes import task_routes as routes
+
+    state, *_ = fixture
+    monkeypatch.setattr(
+        archives,
+        "get_effective_user_concept_id_with_source",
+        access.get_effective_user_concept_id_with_source,
+    )
+    monkeypatch.setattr(
+        archives,
+        "get_effective_organisation_concept_id",
+        access.get_effective_organisation_concept_id,
+    )
+    (Path(state["run_dir"]) / "events.jsonl").write_bytes(b"{}\n")
+    with access.override_current_actor("#V#worker", "#V#org"):
+        capture.checkpoint({}, state, terminal=True)
+    app = Flask(__name__)
+    app.secret_key = "local-test-fixture-only"
+    app.register_blueprint(routes.task_bp, url_prefix="/api/tasks")
+    client = app.test_client()
+    with client.session_transaction() as session:
+        session["user_concept_id"] = "#V#owner"
+        session["organisation_concept_id"] = "#V#org"
+    assert client.get(state["capture"]["archive_url"]).status_code == 200
+    with client.session_transaction() as session:
+        session["user_concept_id"] = "#V#unrelated"
+    assert client.get(state["capture"]["archive_url"]).status_code == 403
diff --git a/tests/browser/taskRunActivity.cjs b/tests/browser/taskRunActivity.cjs
new file mode 100644
index 000000000..e2543b2bf
--- /dev/null
+++ b/tests/browser/taskRunActivity.cjs
@@ -0,0 +1,74 @@
+/** Isolated UI fixture; does not authenticate to Von or invoke a model. */
+const { chromium } = require('playwright');
+const http = require('node:http');
+const fs = require('node:fs');
+const path = require('node:path');
+const assert = require('node:assert/strict');
+
+(async () => {
+    const output = path.resolve(process.argv[2] || '.run/task-run-activity');
+    fs.mkdirSync(output, { recursive: true });
+    const moduleSource = fs.readFileSync(path.resolve('src/frontend/web/von_interface/static/js/components/taskRunActivity.js'), 'utf8');
+    const run = { run_id: 'a'.repeat(32), started_at: '2026-09-12T12:00:00Z', updated_at: '2026-09-12T12:00:15Z', capture_status: 'running', event_count: 2, provenance: { execution_settings: { model: 'gpt-6-astra', reasoning_effort: 'high' } } };
+    let terminal = false;
+    let downloadHeader;
+    const server = http.createServer((req, res) => {
+        if (req.url === '/') {
+            res.setHeader('Content-Type', 'text/html');
+            res.end('<!doctype html><html><body><button id="open">Coding run activity</button><script type="module">import {openTaskRunActivity} from "/components/taskRunActivity.js";document.querySelector("#open").onclick=()=>openTaskRunActivity("#V#task");</script></body></html>');
+        } else if (req.url === '/components/taskRunActivity.js') {
+            res.setHeader('Content-Type', 'text/javascript'); res.end(moduleSource);
+        } else if (req.url === '/apiService.js') {
+            res.setHeader('Content-Type', 'text/javascript');
+            res.end('export const WINDOW_SESSION_HEADER="X-Von-Window-Session"; export async function ensureUniqueWindowSessionId(){return "fixture-window";} export function getUserContext(){return {user_id:"fixture",org_id:"fixture-org"};} export async function getJson(url){const r=await fetch(url);if(!r.ok)throw Error();return r.json();}');
+        } else if (req.url.endsWith('/download')) {
+            downloadHeader = req.headers['x-von-window-session'];
+            res.setHeader('Content-Type', 'application/zip'); res.end('fixture-download-bytes');
+        } else if (req.url.startsWith('/api/tasks/')) {
+            res.setHeader('Content-Type', 'application/json');
+            if (!req.url.includes('/' + run.run_id)) res.end(JSON.stringify({ runs: [run], next_offset: null }));
+            else res.end(JSON.stringify({run: {...run, ...(terminal ? {capture_status:'complete',archive_url:'/api/tasks/task/runs/run/download',manifest:{source_complete:true,reasoning_summaries:{availability:'not exposed in captured exec stream'}}} : {})}, records:[{text:'<script>window.archiveExecuted=true</script>\nCommand output\n'+'x'.repeat(5000),display_truncated:true}],next_cursor:123,has_more:false,projection:'Display is limited; full redacted output is archived.'}));
+        } else { res.statusCode = 404; res.end(); }
+    });
+    await new Promise(resolve => server.listen(0, '127.0.0.1', resolve));
+    let browser;
+    try {
+        browser = await chromium.launch({ headless: true });
+        const page = await browser.newPage({ acceptDownloads: true });
+        const errors = [];
+        page.on('pageerror', error => errors.push(error.message));
+        const url = `http://127.0.0.1:${server.address().port}/`;
+        for (const viewport of [{width:1280,height:900},{width:390,height:844}]) {
+            await page.setViewportSize(viewport);
+            await page.goto(url);
+            await page.locator('#open').click();
+            await page.getByRole('button', {name: new RegExp(run.run_id)}).click();
+            await page.getByRole('status').filter({hasText:'running'}).waitFor();
+            assert.equal(await page.evaluate(() => window.archiveExecuted), undefined);
+            const overflow = await page.locator('dialog').evaluate(el => el.scrollWidth > el.clientWidth + 2);
+            assert.equal(overflow, false);
+            await page.screenshot({path:path.join(output, `activity-${viewport.width}.png`)});
+            terminal = true;
+            await page.getByRole('button', {name:'Refresh this page'}).click();
+            await page.getByRole('status').filter({hasText:'complete'}).waitFor();
+            const downloadPromise = page.waitForEvent('download');
+            await page.getByRole('button', {name:'Download archive'}).click();
+            const download = await downloadPromise;
+            await download.saveAs(path.join(output, `download-${viewport.width}.zip`));
+            assert.equal(downloadHeader, 'fixture-window');
+            await page.getByRole('button', {name:'Close',exact:true}).click();
+            await page.locator('dialog').waitFor({state:'detached'});
+            terminal = false;
+        }
+        await page.locator('#open').click();
+        await page.locator('dialog').waitFor();
+        await page.evaluate(() => document.dispatchEvent(new Event('orgSwitched')));
+        await page.locator('dialog').waitFor({state:'detached'});
+        assert.deepEqual(errors, []);
+        fs.writeFileSync(path.join(output,'receipt.json'), JSON.stringify({scope:'isolated UI fixture, no live authentication or model calls',checks:['desktop/mobile no horizontal overflow','source text inert','running/final refresh','window-scoped download','dialog closes and clears on scope change'],passed:true},null,2));
+        console.log('Task run activity browser fixture passed');
+    } finally {
+        if (browser) await browser.close();
+        await new Promise(resolve => server.close(resolve));
+    }
+})().catch(error => {console.error(error);process.exitCode=1;});