diff --git a/src/brigade/aboyeur.py b/src/brigade/aboyeur.py
index 1dd231b5..f03e641c 100644
--- a/src/brigade/aboyeur.py
+++ b/src/brigade/aboyeur.py
@@ -27,6 +27,7 @@
 from . import localio
 from . import proc, receipt_schema, runguard
 from . import run_control
+from . import run_lifecycle
 from .result_integrity import validate_final_output
 from .run_receipts import (
     agent_result_from_worker as _agent_result_from_worker,
@@ -489,6 +490,17 @@ def _write_json(path: Path, payload: object) -> None:
     # run.json is polled by `brigade runs watch/steer/interrupt` while the run
     # rewrites it, so the write must be atomic or a concurrent reader can
     # observe a truncated file.
+    if path.name == "run.json" and isinstance(payload, dict):
+        status = payload.get("status")
+        if isinstance(status, str) and status:
+            run_lifecycle.record_lifecycle_transition(
+                path.parent,
+                status=status,
+                # The receipt payload identifies the lock workspace
+                # (lock_workspace or cwd); the run directory layout does not.
+                workspace=runguard.resolve_run_lock_workspace(payload, path.parent),
+                incoming_snapshot=payload,
+            )
     localio.write_text_atomic(path, json.dumps(payload, indent=2, sort_keys=True) + "\n")
 
 
@@ -1937,7 +1949,7 @@ def record_artifact_collection(
     payload["artifact_collection"] = collection
     try:
         _write_json(run_path, receipt_schema.stamp_run_receipt(payload))
-    except OSError as exc:
+    except (OSError, run_lifecycle.LifecycleJournalError) as exc:
         raise runguard.RetainRunLockError(f"failed to update run receipt after artifact collection: {exc}") from exc
 
 
@@ -2010,7 +2022,7 @@ def record_run_termination(
             )
     try:
         _write_json(run_path, receipt_schema.stamp_run_receipt(payload))
-    except OSError as exc:
+    except (OSError, run_lifecycle.LifecycleJournalError) as exc:
         raise runguard.RetainRunLockError(f"failed to write terminal run receipt: {exc}") from exc
 
 
@@ -2038,7 +2050,7 @@ def record_dispatch_stage(output_dir: Path, *, stage: int, seats: tuple[str, ...
     )
     try:
         _write_json(run_path, receipt_schema.stamp_run_receipt(payload))
-    except OSError as exc:
+    except (OSError, run_lifecycle.LifecycleJournalError) as exc:
         raise runguard.RetainRunLockError(f"failed to write dispatch stage receipt: {exc}") from exc
 
 
@@ -2067,7 +2079,7 @@ def record_result_processing(output_dir: Path, *, seat: str) -> None:
     payload.pop("active_seats", None)
     try:
         _write_json(run_path, receipt_schema.stamp_run_receipt(payload))
-    except OSError as exc:
+    except (OSError, run_lifecycle.LifecycleJournalError) as exc:
         raise runguard.RetainRunLockError(f"failed to record result-processing phase: {exc}") from exc
 
 
@@ -2148,6 +2160,7 @@ def _run_payload(
     include_git: bool = True,
     pre_run_snapshot: dict[str, object] | None = None,
     scheduler: dict[str, object] | None = None,
+    lifecycle_journal_requested: bool | None = None,
 ) -> dict[str, object]:
     payload: dict[str, object] = {
         "schema": receipt_schema.RUN_RECEIPT_SCHEMA,
@@ -2182,6 +2195,8 @@ def _run_payload(
         payload["cwd"] = str(cwd)
     if scheduler is not None:
         payload["scheduler"] = scheduler
+    if lifecycle_journal_requested:
+        payload["lifecycle_journal_requested"] = True
     if resolution := _roster_resolution_payload(roster):
         payload["roster"] = resolution
     if lock_workspace is not None:
@@ -2280,6 +2295,19 @@ def record_run_start(
     output_dir = output_dir.expanduser().resolve()
     started_at = started_at or datetime.now(timezone.utc)
     output_dir.mkdir(parents=True, exist_ok=True)
+    run_json = output_dir / "run.json"
+    run_json_exists = run_json.is_file()
+    existing_requested = False
+    if run_json_exists:
+        try:
+            existing = json.loads(run_json.read_text())
+        except (OSError, json.JSONDecodeError):
+            existing = None
+        if isinstance(existing, dict) and existing.get("lifecycle_journal_requested") is True:
+            existing_requested = True
+    lifecycle_requested = existing_requested or (
+        not run_json_exists and run_lifecycle.is_lifecycle_journaling_enabled()
+    )
     _write_json(
         output_dir / "run.json",
         _run_payload(
@@ -2301,6 +2329,7 @@ def record_run_start(
             scheduler=(
                 {"requested": scheduler, "used": None, "fallback_reason": None} if scheduler is not None else None
             ),
+            lifecycle_journal_requested=True if lifecycle_requested else None,
         ),
     )
     _write_json(output_dir / "roster.json", _roster_payload(roster))
@@ -2488,6 +2517,15 @@ def scheduler_resolved(used: str, fallback_reason: str | None) -> None:
     def _payload(**kwargs: Any) -> dict[str, object]:
         if "skill_route_policy" not in kwargs and skill_policy is not None:
             kwargs["skill_route_policy"] = skill_policy
+        if output_dir is not None and "lifecycle_journal_requested" not in kwargs:
+            run_path = output_dir / "run.json"
+            if run_path.is_file():
+                try:
+                    existing = json.loads(run_path.read_text())
+                except (OSError, json.JSONDecodeError):
+                    existing = None
+                if isinstance(existing, dict) and existing.get("lifecycle_journal_requested") is True:
+                    kwargs["lifecycle_journal_requested"] = True
         return _run_payload(
             lock_workspace=lock_workspace,
             pre_run_snapshot=pre_run_snapshot_payload,
diff --git a/src/brigade/run_lifecycle.py b/src/brigade/run_lifecycle.py
new file mode 100644
index 00000000..70eea95f
--- /dev/null
+++ b/src/brigade/run_lifecycle.py
@@ -0,0 +1,288 @@
+"""Opt-in lifecycle journaling integration for brigade run status transitions.
+
+Appends ``brigade.run_event.v1`` envelopes to the per-run lifecycle journal
+BEFORE each existing ``run.json`` status snapshot refresh, gated by an opt-in
+feature flag (``BRIGADE_LIFECYCLE_JOURNAL``) that is **per-run and durable**.
+
+Activation model
+----------------
+- Flag off: byte-identical behavior, no journal, no event, no request field.
+- New run started with the flag on: the pre-lock bootstrap writes
+  ``lifecycle_journal_requested: true`` into ``run.json`` but never creates
+  ``events/lifecycle.jsonl``. The request field is the durable opt-in marker
+  until journaling activates.
+- Legacy ``run.json`` with no request field stays snapshot-only even if the
+  environment flag is later enabled.
+- Once the matching run lock is held, the first mapped-status write with a
+  pending request creates the journal (0o700 directory, 0o600 file) and
+  appends ``run.created``. Journal existence is the durable activated fact
+  after that point.
+- Once the journal exists, later transitions remain journaled even if a
+  resumed process lacks the environment flag.
+
+Lock ownership
+--------------
+Every append happens under the active run lock. ``record_run_start`` is
+written once before ``cli/run.py`` acquires ``runguard.run_lock`` (the
+pre-lock bootstrap, which records the request but never creates the journal)
+and again inside the held lock through ``aboyeur.run``. Ownership is verified
+with ``runguard.is_active_run_owner`` against the workspace identified by the
+run receipt payload (``lock_workspace`` or ``cwd``, resolved by
+``runguard.resolve_run_lock_workspace``), never derived from the run
+directory layout, so custom ``--output-dir`` runs verify against the lock
+they actually hold. Once journaling is active, a mapped-status write without
+the matching active lock fails closed as ``LifecycleJournalError`` and the
+caller's ``run.json`` snapshot does not advance.
+
+Transition identity
+-------------------
+Events record status *transitions*, not detail refreshes. The prior canonical
+``run.json`` snapshot is read and hashed before appending. When the prior
+status equals the target status and the journal already holds a committed
+event, nothing is appended even if error/detail fields changed. A real
+transition uses an idempotency key derived from a fixed prefix plus the prior
+snapshot digest and target event request: a crash/retry after the journal fsync
+but before the ``run.json`` replacement sees the unchanged prior snapshot,
+derives the same key, and ``append_event`` returns the committed event
+without a second append. A later legitimate recurrence
+(dispatching -> result-processing -> dispatching, including across an
+unmapped intermediate status) has a different prior snapshot digest and
+appends. The journal tail payload is never used to dedupe.
+
+Bounded failure
+---------------
+Any ``run_journal`` typed error surfaces its already-bounded ``diagnostic``;
+raw ``OSError`` and ``run_events.CanonicalizationError`` from snapshot
+reads, ensure/read/append, or key derivation are wrapped as
+``LifecycleJournalError`` with a generic bounded category (the raw text can
+carry a path or private value) and raised before the caller writes
+``run.json``.
+
+Privacy exclusions are structural: payloads are built from the closed per-type
+allowlist only, so no prompts, model output, tool arguments, credentials,
+provider bodies, stack traces, or raw ``run.json`` error strings can enter a
+payload.
+
+Standard library only. Brigade is zero-runtime-dependency.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+
+from brigade import run_events, run_journal, runguard
+
+_FLAG_ENV = "BRIGADE_LIFECYCLE_JOURNAL"
+_TRUTHY = frozenset({"1", "true", "yes", "on"})
+_JOURNAL_NAME = "lifecycle.jsonl"
+_REQUEST_FIELD = "lifecycle_journal_requested"
+_IDEMPOTENCY_PREFIX = "lifecycle"
+_IO_CATEGORY = "lifecycle journal I/O failure"
+_CANONICALIZATION_CATEGORY = "lifecycle journal canonicalization failure"
+
+# Map of current run.json status values to the allowlisted run_event.v1
+# event_type that records that transition. Statuses without a mapping
+# (e.g. "dry-run", "incomplete", "artifact-collection") are skipped: no event
+# is appended and the run.json snapshot refresh proceeds unchanged. This keeps
+# the integration to the current status transitions and the existing event
+# registry -- no new schemas.
+STATUS_EVENT_TYPE: dict[str, str] = {
+    "started": "run.created",
+    "planning": "run.planning.started",
+    "dispatching": "run.dispatch.requested",
+    "result-processing": "run.dispatch.completed",
+    "synthesizing": "run.synthesis.started",
+    "handoff": "run.synthesis.completed",
+    "ok": "run.completed",
+    "failed": "run.failed",
+    "canceled": "run.interrupted",
+    "timeout": "run.failed",
+}
+
+
+class LifecycleJournalError(RuntimeError):
+    """Bounded lifecycle-journal failure; run.json must not advance past it."""
+
+
+def is_lifecycle_journaling_enabled() -> bool:
+    """True when the opt-in lifecycle journal flag is set in the environment."""
+    return os.environ.get(_FLAG_ENV, "").strip().lower() in _TRUTHY
+
+
+def _run_id_from_dir(run_dir: Path) -> str:
+    return run_dir.expanduser().resolve().name
+
+
+def _journal_path(run_dir: Path) -> Path:
+    return run_dir / "events" / _JOURNAL_NAME
+
+
+def _journal_requested(
+    run_dir: Path,
+    *,
+    incoming: Mapping[str, Any] | None,
+) -> bool:
+    """True when this run has durably requested lifecycle journaling."""
+    run_json = run_dir / "run.json"
+    if run_json.is_file():
+        try:
+            meta = json.loads(run_json.read_bytes())
+        except (OSError, ValueError, UnicodeDecodeError):
+            return False
+        return isinstance(meta, dict) and meta.get(_REQUEST_FIELD) is True
+    if incoming is not None and incoming.get(_REQUEST_FIELD) is True:
+        return True
+    return False
+
+
+def _run_snapshot_state(run_dir: Path) -> tuple[str | None, str | None]:
+    """Return (status, sha256 digest) of the current canonical run.json snapshot.
+
+    The digest binds the exact prior snapshot bytes; the status drives
+    transition detection. Returns (None, None) when run.json is absent. A
+    present-but-unparseable snapshot still yields its digest with a None
+    status. Other read failures propagate as OSError so the caller fails
+    closed with a bounded category.
+    """
+    try:
+        raw = (run_dir / "run.json").read_bytes()
+    except FileNotFoundError:
+        return None, None
+    digest = hashlib.sha256(raw).hexdigest()
+    status: str | None = None
+    try:
+        meta = json.loads(raw)
+    except (ValueError, UnicodeDecodeError):
+        meta = None
+    if isinstance(meta, dict):
+        candidate = meta.get("status")
+        if isinstance(candidate, str) and candidate:
+            status = candidate
+    return status, digest
+
+
+def _allowlisted_payload(event_type: str, *, status: str) -> dict[str, Any]:
+    """Build a payload carrying only keys in the closed per-type allowlist."""
+    allowed = run_events.EVENT_TYPES[event_type]
+    payload: dict[str, Any] = {}
+    if "status" in allowed:
+        payload["status"] = status
+    if "detail" in allowed:
+        text = status
+        if len(text) > run_events.MAX_PAYLOAD_STR_LEN:
+            text = text[: run_events.MAX_PAYLOAD_STR_LEN]
+        payload["detail"] = text
+    return payload
+
+
+def record_lifecycle_transition(
+    run_dir: Path,
+    *,
+    status: str,
+    workspace: Path | None = None,
+    incoming_snapshot: Mapping[str, Any] | None = None,
+) -> run_journal.RunEvent | None:
+    """Append one lifecycle event for a run status transition.
+
+    Called BEFORE the existing ``run.json`` snapshot refresh. Returns the
+    appended (or replayed) ``RunEvent`` when journaling is active and the
+    status transition is recorded; ``None`` when the status is unmapped, the
+    run is not activated, journaling is only requested but the matching lock
+    is not yet held, or the call only recorded the per-run request. Raises
+    ``LifecycleJournalError`` on any bounded journal failure, or when
+    journaling is active but the caller does not hold the matching active run
+    lock, so the caller does not write ``run.json`` past an uncommitted or
+    unowned transition.
+
+    ``workspace`` identifies the run lock to verify ownership against; pass
+    the workspace from the run receipt payload (``lock_workspace`` or
+    ``cwd``), as resolved by ``runguard.resolve_run_lock_workspace``.
+    """
+    run_dir = Path(run_dir).expanduser().resolve()
+    event_type = STATUS_EVENT_TYPE.get(status)
+    if event_type is None:
+        # Unmapped statuses are not lifecycle transitions: no event, no
+        # activation, no ownership gate; the snapshot refresh proceeds.
+        return None
+    run_id = _run_id_from_dir(run_dir)
+    if not run_events._RUN_ID_RE.match(run_id):
+        raise LifecycleJournalError(run_events._bound(f"invalid run_id from run_dir: {run_id!r}"))
+    journal_path = _journal_path(run_dir)
+    requested = _journal_requested(run_dir, incoming=incoming_snapshot)
+    if not journal_path.is_file():
+        if not requested:
+            return None
+        if workspace is None or not runguard.is_active_run_owner(workspace, run_dir):
+            return None
+        try:
+            run_journal.ensure_journal(journal_path)
+        except run_journal.RunJournalError as exc:
+            raise LifecycleJournalError(run_events._bound(exc.diagnostic)) from exc
+        except run_events.CanonicalizationError as exc:
+            raise LifecycleJournalError(_CANONICALIZATION_CATEGORY) from exc
+        except OSError as exc:
+            raise LifecycleJournalError(f"{_IO_CATEGORY} ({type(exc).__name__})") from exc
+
+    # The journal exists: journaling is active for this run, with or without
+    # the env flag. Every append must come from the process holding the
+    # matching active run lock; anything else fails closed so run.json never
+    # advances past an uncommitted or unowned transition.
+    if workspace is None or not runguard.is_active_run_owner(workspace, run_dir):
+        raise LifecycleJournalError("lifecycle journal append requires the active run lock for this run")
+
+    try:
+        run_journal.ensure_journal(journal_path)
+        report = run_journal.read_journal(journal_path)
+        if report.partial_tail is not None or report.chain_errors:
+            raise run_journal.ChainIntegrityError(run_events._bound("lifecycle journal is not derivable"))
+        prior_status, prior_digest = _run_snapshot_state(run_dir)
+        payload = _allowlisted_payload(event_type, status=status)
+        # Status transitions, not detail refreshes: a same-status refresh
+        # appends nothing once any event is committed, even when error/detail
+        # fields changed.
+        if prior_status == status and report.events:
+            return report.events[-1]
+        # Real transition: the idempotency key binds the prior snapshot digest
+        # to the target event request. A crash/retry after the journal fsync
+        # but before the run.json replacement sees the unchanged prior
+        # snapshot, derives the same key, and append_event returns the
+        # committed event; a later genuine recurrence has a different prior
+        # snapshot digest and appends. The fixed prefix keeps the key within
+        # the 128-char envelope bound regardless of run directory name length.
+        key_digest = hashlib.sha256(
+            run_events.canonical_bytes(
+                {
+                    "event_type": event_type,
+                    "payload": payload,
+                    "prior_snapshot_digest": prior_digest,
+                }
+            )
+        ).hexdigest()
+        idempotency_key = f"{_IDEMPOTENCY_PREFIX}:{key_digest[:32]}"
+        return run_journal.append_event(
+            journal_path,
+            run_id=run_id,
+            event_type=event_type,
+            payload=payload,
+            idempotency_key=idempotency_key,
+            expected_previous_sequence=report.events[-1].sequence if report.events else 0,
+        )
+    except run_journal.RunJournalError as exc:
+        raise LifecycleJournalError(run_events._bound(exc.diagnostic)) from exc
+    except run_events.CanonicalizationError as exc:
+        raise LifecycleJournalError(_CANONICALIZATION_CATEGORY) from exc
+    except OSError as exc:
+        raise LifecycleJournalError(f"{_IO_CATEGORY} ({type(exc).__name__})") from exc
+
+
+__all__ = [
+    "LifecycleJournalError",
+    "STATUS_EVENT_TYPE",
+    "is_lifecycle_journaling_enabled",
+    "record_lifecycle_transition",
+]
diff --git a/src/brigade/runguard.py b/src/brigade/runguard.py
index ecf67ab4..010e5b3f 100644
--- a/src/brigade/runguard.py
+++ b/src/brigade/runguard.py
@@ -389,6 +389,29 @@ def _owner_matches_run(owner: dict[str, object] | None, run_dir: Path) -> bool:
     return isinstance(recorded, str) and Path(recorded).expanduser().resolve() == run_dir
 
 
+def is_active_run_owner(workspace: Path, run_dir: Path) -> bool:
+    """True when the current process holds the active run lock for ``run_dir``.
+
+    Reads the run-lock owner metadata (``owner.json`` under the run lock for
+    ``workspace``) and matches the current pid and the exact resolved
+    ``run_dir``. Used by lifecycle journaling so appends happen only under the
+    active run lock, never from a pre-lock bootstrap or an unlocked writer.
+    The workspace is supplied by the caller (from the run receipt's
+    ``lock_workspace`` or ``cwd``), never derived from the run directory
+    layout, so a custom ``--output-dir`` run verifies against the same lock
+    the run holds and a mismatched workspace fails closed. Returns False when
+    the lock is absent or unreadable; it never raises.
+    """
+    path = lock_path(workspace)
+    owner = _read_lock_owner(path)
+    if owner is None:
+        return False
+    owner_pid = owner.get("pid")
+    if not isinstance(owner_pid, int) or owner_pid != os.getpid():
+        return False
+    return _owner_matches_run(owner, run_dir.expanduser().resolve())
+
+
 def _quarantine_unattributable(path: Path, claimed: Path) -> None:
     quarantined = path.with_name(f".{path.name}.{uuid4().hex}.orphaned")
     try:
diff --git a/tests/test_run_detach.py b/tests/test_run_detach.py
index e927d6c2..1f637545 100644
--- a/tests/test_run_detach.py
+++ b/tests/test_run_detach.py
@@ -9,8 +9,9 @@
 
 import pytest
 
-from brigade import aboyeur, cli, proc
+from brigade import aboyeur, cli, proc, run_journal
 from brigade import roster as roster_mod
+from brigade import runguard
 from brigade.cli import run as run_cli
 
 
@@ -500,3 +501,102 @@ def test_run_detach_child_argv_preserves_worker(tmp_path):
     assert argv[argv.index("--wait") + 1] == "2.5"
     assert argv[argv.index("--resolved-roster-source") + 1] == "workspace"
     assert argv[argv.index("--resolved-roster-shadowed") + 1] == str(roster_resolution.shadowed[0])
+
+
+def _minimal_roster() -> roster_mod.Roster:
+    return roster_mod.Roster(
+        orchestrator="chef",
+        agents={"chef": roster_mod.Agent("chef", "codex", "plan")},
+    )
+
+
+def _journal_path(run_dir: Path) -> Path:
+    return run_dir / "events" / "lifecycle.jsonl"
+
+
+@pytest.fixture
+def lifecycle_enabled(monkeypatch):
+    monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1")
+    yield
+    monkeypatch.delenv("BRIGADE_LIFECYCLE_JOURNAL", raising=False)
+
+
+def test_detach_lifecycle_parent_child_pre_lock_writes_request_not_journal(lifecycle_enabled, tmp_path):
+    """Detached parent/child split: request before lock, journal only after lock."""
+    repo = tmp_path / "workspace"
+    repo.mkdir()
+    _write_roster(repo)
+    roster = _minimal_roster()
+    run_dir = repo / ".brigade" / "runs" / "20260728-170000-detach01"
+    run_dir.mkdir(parents=True)
+
+    # Detached parent pre-lock bootstrap.
+    aboyeur.record_run_start(
+        run_dir,
+        task="do work",
+        cwd=repo,
+        roster=roster,
+        read_only=False,
+        lock_workspace=repo,
+    )
+    parent_meta = json.loads((run_dir / "run.json").read_text())
+    assert parent_meta["lifecycle_journal_requested"] is True
+    assert not (run_dir / "events").exists()
+
+    # Detached child pre-lock bootstrap before runguard.run_lock.
+    aboyeur.record_run_start(
+        run_dir,
+        task="do work",
+        cwd=repo,
+        roster=roster,
+        read_only=False,
+        lock_workspace=repo,
+    )
+    child_pre_meta = json.loads((run_dir / "run.json").read_text())
+    assert child_pre_meta["lifecycle_journal_requested"] is True
+    assert not (run_dir / "events").exists()
+
+    # Child in-lock start activates journaling and appends run.created.
+    with runguard.run_lock(repo, run_dir=run_dir):
+        aboyeur.record_run_start(
+            run_dir,
+            task="do work",
+            cwd=repo,
+            roster=roster,
+            read_only=False,
+            lock_workspace=repo,
+        )
+
+    assert _journal_path(run_dir).is_file()
+    journal_events = run_journal.read_journal(_journal_path(run_dir)).events
+    assert len(journal_events) == 1
+    assert journal_events[0].event_type == "run.created"
+
+
+def test_detach_lifecycle_pre_lock_sigterm_writes_terminal_without_journal(lifecycle_enabled, tmp_path, monkeypatch):
+    _write_roster(tmp_path)
+    real_write_json = aboyeur._write_json
+
+    def terminate_during_roster_write(path, payload):
+        if path.name == "roster.json":
+            signal.raise_signal(signal.SIGTERM)
+        return real_write_json(path, payload)
+
+    monkeypatch.setattr(aboyeur, "_write_json", terminate_during_roster_write)
+
+    with pytest.raises(SystemExit) as exc_info:
+        cli.main(["run", "do work", "--cwd", str(tmp_path), "--detach", "--worker", "coder"])
+
+    assert exc_info.value.code == 128 + signal.SIGTERM
+    run_dir = next((tmp_path / ".brigade" / "runs").iterdir())
+    receipt = json.loads((run_dir / "run.json").read_text())
+    assert receipt["schema"] == "brigade.run.v1"
+    assert receipt["status"] == "canceled"
+    assert receipt["lifecycle_journal_requested"] is True
+    assert receipt["failure"] == {
+        "phase": "startup",
+        "kind": "signal",
+        "detail": "run terminated by SIGTERM",
+        "seat": "coder",
+    }
+    assert not (run_dir / "events").exists()
diff --git a/tests/test_run_lifecycle.py b/tests/test_run_lifecycle.py
new file mode 100644
index 00000000..073c2a83
--- /dev/null
+++ b/tests/test_run_lifecycle.py
@@ -0,0 +1,562 @@
+"""Regression tests for opt-in lifecycle journaling (issue #568 slice 2).
+
+Covers the corrected lifecycle-journaling contract: the per-run request field
+is the durable opt-in marker until journaling activates under the run lock,
+active run-lock ownership is verified against the workspace recorded in the run
+receipt (custom ``--output-dir`` layouts included), an unlocked writer on an
+active journal fails closed with ``run.json`` unchanged, status transitions are
+recorded (detail refreshes are not), transition identity is keyed on the
+prior snapshot digest so crash/retry replays the committed event while a
+later recurrence appends, lifecycle payloads never carry raw run.json error
+strings, and raw OSError/CanonicalizationError surface only as bounded generic
+categories that block the snapshot. Every journaling append runs under the real
+``runguard.run_lock`` context.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from brigade import aboyeur, localio, proc, run_events, run_journal, run_lifecycle, runguard
+from brigade import roster as roster_mod
+
+_REQUEST_FIELD = "lifecycle_journal_requested"
+
+
+def _git(repo: Path, *args: str) -> proc.Result:
+    result = proc.run(["git", *args], cwd=repo)
+    assert result.code == 0, result.stderr
+    return result
+
+
+def _repo(tmp_path: Path) -> Path:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    _git(repo, "init")
+    _git(repo, "config", "user.email", "test@example.invalid")
+    _git(repo, "config", "user.name", "Test User")
+    (repo / "tracked.txt").write_text("base\n")
+    _git(repo, "add", "tracked.txt")
+    _git(repo, "commit", "-m", "initial")
+    return repo
+
+
+_RUN_ID = "20260728-160000-abcd1234"
+
+
+def _run_dir(repo: Path, run_id: str = _RUN_ID) -> Path:
+    run_dir = repo / ".brigade" / "runs" / run_id
+    run_dir.mkdir(parents=True)
+    return run_dir
+
+
+def _journal_path(run_dir: Path) -> Path:
+    return run_dir / "events" / "lifecycle.jsonl"
+
+
+def _minimal_roster() -> roster_mod.Roster:
+    return roster_mod.Roster(
+        orchestrator="chef",
+        agents={"chef": roster_mod.Agent("chef", "codex", "plan")},
+    )
+
+
+def _run_payload(status: str, *, error: str | None = None, lock_workspace: Path | None = None) -> dict[str, object]:
+    payload: dict[str, object] = {"schema": "brigade.run.v1", "status": status}
+    if error is not None:
+        payload["error"] = error
+    if lock_workspace is not None:
+        payload["lock_workspace"] = str(lock_workspace)
+    return payload
+
+
+def _apply_lifecycle_request(run_dir: Path, payload: dict[str, object]) -> dict[str, object]:
+    run_json = run_dir / "run.json"
+    if run_json.is_file():
+        existing = json.loads(run_json.read_text())
+        if existing.get(_REQUEST_FIELD) is True:
+            payload[_REQUEST_FIELD] = True
+    elif run_lifecycle.is_lifecycle_journaling_enabled():
+        payload[_REQUEST_FIELD] = True
+    return payload
+
+
+def _write_run_json(
+    run_dir: Path,
+    status: str,
+    *,
+    error: str | None = None,
+    lock_workspace: Path | None = None,
+) -> None:
+    """Write run.json the way aboyeur does (lifecycle append then atomic snapshot)."""
+    payload = _apply_lifecycle_request(
+        run_dir,
+        _run_payload(status, error=error, lock_workspace=lock_workspace),
+    )
+    aboyeur._write_json(run_dir / "run.json", payload)
+
+
+def test_record_run_start_does_not_opt_in_existing_legacy_run(tmp_path, monkeypatch):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+    localio.write_json(run_dir / "run.json", _run_payload("started", lock_workspace=repo))
+    monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1")
+
+    aboyeur.record_run_start(
+        run_dir,
+        task="legacy run",
+        cwd=repo,
+        roster=_minimal_roster(),
+        read_only=False,
+        lock_workspace=repo,
+    )
+
+    receipt = json.loads((run_dir / "run.json").read_text())
+    assert _REQUEST_FIELD not in receipt
+    assert not _journal_path(run_dir).exists()
+
+
+def _write_run_json_locked(
+    repo: Path,
+    run_dir: Path,
+    status: str,
+    *,
+    error: str | None = None,
+    lock_workspace: Path | None = None,
+) -> None:
+    with runguard.run_lock(repo, run_dir=run_dir):
+        _write_run_json(run_dir, status, error=error, lock_workspace=lock_workspace)
+
+
+def _events(run_dir: Path) -> list[run_journal.RunEvent]:
+    return run_journal.read_journal(_journal_path(run_dir)).events
+
+
+@pytest.fixture
+def enabled(monkeypatch):
+    monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1")
+    yield
+    monkeypatch.delenv("BRIGADE_LIFECYCLE_JOURNAL", raising=False)
+
+
+@pytest.fixture
+def disabled(monkeypatch):
+    monkeypatch.delenv("BRIGADE_LIFECYCLE_JOURNAL", raising=False)
+
+
+def test_flag_off_is_byte_compatible_and_creates_no_journal(disabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json_locked(repo, run_dir, "started")
+
+    assert (run_dir / "run.json").is_file()
+    assert not (run_dir / "events").exists()
+    meta = json.loads((run_dir / "run.json").read_text())
+    assert meta["status"] == "started"
+    assert _REQUEST_FIELD not in meta
+
+
+def test_pre_lock_new_run_bootstrap_writes_request_without_journal(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    # The pre-lock bootstrap writes run.json before runguard.run_lock is held:
+    # it records the durable request but must not create the journal.
+    _write_run_json(run_dir, "started")
+
+    assert (run_dir / "run.json").is_file()
+    meta = json.loads((run_dir / "run.json").read_text())
+    assert meta[_REQUEST_FIELD] is True
+    assert not (run_dir / "events").exists()
+
+
+def test_no_journal_file_until_lock_held(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+
+    assert _journal_path(run_dir).is_file()
+    assert sorted(path.name for path in (run_dir / "events").iterdir()) == ["lifecycle.jsonl"]
+
+
+def test_in_lock_write_appends_run_created(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json(run_dir, "started")  # pre-lock bootstrap: request only
+    _write_run_json_locked(repo, run_dir, "started")  # in-lock: append run.created
+
+    events = _events(run_dir)
+    assert len(events) == 1
+    assert events[0].event_type == "run.created"
+    assert events[0].payload == {"status": "started"}
+    assert events[0].sequence == 1
+    assert events[0].previous_digest is None
+
+
+def test_recording_without_matching_lock_fails_closed(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json(run_dir, "started")  # pre-lock bootstrap: request only
+    _write_run_json_locked(repo, run_dir, "started")  # in-lock: run.created
+    run_before = (run_dir / "run.json").read_bytes()
+    journal_before = _journal_path(run_dir).read_bytes()
+
+    # Activated run, but no lock held: the write fails closed and neither the
+    # journal nor the run.json snapshot advances.
+    with pytest.raises(run_lifecycle.LifecycleJournalError):
+        _write_run_json(run_dir, "planning")
+
+    assert (run_dir / "run.json").read_bytes() == run_before
+    assert _journal_path(run_dir).read_bytes() == journal_before
+    assert [e.event_type for e in _events(run_dir)] == ["run.created"]
+
+
+def test_custom_output_dir_appends_under_matching_lock(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    # A custom --output-dir layout: not under <workspace>/.brigade/runs.
+    run_dir = tmp_path / "custom-output" / _RUN_ID
+    run_dir.mkdir(parents=True)
+
+    # The receipt payload carries lock_workspace, exactly as record_run_start
+    # writes it; lock ownership resolves from that, not from the run layout.
+    _write_run_json(run_dir, "started", lock_workspace=repo)  # pre-lock bootstrap
+    _write_run_json_locked(repo, run_dir, "started", lock_workspace=repo)
+
+    events = _events(run_dir)
+    assert [e.event_type for e in events] == ["run.created"]
+
+
+def test_long_custom_output_dir_final_component_journals(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    long_run_id = "x" * 200
+    run_dir = tmp_path / "custom-output" / long_run_id
+    run_dir.mkdir(parents=True)
+
+    _write_run_json(run_dir, "started", lock_workspace=repo)
+    _write_run_json_locked(repo, run_dir, "started", lock_workspace=repo)
+
+    events = _events(run_dir)
+    assert len(events) == 1
+    assert events[0].event_type == "run.created"
+    assert len(events[0].idempotency_key) <= run_events.MAX_IDEMPOTENCY_KEY_LEN
+    assert events[0].idempotency_key.startswith("lifecycle:")
+
+
+def test_mismatched_workspace_skips_journal_until_correct_lock(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    other = tmp_path / "other-workspace"
+    other.mkdir()
+    run_dir = tmp_path / "custom-output" / _RUN_ID
+    run_dir.mkdir(parents=True)
+
+    _write_run_json(run_dir, "started", lock_workspace=other)  # pre-lock bootstrap
+    assert not (run_dir / "events").exists()
+
+    # The lock is held at `repo` but the receipt claims `other`: journaling is
+    # still only requested, so the write proceeds without creating a journal.
+    with runguard.run_lock(repo, run_dir=run_dir):
+        _write_run_json(run_dir, "started", lock_workspace=other)
+
+    assert not (run_dir / "events").exists()
+    meta = json.loads((run_dir / "run.json").read_text())
+    assert meta["status"] == "started"
+    assert meta[_REQUEST_FIELD] is True
+
+    # Once journaling is active, a mismatched workspace fails closed.
+    with runguard.run_lock(repo, run_dir=run_dir):
+        _write_run_json(run_dir, "started", lock_workspace=repo)
+    assert _journal_path(run_dir).is_file()
+    run_before = (run_dir / "run.json").read_bytes()
+    journal_before = _journal_path(run_dir).read_bytes()
+    with runguard.run_lock(repo, run_dir=run_dir):
+        with pytest.raises(run_lifecycle.LifecycleJournalError):
+            _write_run_json(run_dir, "planning", lock_workspace=other)
+    assert (run_dir / "run.json").read_bytes() == run_before
+    assert _journal_path(run_dir).read_bytes() == journal_before
+
+
+def test_legacy_run_dir_stays_snapshot_only_with_flag_on(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+    # Legacy run directory: run.json present, no events/ journal.
+    localio.write_json(run_dir / "run.json", {"schema": "brigade.run.v1", "status": "started"})
+    assert not (run_dir / "events").exists()
+
+    _write_run_json_locked(repo, run_dir, "planning")
+
+    assert (run_dir / "run.json").is_file()
+    assert not (run_dir / "events").exists()
+    meta = json.loads((run_dir / "run.json").read_text())
+    assert meta["status"] == "planning"
+    assert _REQUEST_FIELD not in meta
+
+
+def test_activated_run_continues_without_env_flag(enabled, tmp_path, monkeypatch):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json(run_dir, "started")  # pre-lock bootstrap: request only
+    _write_run_json_locked(repo, run_dir, "started")  # in-lock: run.created
+
+    monkeypatch.delenv("BRIGADE_LIFECYCLE_JOURNAL", raising=False)
+    _write_run_json_locked(repo, run_dir, "planning")
+
+    events = _events(run_dir)
+    assert [e.event_type for e in events] == ["run.created", "run.planning.started"]
+
+
+def test_same_status_refresh_is_noop(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")  # same-status refresh
+
+    events = _events(run_dir)
+    assert len(events) == 1
+    assert events[0].event_type == "run.created"
+
+
+def test_same_status_with_changed_detail_appends_nothing(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+    secret_error = "SECRET_TOKEN=/super/private/path"
+
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "failed", error=secret_error)
+    journal_before = _journal_path(run_dir).read_bytes()
+
+    # A detail refresh is not a status transition: the journal must not grow,
+    # but the run.json snapshot refresh still proceeds.
+    _write_run_json_locked(repo, run_dir, "failed", error="different failure")
+
+    assert _journal_path(run_dir).read_bytes() == journal_before
+    events = _events(run_dir)
+    assert [e.event_type for e in events] == ["run.created", "run.failed"]
+    assert events[1].payload == {"status": "failed", "detail": "failed"}
+    journal_text = _journal_path(run_dir).read_text()
+    assert secret_error not in journal_text
+    assert "/super/private/path" not in journal_text
+    meta = json.loads((run_dir / "run.json").read_text())
+    assert meta["status"] == "failed"
+    assert meta["error"] == "different failure"
+
+
+def test_aba_recurrence_appends_all_three_occurrences(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "dispatching")
+    _write_run_json_locked(repo, run_dir, "result-processing")
+    _write_run_json_locked(repo, run_dir, "dispatching")  # A-B-A recurrence
+
+    events = _events(run_dir)
+    assert [e.event_type for e in events] == [
+        "run.created",
+        "run.dispatch.requested",
+        "run.dispatch.completed",
+        "run.dispatch.requested",
+    ]
+    assert [e.sequence for e in events] == [1, 2, 3, 4]
+    assert events[3].previous_digest == events[2].event_digest
+
+
+def test_unmapped_intermediate_status_still_appends_the_second_a(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "dispatching")
+    # Unmapped status: no event, but run.json still advances to it.
+    _write_run_json_locked(repo, run_dir, "artifact-collection")
+    # A-unmapped-B-A: the second dispatching is a real transition from the
+    # artifact-collection snapshot and must append even though the journal
+    # tail payload matches.
+    _write_run_json_locked(repo, run_dir, "dispatching")
+
+    meta = json.loads((run_dir / "run.json").read_text())
+    assert meta["status"] == "dispatching"
+    events = _events(run_dir)
+    assert [e.event_type for e in events] == [
+        "run.created",
+        "run.dispatch.requested",
+        "run.dispatch.requested",
+    ]
+    assert [e.sequence for e in events] == [1, 2, 3]
+    assert events[2].previous_digest == events[1].event_digest
+
+
+def test_retry_after_interruption_reuses_committed_event(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+
+    # Simulate append-before-snapshot interruption: record the transition
+    # (fsync happens inside append_event) without advancing run.json.
+    with runguard.run_lock(repo, run_dir=run_dir):
+        run_lifecycle.record_lifecycle_transition(run_dir, status="result-processing", workspace=repo)
+    journal_before = _journal_path(run_dir).read_bytes()
+
+    # Retry the same transition (e.g. the caller re-entered after a crash
+    # between journal fsync and run.json replacement): the prior snapshot is
+    # unchanged, so the derived idempotency key matches and the committed
+    # event is returned without a second append.
+    with runguard.run_lock(repo, run_dir=run_dir):
+        replay = run_lifecycle.record_lifecycle_transition(run_dir, status="result-processing", workspace=repo)
+
+    assert replay is not None
+    events = _events(run_dir)
+    assert len(events) == 2
+    assert [e.event_type for e in events] == ["run.created", "run.dispatch.completed"]
+    assert replay.event_id == events[1].event_id
+    assert _journal_path(run_dir).read_bytes() == journal_before
+
+
+def test_distinct_statuses_produce_distinct_sequence_linked_events(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "planning")
+    _write_run_json_locked(repo, run_dir, "failed", error="boom")
+
+    events = _events(run_dir)
+    assert [e.sequence for e in events] == [1, 2, 3]
+    assert events[0].previous_digest is None
+    assert events[1].previous_digest == events[0].event_digest
+    assert events[2].previous_digest == events[1].event_digest
+    assert events[2].payload == {"status": "failed", "detail": "failed"}
+    assert "boom" not in _journal_path(run_dir).read_text()
+
+
+def test_terminal_status_transitions_produce_allowlisted_events(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    cases = [
+        ("failed", "run.failed"),
+        ("canceled", "run.interrupted"),
+        ("timeout", "run.failed"),
+    ]
+    for index, (status, expected_type) in enumerate(cases):
+        run_dir = _run_dir(repo, f"20260728-16000{index}-beef{index:04d}")
+
+        _write_run_json(run_dir, "started")
+        _write_run_json_locked(repo, run_dir, "started")
+        _write_run_json_locked(repo, run_dir, status, error=f"boom {status}")
+
+        events = _events(run_dir)
+        assert [e.event_type for e in events] == ["run.created", expected_type]
+        tail = events[1]
+        assert tail.payload == {"status": status, "detail": status}
+        allowed = run_events.EVENT_TYPES[expected_type]
+        assert set(tail.payload.keys()) <= allowed
+        assert f"boom {status}" not in _journal_path(run_dir).read_text()
+
+
+def test_unmapped_status_writes_run_json_without_journal_event(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "dry-run")
+    _write_run_json_locked(repo, run_dir, "artifact-collection")
+
+    assert (run_dir / "run.json").is_file()
+    assert _events(run_dir) == [_events(run_dir)[0]]
+
+
+def test_recovery_writer_does_not_append(enabled, tmp_path):
+    """The stale-lock recovery writer uses localio directly and must not append."""
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+
+    payload = {"schema": "brigade.run.v1", "status": "failed", "error": "stale owner"}
+    localio.write_json(run_dir / "run.json", payload)
+
+    assert (run_dir / "run.json").is_file()
+    assert not (run_dir / "events").exists()
+
+
+def test_raw_oserror_is_bounded_and_blocks_run_json(enabled, tmp_path, monkeypatch):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+    assert _journal_path(run_dir).is_file()
+
+    before = (run_dir / "run.json").read_bytes()
+
+    def raise_oserror(path: Path) -> None:
+        raise OSError("simulated disk failure")
+
+    monkeypatch.setattr(run_journal, "ensure_journal", raise_oserror)
+
+    with runguard.run_lock(repo, run_dir=run_dir):
+        with pytest.raises(run_lifecycle.LifecycleJournalError) as excinfo:
+            _write_run_json(run_dir, "planning")
+
+    # The raw exception text could carry a path or private value, so only a
+    # generic bounded category surfaces.
+    assert "simulated disk failure" not in str(excinfo.value)
+    assert (run_dir / "run.json").read_bytes() == before
+    events = _events(run_dir)
+    assert [e.event_type for e in events] == ["run.created"]
+
+
+def test_canonicalization_error_is_bounded_and_blocks_run_json(enabled, tmp_path, monkeypatch):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+    assert _journal_path(run_dir).is_file()
+
+    before = (run_dir / "run.json").read_bytes()
+
+    def raise_canon(*args: object, **kwargs: object) -> dict:
+        raise run_events.CanonicalizationError("simulated canonicalization failure")
+
+    monkeypatch.setattr(run_events, "build_event", raise_canon)
+
+    with runguard.run_lock(repo, run_dir=run_dir):
+        with pytest.raises(run_lifecycle.LifecycleJournalError) as excinfo:
+            _write_run_json(run_dir, "planning")
+
+    assert "simulated canonicalization failure" not in str(excinfo.value)
+    assert (run_dir / "run.json").read_bytes() == before
+    events = _events(run_dir)
+    assert [e.event_type for e in events] == ["run.created"]
+
+
+def test_journal_corruption_blocks_run_json_advance(enabled, tmp_path):
+    repo = _repo(tmp_path)
+    run_dir = _run_dir(repo)
+    _write_run_json(run_dir, "started")
+    _write_run_json_locked(repo, run_dir, "started")
+    assert _journal_path(run_dir).is_file()
+
+    _journal_path(run_dir).write_text("not-a-json-line\n")
+    before = (run_dir / "run.json").read_bytes()
+
+    with runguard.run_lock(repo, run_dir=run_dir):
+        with pytest.raises(run_lifecycle.LifecycleJournalError):
+            _write_run_json(run_dir, "planning")
+
+    assert (run_dir / "run.json").read_bytes() == before
+    meta = json.loads((run_dir / "run.json").read_text())
+    assert meta["status"] == "started"