work_queue_file.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/work_queue_file.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_orchestrator/tests/test_258_work_queue_file.py"
"""
The file backend: an append-only operation log plus a Markdown projection.
 
Purpose
-------
Plan 258 P2, from the factory-decomposition design note §5 and §7. This
is the DEFAULT backend and the one the package's own tests run against,
which is the condition that stops it rotting into dead code kept warm
for strangers. It is two files under one injected directory:
 
``work_queue.ops.jsonl``
    The append-only log of typed operations, one JSON object per line
    in the :class:`~oc_orchestrator.work_queue.QueueOp` shape. Opened for
    APPEND only, flushed and fsynced per write, never rewritten. This
    file is the truth.
 
``work_queue.md``
    The projection, rendered by folding the log. A view a human may
    also edit; :func:`FileQueueBackend.reconcile` compares the fold
    with the parsed view and, with ``apply``, records the hand edit as
    operations so the log stays complete.
 
The projection's grammar is lifted from the queue plan 223 retired at
commit ``9ac4b6a1``, with the letters removed: a first-line marker,
three tracked H2 sections, ``- <item_id>`` bullets with an optional
bracketed annotation, and every line outside a tracked bullet list
preserved verbatim on round trip. :func:`_windows_safe_replace` is
lifted VERBATIM from ``e1b5c716:oc_chat/oc_chat/core/agent_queue.py``
because no dispatch-layer library module carries a temp-then-replace
that recovers from a read-only target: ``scripts/plan_manage.py``'s
``_atomic_write`` is script-private and has no recovery, and
``core/provider_secrets.py`` is security code that is not imported for
a queue.
 
Responsibilities
----------------
- :class:`QueuePaths` -- the injected filesystem inputs, in the
  ``PlanPaths`` shape.
- :func:`read_ops` / :func:`fold` -- the log, and the state it folds to.
- :func:`render_projection` / :func:`parse_projection` -- the view.
- :class:`FileQueueBackend` -- the six contract operations, plus
  :meth:`~FileQueueBackend.write_projection` and
  :meth:`~FileQueueBackend.reconcile`.
- :class:`QueueDrift` / :class:`CorruptLogError` -- what reconcile
  reports, and what a bad log line raises.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (a write, a fold, a reconcile)
    L3 - semantic details (a refusal, a drift class)
    L4 - deep tracing (module load, per-line parse)
 
Contracts
---------
- NO LOCKING. Two writers can lose an update: each reads the log,
  decides, and appends, so a concurrent pair can both append an op that
  was computed against the same prefix. That is accepted while the
  operator session is the sole writer (design note §5, and the plan-255
  sidecar carries the same statement). A lock file or a daemon is
  forbidden by operator ruling, not merely out of scope.
- The log is APPEND-ONLY. No code path in this module opens the ops
  file for writing, truncation or replacement. The projection is the
  only file that is rewritten.
- A corrupt log line is :class:`CorruptLogError` naming the file and
  the 1-BASED line number. The backend never silently skips a line.
- A re-render of unchanged state writes NOTHING:
  :meth:`FileQueueBackend.write_projection` returns ``False`` and the
  file's bytes and mtime are untouched.
- ``note`` lives in the log, not in the projection: the annotation
  grammar is ``key=value`` pairs separated by commas, and free prose
  would not survive it. ``lane`` and ``instance`` are annotated.
- Nothing here reads the process working directory. :class:`QueuePaths`
  defaults to the relative literal that is today's location, exactly as
  :class:`~oc_orchestrator.plan_inventory.PlanPaths` does, and the caller
  resolves it.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.work_queue (the contract)
"""
 
from __future__ import annotations
 
import json
import os
import re
import stat
import tempfile
import time
from dataclasses import dataclass, field, replace
from datetime import datetime
from pathlib import Path
from typing import Any
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.work_queue import (
    KIND_PLAN,
    OP_COMPLETE,
    OP_DEQUEUE,
    OP_ENQUEUE,
    OP_MOVE,
    OP_NAMES,
    BadPositionError,
    DuplicateItemError,
    QueueBackendError,
    QueueOp,
    QueueState,
    UnknownItemError,
    WorkItem,
)
 
