orchestrata_ultimate.session.context

Reference › orchestrata_ultimate › session › context

ContextFileManager owns the project's context.md and context.json pair, the durable record of every session (spec 006). It is the only class in the session package that reads and writes those files: the SessionController, the CLI commands (through _manager), the carryover, handover, archive and recovery managers and the hooks all go through it. The data types it reads and writes are in models.

The JSON file is the source of truth and keeps every session; the markdown file is rendered from it with Jinja templates and shows only the last SESSION_LIMIT sessions. If only the markdown exists, read parses it as a migration fallback, and ensure_initialized creates both files when they are missing and a virtual environment is present. write renders both under the write lock and then calls archive_old_sessions, which appends the sessions that fell out of the display window to context.history.md and never touches the JSON.

Several independent processes write these files, so durability is handled in three ways. _transaction takes an exclusive fcntl.flock on a sidecar lock file, re-entrant within a process, so read-modify-write cycles serialize across processes. _atomic_write_text writes a temporary sibling file, flushes and fsyncs it, then replaces the target, so a crash leaves the previous file intact. _recover_json repairs a corrupt context.json on read: it first salvages the leading valid JSON value (and validates it as ContextData), then falls back to the backup copy, and raises ValueError if neither works.

The accessors cover state (read_state, update_state), sessions (add_session, update_session, read_latest_session) and deferred work items (add_work_item, complete_work_item, and index-based complete and reopen). next_session_num takes the highest number from the JSON, the history file and the session-* git tags, so the counter cannot drift. verify_history_integrity checks gaps, duplicates and consistency between JSON and history, and recover_from_git rebuilds lost sessions from commit history, up to max_commits commits.

At a glance

Class diagram

Dependency diagram

API

Classes

ContextFileManager

Line 62

Manages context.md file operations.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/context.py

# mypy: ignore-errors
"""Fresh implementation of context management.

Handles context.md file operations with Jinja2 templates.
Uses JSON for data persistence, markdown for display.

Durability contract
-------------------
``context.json`` is written by several independent OS processes (the
``session_cli.py`` invocations the agent makes, the SessionStart hook that
shells out to that CLI, and the SessionEnd hook that drives this
``SessionController`` in-process — see ``claude/hooks/``). A plain
``Path.write_text`` does ``open('w')`` (truncate) then ``write``, which is
neither atomic nor mutually exclusive: two concurrent writers interleave and
the longer write's tail survives past the shorter one, leaving a file that
decodes as a valid object *followed by* leftover ``}``/``]`` garbage ("Extra
data" JSONDecodeError) — the recurring corruption this module guards against.

Defenses (the lock-file NAME and flock protocol below must stay compatible
with the standalone session-CLI implementation at
``~/.claude/tools/orchestrata/context.py`` — the module ``session_cli.py``
actually imports — so both writers mutually exclude on the same
``.context.lock``. The old ``~/.claude/tools/session/context.py`` copy is a
dead legacy artifact: it imports the retired ``orchestrata.models`` monorepo
package and is never imported by the CLI.):

1. **Atomic writes** — serialize to a sibling temp file, ``fsync``, then
   ``os.replace`` (atomic on POSIX); readers see only whole files.
2. **Cross-process locking** — every read-modify-write holds an exclusive
   ``fcntl.flock`` on a sidecar ``.context.lock``. Both context-manager
   implementations lock the SAME file, so they mutually exclude.
3. **Validated, self-healing reads** — writes round-trip through
   ``json.loads`` before commit, a ``.bak`` of the last good file is kept, and
   a corrupt read salvages the valid JSON prefix or restores the backup.
"""

import fcntl
import json
import logging
import os
import re
import subprocess
from collections.abc import Mapping
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path

from jinja2 import Environment, FileSystemLoader

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.models import (
    ContextData,
    ProjectState,
    SessionRecord,
    TaskState,
    WorkItem,
)

logger = logging.getLogger(__name__)