_dbg("OC_CHAT.MAIN", "module", "work_queue_file loaded", 4)
 
 
# ── Constants ────────────────────────────────────────────────────────────
 
 
#: Where the two files live, relative to the caller's root. Today's
#: value; injected through :class:`QueuePaths`, never discovered. It is
#: the monorepo's layout and a DEFAULT, not a requirement: a consumer
#: whose queue lives elsewhere passes its own ``QueuePaths(queue_dir=...)``
#: and this literal is never read.
DEFAULT_QUEUE_DIR = Path(".option_c/orchestration")
 
#: The append-only typed operation log.
OPS_FILENAME = "work_queue.ops.jsonl"
 
#: The Markdown projection folded out of the log.
PROJECTION_FILENAME = "work_queue.md"
 
#: The first-line marker the writer emits and the parser tolerates.
MARKER = "<!-- work-queue: v1 -->"
 
#: The three tracked H2 sections, in render order, paired with the
#: :class:`~oc_orchestrator.work_queue.QueueState` field each one holds.
SECTIONS: tuple[tuple[str, str], ...] = (
    ("Pending", "pending"),
    ("In progress", "in_progress"),
    ("Completed", "completed"),
)
 
#: Rendered in a tracked section that holds nothing. Lifted from the
#: retired renderer so an operator reading the file sees what they saw.
EMPTY_MARK = "_(none)_"
 
# OCT-LINT: disable=no_hardcoded_secrets reason="Markdown grammar: the tracked-H2 header pattern for the projection parser; no credential content"
_SECTION_RE = re.compile(r"^##\s+(?P<name>Pending|In progress|Completed)\s*$", re.I)
_BULLET_RE = re.compile(
    # OCT-LINT: disable=no_hardcoded_secrets reason="Markdown grammar: the work-item bullet pattern (id plus optional [k=v] annotation); no credential content"
    r"^-\s+(?P<item_id>[A-Za-z0-9][A-Za-z0-9_\-./]*)"
    r"(?:\s+\‍[(?P<annotations>[^\‍]]*)\‍])?\s*$"
)
 
_ACTOR_RECONCILE = "reconcile"
 
 
class CorruptLogError(QueueBackendError):
    """A line in the operation log does not parse; the message names it."""
 
 
# ── Paths ────────────────────────────────────────────────────────────────
 
 
@dataclass(frozen=True)
class QueuePaths:
    """All caller-controlled filesystem inputs to one queue.
 
    The :class:`~oc_orchestrator.plan_inventory.PlanPaths` shape: a frozen
    dataclass whose defaults are today's values, constructed at the
    boundary and passed whole. ``queue_dir`` is late-bound through a
    default factory so a caller that monkeypatches
    :data:`DEFAULT_QUEUE_DIR` is still honoured.
    """
 
    queue_dir: Path = field(default_factory=lambda: DEFAULT_QUEUE_DIR)
    #: Overrides the derived ``<queue_dir>/work_queue.ops.jsonl``.
    ops_path_override: Path | None = None
    #: Overrides the derived ``<queue_dir>/work_queue.md``.
    projection_path_override: Path | None = None
 
    @property
    def ops_path(self) -> Path:
        """The append-only log: the override, else derived."""
        if self.ops_path_override is not None:
            return self.ops_path_override
        return self.queue_dir / OPS_FILENAME
 
    @property
    def projection_path(self) -> Path:
        """The Markdown view: the override, else derived."""
        if self.projection_path_override is not None:
            return self.projection_path_override
        return self.queue_dir / PROJECTION_FILENAME
 
 
# ── The log ──────────────────────────────────────────────────────────────
 
 
def _now_iso() -> str:
    """A local-with-offset ISO 8601 timestamp, seconds resolution.
 
    The project-wide convention, matching
    ``core/plan_inventory.py::_now_iso``.
    """
    return datetime.now().astimezone().replace(microsecond=0).isoformat()
 
 
def read_ops(ops_path: Path) -> tuple[QueueOp, ...]:
    """Read every operation in *ops_path*, oldest first.
 
    A missing file is an empty log, never a ``FileNotFoundError`` --
    the same tolerance the retired ``load_queue`` had.
 
    Raises:
        CorruptLogError: SYSTEM_ERROR (L1 for the caller) -- a
            non-blank line is not a JSON object with the operation
            fields, or names an operation that is not in
            :data:`~oc_orchestrator.work_queue.OP_NAMES`. The message
            names the file and the 1-based line number; the reader
            never silently skips it.
    """
    func = "read_ops"
    path = Path(ops_path)
    try:
        text = path.read_text(encoding="utf-8")
    except FileNotFoundError:
        _dbg("OC_CHAT.MAIN", func, f"{path}: absent; empty log", 2)
        return ()
    ops: list[QueueOp] = []
    for number, line in enumerate(text.splitlines(), start=1):
        if not line.strip():
            continue
        ops.append(_decode_op(line, path, number))
    _dbg("OC_CHAT.MAIN", func, f"{path}: {len(ops)} operation(s)", 2)
    return tuple(ops)
 
 
def _decode_op(line: str, path: Path, number: int) -> QueueOp:
    """One log line to a :class:`QueueOp`, or a refusal that names it."""
    func = "_decode_op"
    try:
        raw = json.loads(line)
    except json.JSONDecodeError as exc:
        _dbg("OC_CHAT.MAIN", func, f"SYSTEM_ERROR: line {number}: {exc}", 1, override=True)
        raise CorruptLogError(
            f"{path.name}: line {number} is not JSON: {exc}"
        ) from exc
    if not isinstance(raw, dict):
        raise CorruptLogError(
            f"{path.name}: line {number} is not a JSON object: {type(raw).__name__}"
        )
    missing = {"seq", "at", "op", "item_id"} - set(raw)
    if missing:
        raise CorruptLogError(
            f"{path.name}: line {number} is missing {', '.join(sorted(missing))}"
        )
    op_name = str(raw["op"])
    if op_name not in OP_NAMES:
        raise CorruptLogError(
            f"{path.name}: line {number} names an unknown operation {op_name!r}; "
            f"expected one of {', '.join(OP_NAMES)}"
        )
    payload = raw.get("payload") or {}
    if not isinstance(payload, dict):
        raise CorruptLogError(
            f"{path.name}: line {number} has a non-object payload"
        )
    return QueueOp(
        seq=int(raw["seq"]),
        at=str(raw["at"]),
        op=op_name,
        item_id=str(raw["item_id"]),
        payload=payload,
        actor=str(raw.get("actor", "")),
    )
 
 
def _encode_op(op: QueueOp) -> str:
    """One :class:`QueueOp` to its log line, without the newline."""
    return json.dumps(
        {
            "seq": op.seq,
            "at": op.at,
            "op": op.op,
            "item_id": op.item_id,
            "payload": dict(op.payload),
            "actor": op.actor,
        },
        ensure_ascii=False,
        sort_keys=True,
    )
 
 