class ContextFileManager:
    """Manages context.md file operations.

    Uses JSON for data persistence, markdown for human-readable display.
    """

    CTX_FILE = "context.md"
    JSON_FILE = "context.json"
    HISTORY_FILE = "context.history.md"
    LOCK_FILE = ".context.lock"
    BACKUP_SUFFIX = ".bak"
    LINE_LIMIT = 500
    SESSION_LIMIT = 3

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-001"])
    def __init__(self, root: str | Path, env: Mapping[str, str] | None = None) -> None:
        """Initialize manager for project.

        Args:
            root: Project root directory.
            env: Optional explicit environment mapping passed to the ``git``
                subprocess calls used for tag/history recovery (for testing —
                e.g. simulating a missing ``git`` executable via a stripped
                ``PATH``); when omitted, those subprocesses inherit the real
                process environment exactly as before.
        """
        self.root = Path(root)
        self._env = env
        self.ctx_path = self.root / self.CTX_FILE
        self.json_path = self.root / self.JSON_FILE
        self.history_path = self.root / self.HISTORY_FILE
        self.lock_path = self.root / self.LOCK_FILE
        self.json_backup_path = self.root / (self.JSON_FILE + self.BACKUP_SUFFIX)

        # Re-entrancy state for the cross-process write lock. ``_lock_depth``
        # lets nested ``_transaction()`` blocks (e.g. add_session -> write ->
        # archive_old_sessions) share one held flock instead of self-deadlocking
        # on a second descriptor.
        self._lock_depth = 0
        self._lock_fp = None

        templates_dir = Path(__file__).parent / "templates"
        self.jinja = Environment(
            loader=FileSystemLoader(templates_dir),
            trim_blocks=True,
            lstrip_blocks=True,
        )

    @contextmanager
    def _transaction(self):
        """Hold an exclusive cross-process lock for a read-modify-write.

        Uses ``fcntl.flock`` on a dedicated sidecar lock file (never the data
        file, whose inode is swapped by every atomic write). Advisory but
        honored by every writer of this ``context.json`` — including the
        standalone ``session_cli.py`` copy, which locks the same path — so
        concurrent processes serialize instead of interleaving. Re-entrant
        within a process via ``_lock_depth``.
        """
        outermost = self._lock_depth == 0
        if outermost:
            # 'a' (not 'w') so we never truncate a lock file a peer may hold.
            self._lock_fp = open(self.lock_path, "a")
            fcntl.flock(self._lock_fp.fileno(), fcntl.LOCK_EX)
        self._lock_depth += 1
        try:
            yield
        finally:
            self._lock_depth -= 1
            if self._lock_depth == 0 and self._lock_fp is not None:
                try:
                    fcntl.flock(self._lock_fp.fileno(), fcntl.LOCK_UN)
                finally:
                    self._lock_fp.close()
                    self._lock_fp = None

    @staticmethod
    def _atomic_write_text(path: Path, content: str) -> None:
        """Write ``content`` to ``path`` atomically via temp file + os.replace.

        The temp file is a sibling (same filesystem, so the rename is atomic),
        flushed and ``fsync``ed before the rename. A crash leaves either the
        intact previous file or a leftover ``*.tmp`` — never a torn target.
        """
        tmp = path.with_name(f"{path.name}.tmp.{os.getpid()}")
        try:
            with open(tmp, "w", encoding="utf-8") as f:
                f.write(content)
                f.flush()
                os.fsync(f.fileno())
            os.replace(tmp, path)
        finally:
            try:
                if tmp.exists():
                    tmp.unlink()
            except OSError:
                pass

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-001"])
    def file_exists(self) -> bool:
        """Check if context files exist (JSON or markdown)."""
        return self.json_path.exists() or self.ctx_path.exists()

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-001", "REQ-006-FR14-004"])
    def ensure_initialized(self) -> ContextData:
        """Auto-create context files if ``.venv`` exists and files are missing.

        Detects the project root by the presence of ``.venv`` (or the ``root``
        passed to the constructor if it already contains ``context.json``).
        If context files don't exist, initializes them with a fresh
        ``ProjectState``. Also creates an empty ``context.history.md`` and
        ``CHANGELOG.md`` if they don't exist.

        Returns:
            Context data (new or existing).
        """
        if self.file_exists():
            return self.read()

        # Determine if this looks like a project root
        is_project = (self.root / ".venv").is_dir()
        if not is_project:
            # Maybe we're already at a known project root
            is_project = (self.root / "pyproject.toml").exists()

        if not is_project:
            raise FileNotFoundError(
                f"No context files and no .venv/pyproject.toml at {self.root}. "
                f"Run from a project root or create context files manually."
            )

        # Create required files
        logger.info("Auto-initializing session files at %s", self.root)
        ctx = self.initialize()

        # Ensure history file exists
        if not self.history_path.exists():
            self._atomic_write_text(
                self.history_path,
                "# Session History\n\n"
                "*Sessions are automatically archived here after "
                f"{self.SESSION_LIMIT} active sessions.*\n",
            )

        # Ensure CHANGELOG.md exists
        changelog = self.root / "CHANGELOG.md"
        if not changelog.exists():
            self._atomic_write_text(
                changelog,
                "# Changelog\n\n"
                "All notable changes to this project will be documented in this file.\n\n"
                "## [Unreleased]\n\n",
            )

        return ctx

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-001", "REQ-006-FR14-003"])
    def initialize(self, state: ProjectState | None = None) -> ContextData:
        """Create new context.md.

        Args:
            state: Optional initial state.

        Returns:
            Created context data.
        """
        if state is None:
            state = ProjectState()

        ctx = ContextData(state=state)
        self.write(ctx)
        return ctx

    @traces(spec="006", story="FR-016", reqs=["REQ-006-FR16-003"])
    def read(self) -> ContextData:
        """Read context data.

        Loads from JSON (source of truth) if available,
        falls back to markdown parsing for migration.

        Returns:
            Parsed context data.

        Raises:
            FileNotFoundError: If no context files exist.
        """
        if not self.file_exists():
            raise FileNotFoundError(f"context.md not found at {self.ctx_path}")

        # JSON is the source of truth
        if self.json_path.exists():
            return self._read_json()

        # Fallback: parse markdown (for migration)
        content = self.ctx_path.read_text()
        return self._parse(content)

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-001"])
    def _read_json(self) -> ContextData:
        """Read context from JSON, self-healing past residual corruption.

        Happy path is a plain ``json.loads``. On a JSONDecodeError (historically:
        a valid object plus leftover ``}``/``]`` from an interleaved write) it
        salvages the leading valid object, else restores the ``.bak`` copy, and
        rewrites the file clean so the corruption does not recur.
        """
        raw = self.json_path.read_text()
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            data = self._recover_json(raw)
        return ContextData.model_validate(data)

    def _recover_json(self, raw: str) -> dict:
        """Best-effort recovery of a corrupt context.json payload."""
        # 1) Salvage the leading valid JSON value (handles trailing garbage).
        try:
            data, _ = json.JSONDecoder().raw_decode(raw.lstrip())
            if isinstance(data, dict):
                ContextData.model_validate(data)  # must be a real ContextData
                self._atomic_write_text(
                    self.json_path,
                    ContextData.model_validate(data).model_dump_json(indent=2),
                )
                return data
        except json.JSONDecodeError, ValueError:
            pass

        # 2) Fall back to the last-known-good backup.
        if self.json_backup_path.exists():
            backup_raw = self.json_backup_path.read_text()
            data = json.loads(backup_raw)  # propagate if the backup is also bad
            self._atomic_write_text(self.json_path, backup_raw)
            return data

        raise ValueError(
            f"context.json at {self.json_path} is corrupt and unrecoverable "
            "(no salvageable prefix, no usable backup)"
        )

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-001"])
    def _write_json(self, ctx: ContextData) -> None:
        """Write context to JSON: validate, back up, then write atomically."""
        payload = ctx.model_dump_json(indent=2)
        json.loads(payload)  # guard: never persist anything unparseable

        if self.json_path.exists():
            try:
                current = self.json_path.read_text()
                json.loads(current)  # only back up a *valid* current file
                self._atomic_write_text(self.json_backup_path, current)
            except OSError, json.JSONDecodeError:
                pass  # current missing/corrupt -> nothing worth backing up

        self._atomic_write_text(self.json_path, payload)

    @traces(
        spec="006",
        story="FR-002",
        reqs=["REQ-006-FR14-002", "REQ-006-FR14-003", "REQ-006-FR2-004"],
    )
    def write(self, ctx: ContextData) -> None:
        """Save context to both JSON and markdown under the write lock.

        JSON keeps ALL sessions (never trimmed).
        Markdown shows only the last SESSION_LIMIT sessions.
        """
        with self._transaction():
            self._write_json(ctx)  # JSON keeps ALL sessions

            sorted_sessions = sorted(ctx.session_history, key=lambda s: s.seq)
            display_sessions = sorted_sessions[-self.SESSION_LIMIT :]

            tpl = self.jinja.get_template("context.md.j2")
            content = tpl.render(
                current_state=ctx.state,
                current_date=datetime.now().strftime("%Y-%m-%d"),
                carryover_tasks=ctx.deferred_work,
                sessions=display_sessions,
            )
            self._atomic_write_text(self.ctx_path, content)

            self.archive_old_sessions()

    @traces(spec="020", story="US-006", reqs=["REQ-020-173"])
    def read_state(self) -> ProjectState:
        """Get current state."""
        ctx = self.read()
        return ctx.state

    @traces(spec="020", story="US-006", reqs=["REQ-020-173"])
    def update_state(self, **kwargs: object) -> ProjectState:
        """Update state fields.

        Args:
            **kwargs: Fields to update.

        Returns:
            Updated state.
        """
        with self._transaction():
            ctx = self.read()
            for k, v in kwargs.items():
                if hasattr(ctx.state, k):
                    setattr(ctx.state, k, v)
            self.write(ctx)
            return ctx.state

    @traces(spec="020", story="US-006", reqs=["REQ-020-172"])
    def read_sessions(self) -> list[SessionRecord]:
        """Get all sessions."""
        ctx = self.read()
        return ctx.session_history

    @traces(spec="006", story="FR-010", reqs=["REQ-006-FR10-004"])
    def read_latest_session(self) -> SessionRecord | None:
        """Get most recent session."""
        ctx = self.read()
        return ctx.latest_session()

    @traces(spec="020", story="US-006", reqs=["REQ-020-172"])
    def add_session(self, session: SessionRecord) -> None:
        """Add new session.

        Args:
            session: Session to add.
        """
        with self._transaction():
            ctx = self.read()
            ctx.append_session(session)
            self.write(ctx)
            self.archive_old_sessions()

    @traces(spec="020", story="US-006", reqs=["REQ-020-172"])
    def update_session(self, seq: int, **kwargs: object) -> SessionRecord | None:
        """Update existing session.

        Args:
            seq: Session number.
            **kwargs: Fields to update.

        Returns:
            Updated session or None.
        """
        with self._transaction():
            ctx = self.read()
            for session in ctx.session_history:
                if session.seq == seq:
                    for k, v in kwargs.items():
                        if hasattr(session, k):
                            setattr(session, k, v)
                    self.write(ctx)
                    return session
            return None

    @traces(spec="020", story="US-006", reqs=["REQ-020-174"])
    def read_work_items(self) -> list[WorkItem]:
        """Get all deferred work items."""
        ctx = self.read()
        return ctx.deferred_work

    @traces(spec="020", story="US-006", reqs=["REQ-020-174"])
    def add_work_item(self, item: WorkItem) -> None:
        """Add deferred work item.

        Args:
            item: Work item to add.
        """
        with self._transaction():
            ctx = self.read()
            ctx.deferred_work.append(item)
            self.write(ctx)

    @traces(spec="020", story="US-006", reqs=["REQ-020-174"])
    def reopen_work_item_at(self, index: int) -> WorkItem | None:
        """Return the work item at *index* to TODO.

        The undo for :meth:`complete_work_item_at`. ``task done`` takes a bare
        index with no confirmation, and hand-editing context.json is
        forbidden, so without this a mistyped index is an unrecoverable state
        change.

        Args:
            index: Position in ``deferred_work``.

        Returns:
            The reopened item, or None when *index* is out of range.
        """
        with self._transaction():
            ctx = self.read()
            if index < 0 or index >= len(ctx.deferred_work):
                return None
            item = ctx.deferred_work[index]
            item.mark_todo()
            self.write(ctx)
            return item

    @traces(spec="020", story="US-006", reqs=["REQ-020-174"])
    def complete_work_item_at(self, index: int) -> WorkItem | None:
        """Mark the work item at *index* as done.

        The index-based twin of :meth:`complete_work_item`. Both read, mutate
        and write ONE context inside a single transaction — mutating an item
        obtained from a separate read and then writing a freshly-read context
        discards the change silently, which is exactly what the CLI used to do.

        Args:
            index: Position in ``deferred_work``.

        Returns:
            The completed item, or None when *index* is out of range.
        """
        with self._transaction():
            ctx = self.read()
            if index < 0 or index >= len(ctx.deferred_work):
                return None
            item = ctx.deferred_work[index]
            item.mark_done()
            self.write(ctx)
            return item

    @traces(spec="020", story="US-006", reqs=["REQ-020-174"])
    def complete_work_item(self, text: str) -> bool:
        """Mark work item as done.

        Args:
            text: Item text to match.

        Returns:
            True if found and completed.
        """
        with self._transaction():
            ctx = self.read()
            for item in ctx.deferred_work:
                if item.text == text:
                    item.mark_done()
                    self.write(ctx)
                    return True
            return False

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-005", "REQ-006-FR15-001"])
    def archive_old_sessions(self) -> None:
        """Write sessions older than the last SESSION_LIMIT to context.history.md.

        Appends only sessions not already archived. Never modifies context.json.
        """
        with self._transaction():
            ctx = self.read()
            if len(ctx.session_history) <= self.SESSION_LIMIT:
                return

            sorted_sessions = sorted(ctx.session_history, key=lambda s: s.seq)
            older_sessions = sorted_sessions[: -self.SESSION_LIMIT]

            already_archived: set[int] = set()
            if self.history_path.exists():
                already_archived.update(
                    int(match.group(1))
                    for match in re.finditer(
                        r"### Session (\d+)", self.history_path.read_text()
                    )
                )

            new_to_archive = [
                s for s in older_sessions if s.seq not in already_archived
            ]
            if not new_to_archive:
                return

            tpl = self.jinja.get_template("session.md.j2")
            archive_text = "\n\n".join(tpl.render(session=s) for s in new_to_archive)

            if self.history_path.exists():
                existing = self.history_path.read_text()
                self._atomic_write_text(
                    self.history_path, existing + "\n\n" + archive_text
                )
            else:
                self._atomic_write_text(self.history_path, archive_text)

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-007"])
    def _get_max_session_from_history(self) -> int:
        """Get highest session number from history file.

        Returns:
            Max session number found, or -1 if no history.
        """
        if not self.history_path.exists():
            return -1

        content = self.history_path.read_text()
        session_nums = re.findall(r"### Session (\d+)", content)
        if not session_nums:
            return -1

        return max(int(n) for n in session_nums)

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-007"])
    def _get_max_session_from_tags(self) -> int:
        """Get highest session number from git tags.

        Parses tags matching 'session-*' pattern and extracts
        session numbers to prevent counter drift when sessions
        happen outside the hook system.

        Returns:
            Max session number found in tags, or -1 if none.
        """
        try:
            result = subprocess.run(
                ["git", "tag", "-l", "session-*"],
                cwd=self.root,
                capture_output=True,
                text=True,
                timeout=10,
                env=self._env,
            )
            if result.returncode != 0 or not result.stdout.strip():
                return -1

            max_num = -1
            for tag in result.stdout.strip().splitlines():
                match = re.match(r"session-(\d+)", tag.strip())
                if match:
                    num = int(match.group(1))
                    if num > max_num:
                        max_num = num
            return max_num
        except subprocess.TimeoutExpired, FileNotFoundError, OSError:
            return -1

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-007"])
    def next_session_num(self) -> int:
        """Get next session number, checking context.md, history, and git tags.

        Considers all three sources to prevent counter drift:
        - context.json session_history
        - context.history.md archived sessions
        - git tags matching session-*

        Returns:
            Next session number.
        """
        max_from_history = self._get_max_session_from_history()
        max_from_tags = self._get_max_session_from_tags()

        max_from_context = -1
        if self.file_exists():
            ctx = self.read()
            if ctx.session_history:
                max_from_context = max(s.seq for s in ctx.session_history)

        return max(max_from_history, max_from_context, max_from_tags) + 1

    @traces(
        spec="006",
        story="FR-014",
        reqs=[
            "REQ-006-FR14-007",
            "REQ-006-FR16-001",
            "REQ-006-FR16-002",
            "REQ-006-FR16-003",
            "REQ-006-FR16-004",
            "REQ-006-FR18-002",
            "REQ-006-NFR9-002",
        ],
    )
    def verify_history_integrity(self) -> dict:
        """Verify integrity of session history across context.json and context.history.md.

        Checks:
        - Gap detection in session numbering
        - Duplicate detection across files
        - Sync validation (context.md matches context.json render)
        - History continuity (max history < min context)

        Returns:
            Dict with keys: valid, gaps, duplicates, sync_ok, continuity_ok, errors.
        """
        result = {
            "valid": True,
            "gaps": [],
            "duplicates": [],
            "sync_ok": True,
            "continuity_ok": True,
            "errors": [],
        }

        json_nums = self._collect_json_session_nums(result)
        history_nums = self._collect_history_session_nums()

        self._check_gaps(history_nums, json_nums, result)
        self._check_duplicates(history_nums, json_nums, result)
        self._check_sync(result)
        self._check_continuity(history_nums, json_nums, result)

        return result

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-007"])
    def _collect_json_session_nums(self, result: dict) -> list[int]:
        """Return sorted session seqs from context.json (records errors on failure)."""
        if not self.json_path.exists():
            return []
        try:
            ctx = self._read_json()
            return sorted(s.seq for s in ctx.session_history)
        except Exception as e:
            result["errors"].append(f"Failed to read context.json: {e}")
            result["valid"] = False
            return []

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-007"])
    def _collect_history_session_nums(self) -> list[int]:
        """Return sorted session seqs parsed from context.history.md."""
        if not self.history_path.exists():
            return []
        content = self.history_path.read_text()
        matches = re.findall(r"### Session (\d+)", content)
        return sorted(int(m) for m in matches)

    @staticmethod
    @traces(spec="006", story="FR-016", reqs=["REQ-006-FR16-001"])
    def _check_gaps(
        history_nums: list[int], json_nums: list[int], result: dict
    ) -> None:
        """Flag missing session numbers in the combined range."""
        all_nums = sorted(set(history_nums + json_nums))
        if not all_nums:
            return
        expected = range(min(all_nums), max(all_nums) + 1)
        gaps = [n for n in expected if n not in all_nums]
        if gaps:
            result["gaps"] = gaps
            result["valid"] = False

    @staticmethod
    @traces(spec="006", story="FR-016", reqs=["REQ-006-FR16-002"])
    def _check_duplicates(
        history_nums: list[int], json_nums: list[int], result: dict
    ) -> None:
        """Flag session numbers that appear more than once across both files."""
        seen: set[int] = set()
        dupes: list[int] = []
        for n in history_nums + json_nums:
            if n in seen:
                dupes.append(n)
            seen.add(n)
        if dupes:
            result["duplicates"] = sorted(set(dupes))
            result["valid"] = False

    @traces(spec="006", story="FR-016", reqs=["REQ-006-FR16-003"])
    def _check_sync(self, result: dict) -> None:
        """Verify context.md matches the rendered context.json."""
        if not (self.json_path.exists() and self.ctx_path.exists()):
            return
        try:
            ctx = self._read_json()
            tpl = self.jinja.get_template("context.md.j2")
            from datetime import datetime as dt

            sorted_sessions = sorted(ctx.session_history, key=lambda s: s.seq)
            display_sessions = sorted_sessions[-self.SESSION_LIMIT :]
            rendered = tpl.render(
                current_state=ctx.state,
                current_date=dt.now().strftime("%Y-%m-%d"),
                carryover_tasks=ctx.deferred_work,
                sessions=display_sessions,
            )
            actual = self.ctx_path.read_text()
            if rendered.strip() != actual.strip():
                result["sync_ok"] = False
                result["valid"] = False
        except Exception as e:
            result["errors"].append(f"Sync check failed: {e}")
            result["sync_ok"] = False
            result["valid"] = False

    @staticmethod
    @traces(spec="006", story="FR-016", reqs=["REQ-006-FR16-004"])
    def _check_continuity(
        history_nums: list[int], json_nums: list[int], result: dict
    ) -> None:
        """Verify history sessions all precede context.json sessions."""
        if history_nums and json_nums and max(history_nums) >= min(json_nums):
            result["continuity_ok"] = False
            result["valid"] = False

    @traces(
        spec="006",
        story="FR-017",
        reqs=[
            "REQ-006-FR17-001",
            "REQ-006-FR17-002",
            "REQ-006-FR17-003",
            "REQ-006-FR17-004",
            "REQ-006-FR18-003",
            "REQ-006-NFR9-003",
        ],
    )
    def recover_from_git(self, max_commits: int = 200) -> dict:
        """Recover lost sessions from git history.

        Scans git log for historical context.json commits, extracts session
        data, and appends any missing sessions to context.history.md with
        a [RECOVERED] annotation.

        Non-destructive: never overwrites existing session data.

        Args:
            max_commits: Maximum number of commits to scan.

        Returns:
            Dict with keys: recovered, already_present, commits_scanned, errors.
        """
        result = {
            "recovered": [],
            "already_present": [],
            "commits_scanned": 0,
            "errors": [],
        }

        known_nums = self._collect_known_session_nums(result)

        commits = self._git_context_commits(max_commits, result)
        if commits is None:
            return result
        result["commits_scanned"] = len(commits)

        recovered_sessions = self._scan_commits_for_sessions(
            commits, known_nums, result
        )
        self._append_recovered_sessions(recovered_sessions)

        result["already_present"] = sorted(set(result["already_present"]))
        result["recovered"] = sorted(set(result["recovered"]))
        return result

    @traces(spec="006", story="FR-017", reqs=["REQ-006-FR17-004"])
    def _collect_known_session_nums(self, result: dict) -> set[int]:
        """Gather session seqs already present in context.json and history.md."""
        known_nums: set[int] = set()
        if self.json_path.exists():
            try:
                ctx = self._read_json()
                known_nums.update(s.seq for s in ctx.session_history)
            except Exception as e:
                result["errors"].append(f"Failed to read context.json: {e}")

        if self.history_path.exists():
            content = self.history_path.read_text()
            matches = re.findall(r"### Session (\d+)", content)
            known_nums.update(int(m) for m in matches)
        return known_nums

    @traces(spec="006", story="FR-017", reqs=["REQ-006-FR17-001", "REQ-006-NFR9-003"])
    def _git_context_commits(self, max_commits: int, result: dict) -> list[str] | None:
        """Return commit hashes that touched context.json, or None to abort recovery."""
        try:
            log_result = subprocess.run(
                [
                    "git",
                    "log",
                    f"-n{max_commits}",
                    "--format=%H",
                    "--",
                    "context.json",
                ],
                cwd=self.root,
                capture_output=True,
                text=True,
                timeout=30,
                env=self._env,
            )
        except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as e:
            result["errors"].append(f"Git log failed: {e}")
            return None

        if log_result.returncode != 0 or not log_result.stdout.strip():
            return None
        return log_result.stdout.strip().splitlines()

    @traces(spec="006", story="FR-017", reqs=["REQ-006-FR17-002"])
    def _scan_commits_for_sessions(
        self, commits: list[str], known_nums: set[int], result: dict
    ) -> list[SessionRecord]:
        """Extract sessions missing from the current history out of the given commits."""
        recovered_sessions: list[SessionRecord] = []
        for commit_hash in commits:
            try:
                show_result = subprocess.run(
                    ["git", "show", f"{commit_hash}:context.json"],
                    cwd=self.root,
                    capture_output=True,
                    text=True,
                    timeout=10,
                    env=self._env,
                )
                if show_result.returncode != 0:
                    continue

                data = json.loads(show_result.stdout)
                ctx_data = ContextData.model_validate(data)

                for session in ctx_data.session_history:
                    if session.seq in known_nums:
                        if session.seq not in [s.seq for s in recovered_sessions]:
                            result["already_present"].append(session.seq)
                        continue
                    known_nums.add(session.seq)
                    recovered_sessions.append(session)
                    result["recovered"].append(session.seq)

            except (json.JSONDecodeError, Exception) as e:
                result["errors"].append(
                    f"Failed to parse commit {commit_hash[:8]}: {e}"
                )
                continue
        return recovered_sessions

    @traces(spec="006", story="FR-017", reqs=["REQ-006-FR17-003"])
    def _append_recovered_sessions(
        self, recovered_sessions: list[SessionRecord]
    ) -> None:
        """Render recovered sessions and append them to context.history.md."""
        if not recovered_sessions:
            return
        recovered_sessions.sort(key=lambda s: s.seq)
        tpl = self.jinja.get_template("session.md.j2")
        new_content = ""
        for session in recovered_sessions:
            header_annotation = "\n> [RECOVERED] from git history\n\n"
            new_content += tpl.render(session=session) + header_annotation

        if self.history_path.exists():
            existing = self.history_path.read_text()
            self._atomic_write_text(self.history_path, existing + "\n" + new_content)
        else:
            header = "# Session History Archive\n\n"
            self._atomic_write_text(self.history_path, header + new_content)

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-002"])
    def _parse_current_state(self, content: str, ctx: ContextData) -> None:
        """Parse the ## Current State block and populate ctx.state.

        Args:
            content: Raw markdown content.
            ctx: ContextData instance to mutate.
        """
        state_match = re.search(
            r"## Current State\n(.*?)(?=\n## |\Z)", content, re.DOTALL
        )
        if not state_match:
            return

        state_text = state_match.group(1)

        session_match = re.search(r"Session (\d+)", state_text)
        if session_match:
            ctx.state.active_session = int(session_match.group(1))

        version_match = re.search(r"Current Version: (.+)", state_text)
        if version_match:
            ctx.state.version = version_match.group(1).strip()

        branch_match = re.search(r"Branch: (.+)", state_text)
        if branch_match:
            ctx.state.git_branch = branch_match.group(1).strip()

        test_match = re.search(r"Test Status: (.+)", state_text)
        if test_match:
            ctx.state.test_state = test_match.group(1).strip()

        status_match = re.search(r"Status: (.+)", state_text)
        if status_match:
            ctx.state.status_line = status_match.group(1).strip()

    @traces(spec="020", story="US-006", reqs=["REQ-020-174"])
    def _parse_carryover_tasks(self, content: str, ctx: ContextData) -> None:
        """Parse the ## Carryover Tasks block and populate ctx.deferred_work.

        Args:
            content: Raw markdown content.
            ctx: ContextData instance to mutate.
        """
        carryover_match = re.search(
            r"## Carryover Tasks.*?\n(.*?)(?=\n## |\Z)", content, re.DOTALL
        )
        if not carryover_match:
            return

        tasks_text = carryover_match.group(1)
        for line in tasks_text.strip().splitlines():
            line = line.strip()
            if not line.startswith("- ") or "{none" in line.lower():
                continue

            task_desc = line[2:].strip()

            # Parse checkbox: [x] or [ ]
            state = TaskState.DEFERRED
            checkbox_match = re.match(r"\[([ x])\]\s*(.+)", task_desc)
            if checkbox_match:
                if checkbox_match.group(1) == "x":
                    state = TaskState.DONE
                task_desc = checkbox_match.group(2)

            # Parse priority: [P0], [P1], etc.
            priority = None
            priority_match = re.match(r"\[([^\]]+)\]\s*(.+)", task_desc)
            if priority_match:
                priority = priority_match.group(1)
                task_desc = priority_match.group(2)

            # Parse notes: (notes text)
            notes = None
            notes_match = re.match(r"(.+?)\s*\(([^)]+)\)\s*$", task_desc)
            if notes_match:
                task_desc = notes_match.group(1)
                notes = notes_match.group(2)

            ctx.deferred_work.append(
                WorkItem(text=task_desc, priority=priority, state=state, notes=notes)
            )

    @traces(spec="020", story="US-006", reqs=["REQ-020-172"])
    def _parse_session_record(self, match: re.Match, content: str) -> SessionRecord:
        """Build a SessionRecord from a session header regex match.

        Args:
            match: Regex match object for a ``### Session NNN`` header line.
            content: Full raw markdown content (used for sub-searches).

        Returns:
            Populated SessionRecord.
        """
        seq = int(match.group(1))
        seq_str = match.group(1)  # Preserve the original string (e.g., "001")
        date_str = match.group(2)
        model = match.group(3).strip()

        try:
            when = datetime.strptime(date_str, "%Y-%m-%d %H:%M")
        except ValueError:
            when = datetime.now()

        # Use the original seq_str to match the zero-padded format from template
        focus_match = re.search(
            rf"### Session {seq_str}.*?Focus: (.+?)(?:\n|$)", content, re.DOTALL
        )
        purpose = focus_match.group(1).strip() if focus_match else "Unknown"

        # Parse Next Steps section for upcoming items
        upcoming: list[str] = []
        next_steps_match = re.search(
            rf"### Session {seq_str}.*?#### Next Steps\n(.*?)(?=\n####|\n### Session|\Z)",
            content,
            re.DOTALL,
        )
        if next_steps_match:
            steps_text = next_steps_match.group(1)
            for line in steps_text.strip().splitlines():
                line = line.strip()
                if line.startswith("- "):
                    upcoming.append(line[2:].strip())

        return SessionRecord(
            seq=seq, when=when, ai_model=model, purpose=purpose, upcoming=upcoming
        )

    @traces(spec="020", story="US-006", reqs=["REQ-020-172"])
    def _parse_session_records(self, content: str, ctx: ContextData) -> None:
        """Parse all ### Session NNN header blocks and populate ctx.session_history.

        Args:
            content: Raw markdown content.
            ctx: ContextData instance to mutate.
        """
        session_pattern = r"### Session (\d+) - ([\d-]+ [\d:]+) - (.+?)\n"
        for match in re.finditer(session_pattern, content):
            ctx.session_history.append(self._parse_session_record(match, content))

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-002"])
    def _parse(self, content: str) -> ContextData:
        """Parse context.md content.

        Args:
            content: Raw markdown content.

        Returns:
            Parsed context data.
        """
        ctx = ContextData()
        self._parse_current_state(content, ctx)
        self._parse_carryover_tasks(content, ctx)
        self._parse_session_records(content, ctx)
        return ctx