def append_op(ops_path: Path, op: QueueOp) -> None:
    """Append *op* to the log, flushed and fsynced.
 
    The ONLY write path for the log, and it opens the file in append
    mode. Nothing in this module truncates or replaces it.
    """
    func = "append_op"
    path = Path(ops_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8", newline="\n") as handle:
        handle.write(_encode_op(op) + "\n")
        handle.flush()
        os.fsync(handle.fileno())
    _dbg("OC_CHAT.MAIN", func, f"{path.name} += {op.op} {op.item_id} (seq {op.seq})", 2)
 
 
def _item_from_payload(item_id: str, payload: Any) -> WorkItem:
    """Rebuild the work item an ``enqueue`` payload carried."""
    return WorkItem(
        kind=str(payload.get("kind", KIND_PLAN)),
        item_id=item_id,
        lane=_optional_str(payload.get("lane")),
        instance=_optional_str(payload.get("instance")),
        note=_optional_str(payload.get("note")),
    )
 
 
def _optional_str(value: object) -> str | None:
    return None if value is None else str(value)
 
 
def fold(ops: tuple[QueueOp, ...]) -> QueueState:
    """Replay *ops* into the three buckets. The fold IS the state.
 
    Raises:
        CorruptLogError: SYSTEM_ERROR -- an operation refers to an item
            the earlier operations never put anywhere. A log that
            cannot be folded is corrupt, and saying so beats guessing.
    """
    func = "fold"
    pending: list[WorkItem] = []
    in_progress: list[WorkItem] = []
    completed: list[WorkItem] = []
 
    def _take(bucket: list[WorkItem], item_id: str) -> WorkItem | None:
        for index, item in enumerate(bucket):
            if item.item_id == item_id:
                return bucket.pop(index)
        return None
 
    for op in ops:
        if op.op == OP_ENQUEUE:
            item = _item_from_payload(op.item_id, op.payload)
            # A re-enqueue takes the item back out of completed: it is
            # being worked again, which is the motion the registry's own
            # vocabulary calls promoted_from_partial and whose projection
            # puts a completed plan back into pending.
            _take(completed, op.item_id)
            position = op.payload.get("position")
            index = len(pending) if position is None else int(position) - 1
            pending.insert(index, item)
        elif op.op == OP_MOVE:
            item = _take(pending, op.item_id)
            if item is None:
                raise CorruptLogError(
                    f"operation {op.seq} moves {op.item_id!r}, which is not pending"
                )
            pending.insert(int(op.payload["position"]) - 1, item)
        elif op.op == OP_DEQUEUE:
            item = _take(pending, op.item_id)
            if item is None:
                raise CorruptLogError(
                    f"operation {op.seq} dequeues {op.item_id!r}, which is not pending"
                )
            in_progress.append(
                replace(item, instance=_optional_str(op.payload.get("instance")))
            )
        elif op.op == OP_COMPLETE:
            item = _take(in_progress, op.item_id) or _take(pending, op.item_id)
            if item is None:
                raise CorruptLogError(
                    f"operation {op.seq} completes {op.item_id!r}, which the queue "
                    "does not hold"
                )
            completed.append(item)
 
    _dbg(
        "OC_CHAT.MAIN",
        func,
        f"{len(ops)} op(s) -> {len(pending)}p/{len(in_progress)}i/{len(completed)}c",
        3,
    )
    return QueueState(
        pending=tuple(pending),
        in_progress=tuple(in_progress),
        completed=tuple(completed),
    )
 
 
# ── The projection ───────────────────────────────────────────────────────
 
 
def _annotation(item: WorkItem) -> str:
    """The bracketed ``[k=v, k=v]`` suffix for *item*, or ``""``."""
    parts = [
        f"{key}={value}"
        for key, value in (("lane", item.lane), ("instance", item.instance))
        if value
    ]
    return f" [{', '.join(parts)}]" if parts else ""
 
 
def _bullets(items: tuple[WorkItem, ...]) -> list[str]:
    if not items:
        return [EMPTY_MARK]
    return [f"- {item.item_id}{_annotation(item)}" for item in items]
 
 
def _canonical_text(state: QueueState) -> str:
    """A fresh projection, written when there is no file to splice into."""
    lines: list[str] = [
        MARKER,
        "",
        "# Work queue",
        "",
        (
            "The view. The truth is the append-only operation log beside this "
            f"file, `{OPS_FILENAME}`. Edit this file by hand if you like: "
            "`queue_manage reconcile` reports the difference and "
            "`reconcile --apply` records it as operations. Every line outside "
            "the three tracked bullet lists is preserved verbatim."
        ),
        "",
    ]
    for title, attribute in SECTIONS:
        lines.append(f"## {title}")
        lines.append("")
        lines.extend(_bullets(getattr(state, attribute)))
        lines.append("")
    return "\n".join(lines).rstrip() + "\n"
 
 
def _splice_into_existing(text: str, state: QueueState) -> str:
    """Replace every tracked bullet list inside *text* with fresh content.
 
    Lifted in shape from the retired ``agent_queue._splice_into_existing``
    with the letters removed. Lines outside any tracked section --
    preamble, freeform paragraphs, unrelated H2s -- are preserved
    verbatim, and the marker is inserted at the top when absent.
    """
    func = "_splice_into_existing"
    out_lines: list[str] = []
    skip_existing = False
 
    for line in text.splitlines():
        match = _SECTION_RE.match(line)
        if match:
            attribute = _attribute_for(match.group("name"))
            out_lines.append(line)
            out_lines.append("")
            out_lines.extend(_bullets(getattr(state, attribute)))
            out_lines.append("")
            skip_existing = True
            continue
        if skip_existing:
            # Inside a tracked section: drop the OLD bullets until any
            # header-shaped line ends the tracked region, then fall
            # through so that terminating header is emitted.
            if line.startswith("## ") or line.startswith("# "):
                skip_existing = False
            else:
                continue
        out_lines.append(line)
 
    if MARKER not in "\n".join(out_lines):
        out_lines.insert(0, "")
        out_lines.insert(0, MARKER)
        _dbg("OC_CHAT.MAIN", func, "marker injected on save", 3)
 
    rendered = "\n".join(out_lines).rstrip()
    tail: list[str] = []
    for title, attribute in SECTIONS:
        if not re.search(rf"^##\s+{re.escape(title)}\s*$", rendered, re.I | re.M):
            tail.append(f"## {title}")
            tail.append("")
            tail.extend(_bullets(getattr(state, attribute)))
            tail.append("")
    if tail:
        rendered = rendered + "\n\n" + "\n".join(tail).rstrip()
        _dbg("OC_CHAT.MAIN", func, f"appended {len(tail)} line(s) for missing sections", 3)
    return rendered.rstrip() + "\n"
 
 
def _attribute_for(title: str) -> str:
    """The :class:`QueueState` field a tracked H2 title holds."""
    lowered = title.strip().lower()
    for section_title, attribute in SECTIONS:
        if section_title.lower() == lowered:
            return attribute
    raise CorruptLogError(f"unknown tracked section {title!r}")
 
 
def render_projection(state: QueueState, existing: str | None) -> str:
    """The projection text for *state*, spliced into *existing* when given."""
    if existing is None:
        return _canonical_text(state)
    return _splice_into_existing(existing, state)
 
 
def parse_projection(text: str) -> QueueState:
    """Parse a projection back into a state.
 
    Regex-anchored on the three tracked H2 sections, exactly as the
    retired parser was anchored on its eight. A bullet under any other
    heading is not a work item and is ignored.
    """
    func = "parse_projection"
    buckets: dict[str, list[WorkItem]] = {
        attribute: [] for _title, attribute in SECTIONS
    }
    current: str | None = None
    for line in text.splitlines():
        match = _SECTION_RE.match(line)
        if match:
            current = _attribute_for(match.group("name"))
            continue
        if line.startswith("## ") or line.startswith("# "):
            current = None
            continue
        bullet = _BULLET_RE.match(line)
        if bullet is None or current is None:
            continue
        buckets[current].append(
            _item_from_annotations(bullet.group("item_id"), bullet.group("annotations"))
        )
    _dbg(
        "OC_CHAT.MAIN",
        func,
        ", ".join(f"{name}: {len(items)}" for name, items in buckets.items()),
        4,
    )
    return QueueState(
        pending=tuple(buckets["pending"]),
        in_progress=tuple(buckets["in_progress"]),
        completed=tuple(buckets["completed"]),
    )
 
 
def _item_from_annotations(item_id: str, annotations: str | None) -> WorkItem:
    """A work item from a bullet's id and its ``[k=v, k=v]`` suffix."""
    values: dict[str, str] = {}
    for part in (annotations or "").split(","):
        key, separator, value = part.partition("=")
        if separator:
            values[key.strip()] = value.strip()
    return WorkItem(
        kind=KIND_PLAN,
        item_id=item_id,
        lane=values.get("lane") or None,
        instance=values.get("instance") or None,
    )
 
 
def _windows_safe_replace(src: Path, dst: Path) -> None:
    """Atomically replace *dst* with *src*; tolerate Windows handle/read-only quirks.
 
    Windows `os.replace` fails with `PermissionError` when the target is
    read-only or briefly held by another reader (antivirus, indexer,
    pytest cleanup). Clear the read-only attribute and retry up to twice
    with short backoff. Atomicity is preserved on every attempt.
    """
    # LIFTED VERBATIM from e1b5c716:oc_chat/oc_chat/core/agent_queue.py
    # (plan 258 P2). No dispatch-layer library module carries a
    # temp-then-replace with read-only recovery; plan_manage's
    # _atomic_write is script-private and has none.
    for attempt in range(3):
        try:
            src.replace(dst)
            return
        except PermissionError:
            if dst.exists():
                try:
                    dst.chmod(stat.S_IWRITE | stat.S_IREAD)
                except OSError:
                    pass
            if attempt == 2:
                raise
            time.sleep(0.05 * (attempt + 1))
 
 
# ── Drift ────────────────────────────────────────────────────────────────
 
 
@dataclass(frozen=True)
class QueueDrift:
    """One difference between the fold of the log and the parsed view."""
 
    #: ``added`` / ``advanced`` / ``moved`` / ``removed``.
    kind: str
    item_id: str
    #: What ``--apply`` would record, in words.
    detail: str
    #: ``False`` for ``removed``: the contract has no remove operation,
    #: so a hand deletion is drift the log cannot absorb.
    fixable: bool = True
 
 
# ── The backend ──────────────────────────────────────────────────────────
 
 
class FileQueueBackend:
    """The default backend: an append-only log and a Markdown view."""
 
    def __init__(self, paths: QueuePaths | None = None, *, actor: str = "operator") -> None:
        func = "FileQueueBackend.__init__"
        self.paths = paths if paths is not None else QueuePaths()
        self.actor = actor
        _dbg("OC_CHAT.MAIN", func, f"over {self.paths.queue_dir}", 2)
 
    # -- reads ------------------------------------------------------------
 
    def status(self) -> QueueState:
        """The three buckets, folded out of the log."""
        return fold(read_ops(self.paths.ops_path))
 
    def history(self, item_id: str) -> tuple[QueueOp, ...]:
        """Every recorded operation about *item_id*, oldest first."""
        return tuple(
            op for op in read_ops(self.paths.ops_path) if op.item_id == item_id
        )
 
    # -- writes -----------------------------------------------------------
 
    def enqueue(self, item: WorkItem, position: int | None = None) -> QueueState:
        """Put *item* into pending at *position*, or at the end.
 
        A COMPLETED item may be enqueued again: the item comes back out
        of completed and the log shows both runs. That is the motion
        the registry's own enqueue vocabulary calls
        ``promoted_from_partial`` / ``restored_from_aborted_round``, and
        whose projection puts a completed plan back into ``pending``.
        The retired ``add_task`` refused it because a slot's completed
        LIST was that model's only record; here the log is the record,
        so nothing is lost by allowing it.
 
        Raises:
            DuplicateItemError: VALIDATION_ERROR (L3) -- the item is
                already pending or already in progress.
            BadPositionError: VALIDATION_ERROR (L3) -- *position* is
                not a 1-based index into the pending list.
        """
        func = "FileQueueBackend.enqueue"
        state = self.status()
        held = self._bucket_of(state, item.item_id)
        if held is not None and held != "completed":
            _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: duplicate {item.item_id}", 3)
            raise DuplicateItemError(
                f"{item.item_id!r} is already {held} in this queue"
            )
        index = self._checked_index(position, len(state.pending) + 1)
        payload: dict[str, object] = {
            "kind": item.kind,
            "lane": item.lane,
            "instance": item.instance,
            "note": item.note,
            "position": index,
        }
        return self._record(OP_ENQUEUE, item.item_id, payload)
 
    def move(self, item_id: str, position: int, *, reason: str) -> QueueState:
        """Move a pending item to *position*, for a stated *reason*."""
        func = "FileQueueBackend.move"
        state = self.status()
        current = self._pending_index(state, item_id)
        index = self._checked_index(position, len(state.pending))
        _dbg("OC_CHAT.MAIN", func, f"{item_id}: {current} -> {index}", 3)
        return self._record(
            OP_MOVE,
            item_id,
            {"position": index, "from_position": current, "reason": reason},
        )
 
    def dequeue(self, item_id: str, *, instance: str, reason: str) -> QueueState:
        """Take an item out of pending into progress, naming *instance*."""
        state = self.status()
        self._pending_index(state, item_id)
        return self._record(
            OP_DEQUEUE, item_id, {"instance": instance, "reason": reason}
        )
 
    def complete(
        self,
        item_id: str,
        *,
        reason: str,
        round_no: int | None = None,
        merge_commit_sha: str | None = None,
    ) -> QueueState:
        """Finish an item, for a stated *reason*.
 
        Idempotent: completing an already-completed item records
        nothing and returns the state unchanged, which is what the
        retired ``complete_task`` did when it refused to append a plan
        to a completed list twice.
 
        ``round_no`` and ``merge_commit_sha`` are the REGISTRY backend's
        demands (``queue_complete`` requires both). They are accepted
        here so one call site drives both backends; this backend
        records them in the op payload, because the log is the truth,
        and derives no state from them.
 
        Raises:
            UnknownItemError: VALIDATION_ERROR (L3) -- the queue holds
                no such item.
        """
        func = "FileQueueBackend.complete"
        state = self.status()
        held = self._bucket_of(state, item_id)
        if held == "completed":
            _dbg("OC_CHAT.MAIN", func, f"{item_id} already completed; no-op", 2)
            return state
        if held is None:
            raise UnknownItemError(
                f"{item_id!r} is not pending or in progress in this queue"
            )
        payload: dict[str, object] = {"reason": reason}
        if round_no is not None:
            payload["round_no"] = round_no
        if merge_commit_sha is not None:
            payload["merge_commit_sha"] = merge_commit_sha
        return self._record(OP_COMPLETE, item_id, payload)
 
    # -- the projection ---------------------------------------------------
 
    def write_projection(self) -> bool:
        """Render the fold into the projection. ``True`` when it wrote.
 
        A re-render of unchanged state writes NOTHING: no temp file, no
        replace, so the bytes and the mtime both stand still.
        """
        func = "FileQueueBackend.write_projection"
        path = self.paths.projection_path
        try:
            existing: str | None = path.read_text(encoding="utf-8")
        except FileNotFoundError:
            existing = None
        text = render_projection(self.status(), existing)
        if existing is not None and text == existing:
            _dbg("OC_CHAT.MAIN", func, f"{path.name} unchanged; wrote nothing", 2)
            return False
 
        path.parent.mkdir(parents=True, exist_ok=True)
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            newline="\n",
            dir=path.parent,
            prefix=".work_queue_",
            suffix=".tmp",
            delete=False,
        ) as tmp:
            tmp.write(text)
            tmp_path = Path(tmp.name)
        _windows_safe_replace(tmp_path, path)
        _dbg("OC_CHAT.MAIN", func, f"wrote {path} ({len(text)} bytes)", 2)
        return True
 
    def reconcile(self, *, apply: bool = False) -> tuple[QueueDrift, ...]:
        """Compare the fold with the parsed projection; report the difference.
 
        Dry-run by default: it writes nothing at all, neither file. With
        *apply* it records each fixable difference as an operation with
        ``actor="reconcile"`` -- so a hand edit ends up in the log like
        every other change -- and then re-renders the projection from
        the new fold.
 
        A hand DELETION is reported and never applied: the contract has
        no remove operation, and inventing one to absorb a deletion
        would make the log lossy in exactly the way it exists to avoid.
        ``--apply`` still re-renders afterwards, so the view comes back
        into agreement with the truth and the operator has been told
        which edit did not survive.
 
        Returns what the DRY RUN would have reported, whether or not
        *apply* was set, so a caller can print one list either way.
        """
        func = "FileQueueBackend.reconcile"
        try:
            text = self.paths.projection_path.read_text(encoding="utf-8")
        except FileNotFoundError:
            _dbg("OC_CHAT.MAIN", func, "no projection; nothing to compare", 2)
            return ()
        parsed = parse_projection(text)
        drifts = self._drift(self.status(), parsed)
        _dbg("OC_CHAT.MAIN", func, f"{len(drifts)} drift(s); apply={apply}", 2)
        if not apply or not drifts:
            return drifts
 
        # One fixable drift per pass, because applying one changes the
        # positions the next one is measured against. The bound is the
        # number of differences there were to begin with.
        for _pass in range(len(drifts) + 1):
            remaining = [d for d in self._drift(self.status(), parsed) if d.fixable]
            if not remaining:
                break
            self._apply_drift(remaining[0], parsed)
        self.write_projection()
        return drifts
 
    # -- internals --------------------------------------------------------
 
    def _record(
        self, op_name: str, item_id: str, payload: dict[str, object]
    ) -> QueueState:
        """Append one operation, re-render the view, return the new state."""
        ops = read_ops(self.paths.ops_path)
        append_op(
            self.paths.ops_path,
            QueueOp(
                seq=len(ops) + 1,
                at=_now_iso(),
                op=op_name,
                item_id=item_id,
                payload=payload,
                actor=self.actor,
            ),
        )
        self.write_projection()
        return self.status()
 
    @staticmethod
    def _bucket_of(state: QueueState, item_id: str) -> str | None:
        """Which bucket holds *item_id*, in words, or ``None``."""
        for label, attribute in (
            ("pending", "pending"),
            ("in progress", "in_progress"),
            ("completed", "completed"),
        ):
            if any(item.item_id == item_id for item in getattr(state, attribute)):
                return label
        return None
 
    def _pending_index(self, state: QueueState, item_id: str) -> int:
        """The 1-based position of *item_id* in pending, or a refusal."""
        for index, item in enumerate(state.pending, start=1):
            if item.item_id == item_id:
                return index
        held = self._bucket_of(state, item_id)
        where = f"it is {held}" if held else "the queue does not hold it"
        raise UnknownItemError(f"{item_id!r} is not pending: {where}")
 
    @staticmethod
    def _checked_index(position: int | None, upper: int) -> int:
        """*position* as a 1-based index in ``1..upper``, or the end."""
        if position is None:
            return max(upper, 1)
        if not 1 <= position <= max(upper, 1):
            raise BadPositionError(
                f"position {position} is outside 1..{max(upper, 1)}"
            )
        return position
 
    @staticmethod
    def _drift(folded: QueueState, parsed: QueueState) -> tuple[QueueDrift, ...]:
        """Every difference between the fold and the parsed view."""
        drifts: list[QueueDrift] = []
        folded_where = {
            item.item_id: attribute
            for _title, attribute in SECTIONS
            for item in getattr(folded, attribute)
        }
        parsed_where = {
            item.item_id: attribute
            for _title, attribute in SECTIONS
            for item in getattr(parsed, attribute)
        }
        for item_id, attribute in parsed_where.items():
            if item_id not in folded_where:
                drifts.append(
                    QueueDrift(
                        kind="added",
                        item_id=item_id,
                        detail=f"hand-added to {attribute}",
                    )
                )
            elif folded_where[item_id] != attribute:
                drifts.append(
                    QueueDrift(
                        kind="advanced",
                        item_id=item_id,
                        detail=f"{folded_where[item_id]} -> {attribute} by hand",
                    )
                )
        for item_id in folded_where:
            if item_id not in parsed_where:
                drifts.append(
                    QueueDrift(
                        kind="removed",
                        item_id=item_id,
                        detail="deleted from the projection by hand; the log has "
                        "no remove operation, so this is reported, not applied",
                        fixable=False,
                    )
                )
        # Reordering, judged over the items BOTH sides agree are pending:
        # an item that was added or advanced is already reported above and
        # would otherwise be counted twice.
        shared = [
            item.item_id
            for item in parsed.pending
            if folded_where.get(item.item_id) == "pending"
        ]
        folded_order = [
            item.item_id for item in folded.pending if item.item_id in set(shared)
        ]
        for position, item_id in enumerate(shared, start=1):
            if folded_order.index(item_id) != position - 1:
                drifts.append(
                    QueueDrift(
                        kind="moved",
                        item_id=item_id,
                        detail=f"reordered to position {position} by hand",
                    )
                )
                break
        return tuple(drifts)
 
    def _apply_drift(self, drift: QueueDrift, parsed: QueueState) -> None:
        """Record one fixable drift as an operation attributed to reconcile."""
        func = "FileQueueBackend._apply_drift"
        if not drift.fixable:
            _dbg("OC_CHAT.MAIN", func, f"{drift.item_id}: {drift.kind} not applied", 3)
            return
        backend = FileQueueBackend(self.paths, actor=_ACTOR_RECONCILE)
        item, bucket, position = _locate(parsed, drift.item_id)
        if drift.kind == "added":
            backend.enqueue(
                item, position=position if bucket == "pending" else None
            )
            if bucket == "in_progress":
                backend.dequeue(
                    drift.item_id,
                    instance=item.instance or "unknown",
                    reason="reconciled from the projection",
                )
            elif bucket == "completed":
                backend.complete(
                    drift.item_id, reason="reconciled from the projection"
                )
        elif drift.kind == "advanced":
            if bucket == "in_progress":
                backend.dequeue(
                    drift.item_id,
                    instance=item.instance or "unknown",
                    reason="reconciled from the projection",
                )
            elif bucket == "completed":
                backend.complete(
                    drift.item_id, reason="reconciled from the projection"
                )
        elif drift.kind == "moved":
            backend.move(
                drift.item_id,
                position,
                reason="reconciled from the projection",
            )
 
 
def _locate(state: QueueState, item_id: str) -> tuple[WorkItem, str, int]:
    """*item_id*'s item, bucket name and 1-based position inside it."""
    for _title, attribute in SECTIONS:
        for position, item in enumerate(getattr(state, attribute), start=1):
            if item.item_id == item_id:
                return item, attribute, position
    raise UnknownItemError(f"{item_id!r} is not in the parsed projection")
 
 
__all__ = [
    "DEFAULT_QUEUE_DIR",
    "EMPTY_MARK",
    "MARKER",
    "OPS_FILENAME",
    "PROJECTION_FILENAME",
    "SECTIONS",
    "CorruptLogError",
    "FileQueueBackend",
    "QueueDrift",
    "QueuePaths",
    "append_op",
    "fold",
    "parse_projection",
    "read_ops",
    "render_projection",
]