refactor(python): simplify plugin mechanics with standard idioms

docs/specs/python-quality.md

@@ -49,6 +49,11 @@ Use behavioral and differential checks to preserve the required interface while
 changing the internals. High coverage is a guard for simplification, not a reason
 to keep redundant code or mirror the old implementation in new tests.
 
+Behavior-preserving simplifications retain literal argument values, diagnostic
+precedence, serialized bytes, and resource lifetimes. Standard-library replacements
+must preserve those contracts across the package's declared Python versions;
+resource cleanup must not implicitly change transaction or publication semantics.
+
 ## Tests and coverage
 
 Each package keeps fast unit tests for local behavior and integration tests for

plugins/capability/darrow-discovery/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "darrow-discovery",
   "description": "Explicit grilling, feature discovery, and implementation planning",
-  "version": "0.2.2",
+  "version": "0.2.3",
   "license": "BUSL-1.1",
   "author": {
     "name": "Björn Rochel",

plugins/capability/darrow-discovery/.codex-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "darrow-discovery",
-  "version": "0.2.2",
+  "version": "0.2.3",
   "description": "Explicit grilling, feature discovery, and implementation planning",
   "author": {
     "name": "Björn Rochel",

plugins/capability/darrow-discovery/skills/plan-implementation/backend/src/darrow_discovery/frontier.py

@@ -49,8 +49,7 @@ def parse(arguments: Sequence[str]) -> Frontier:
     options: list[str] = []
     if len(arguments) % 2:
         raise UsageError
-    for index in range(0, len(arguments), 2):
-        flag, value = arguments[index : index + 2]
+    for flag, value in zip(arguments[::2], arguments[1::2], strict=True):
         if flag == "--option":
             options.append(value)
         elif flag in FIELD_FLAGS:

plugins/capability/darrow-git/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "darrow-git",
   "description": "Intent-triggered Git skills for branches, commits, PRs, and candidate-bound PR evidence",
-  "version": "0.8.2",
+  "version": "0.8.3",
   "hooks": "./.claude-plugin/hooks.json",
   "license": "BUSL-1.1"
 }

plugins/capability/darrow-git/.codex-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "darrow-git",
-  "version": "0.8.2",
+  "version": "0.8.3",
   "description": "Intent-triggered Git skills for branches, commits, PRs, and candidate-bound PR evidence",
   "author": {
     "name": "Björn Rochel"

plugins/capability/darrow-git/backend/src/darrow_git/evidence_identity.py

@@ -10,6 +10,7 @@
 PLACEHOLDER = re.compile(
     r"darrow-evidence-attachment-[0-9]+\.(png|jpg|jpeg|gif|webp|svg|mp4|mov|webm)"
 )
+TSV_ESCAPES = str.maketrans({"\\": "\\\\", "\r": "\\r", "\t": "\\t", "\n": "\\n"})
 
 
 @dataclass(frozen=True)
@@ -29,12 +30,7 @@ class Reconciliation:
 
 
 def escape_tsv(text: str) -> str:
-    return (
-        text.replace("\\", "\\\\")
-        .replace("\r", "\\r")
-        .replace("\t", "\\t")
-        .replace("\n", "\\n")
-    )
+    return text.translate(TSV_ESCAPES)
 
 
 def build(repository: str, number: str, expected: str, snapshot: Snapshot) -> Identity:

plugins/capability/darrow-observability-langfuse/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "darrow-observability-langfuse",
   "description": "Export Codex rollout turns to Langfuse with attribution epochs",
-  "version": "0.5.1",
+  "version": "0.5.2",
   "license": "BUSL-1.1",
   "author": {
     "name": "Björn Rochel",

plugins/capability/darrow-observability-langfuse/.codex-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "darrow-observability-langfuse",
-  "version": "0.5.1",
+  "version": "0.5.2",
   "description": "Export Codex rollout turns to Langfuse with attribution epochs",
   "author": {
     "name": "Björn Rochel",

plugins/capability/darrow-observability-langfuse/backend/src/darrow_observability_langfuse/capture.py

@@ -8,7 +8,7 @@
 import os
 import sqlite3
 from collections.abc import Callable, Iterator
-from contextlib import contextmanager
+from contextlib import closing, contextmanager
 from dataclasses import replace
 from pathlib import Path
 from typing import Any, cast
@@ -67,9 +67,8 @@ def database(rollout: Path) -> Iterator[sqlite3.Connection]:
     path = database_path(rollout)
     descriptor = os.open(path, os.O_CREAT | os.O_RDWR, 0o600)
     os.close(descriptor)
-    connection = sqlite3.connect(path, timeout=30, isolation_level=None)
-    connection.row_factory = sqlite3.Row
-    try:
+    with closing(sqlite3.connect(path, timeout=30, isolation_level=None)) as connection:
+        connection.row_factory = sqlite3.Row
         connection.execute("PRAGMA synchronous=FULL")
         connection.execute("PRAGMA foreign_keys=ON")
         connection.executescript("""
@@ -92,8 +91,6 @@ def database(rollout: Path) -> Iterator[sqlite3.Connection]:
             CREATE INDEX IF NOT EXISTS captured_turns_pending ON captured_turns(finalized);
         """)
         yield connection
-    finally:
-        connection.close()
 
 
 def _get(connection: sqlite3.Connection, key: str, default: Any = None) -> Any:
@@ -202,17 +199,14 @@ def load_capture_snapshots(rollout: Path) -> dict[str, dict[str, Any]]:
     path = database_path(rollout)
     if not path.exists():
         return {}
-    connection = sqlite3.connect(f"{path.as_uri()}?mode=ro", uri=True)
-    try:
+    with closing(sqlite3.connect(f"{path.as_uri()}?mode=ro", uri=True)) as connection:
         return cast(
             "dict[str, dict[str, Any]]",
             {
                 row[0]: json.loads(row[1])
                 for row in connection.execute("SELECT * FROM snapshots")
             },
         )
-    finally:
-        connection.close()
 
 
 def capture(

plugins/capability/darrow-observability-langfuse/backend/src/darrow_observability_langfuse/delivery.py

@@ -9,7 +9,7 @@
 import sqlite3
 import time
 from collections.abc import Callable, Iterator
-from contextlib import contextmanager
+from contextlib import closing, contextmanager
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -25,18 +25,16 @@ def await_capture(rollout: Path, turn_id: str, timeout: float = 45) -> bool:
     while time.monotonic() < deadline:
         path = database_path(rollout)
         if path.exists():
-            connection = sqlite3.connect(
-                f"{path.as_uri()}?mode=ro", uri=True, timeout=1
-            )
-            try:
-                if connection.execute(
-                    "SELECT 1 FROM receipts WHERE turn_id=?", (turn_id,)
-                ).fetchone():
-                    return True
-            except sqlite3.OperationalError:
-                pass  # Capture may still be creating the schema.
-            finally:
-                connection.close()
+            with closing(
+                sqlite3.connect(f"{path.as_uri()}?mode=ro", uri=True, timeout=1)
+            ) as connection:
+                try:
+                    if connection.execute(
+                        "SELECT 1 FROM receipts WHERE turn_id=?", (turn_id,)
+                    ).fetchone():
+                        return True
+                except sqlite3.OperationalError:
+                    pass  # Capture may still be creating the schema.
         time.sleep(0.05)
     return False
 
@@ -160,8 +158,6 @@ def _attempt(self, batch: _Batch) -> None:
         except DeliveryError as error:
             self._record_failure(batch, error)
             self.failure = error
-        except BaseException:
-            raise  # Cancellation retains the durable uncertain state.
         else:
             self._acknowledge(batch)
             self.acknowledged += len(batch.rows)

plugins/capability/darrow-observability-langfuse/backend/src/darrow_observability_langfuse/lifecycle.py

@@ -8,7 +8,7 @@
 import sqlite3
 import tempfile
 from collections.abc import Iterator
-from contextlib import suppress
+from contextlib import closing, suppress
 from pathlib import Path
 from typing import Any
 
@@ -58,17 +58,16 @@ def _validate_legacy_context(rollout: Path, context: dict[str, Any]) -> None:
     previous = Path(f"{rollout}.darrow-langfuse.sqlite3")
     if not previous.exists():
         return
-    connection = sqlite3.connect(f"{previous.as_uri()}?mode=ro", uri=True, timeout=0)
-    try:
+    with closing(
+        sqlite3.connect(f"{previous.as_uri()}?mode=ro", uri=True, timeout=0)
+    ) as connection:
         binding = connection.execute(
             "SELECT value FROM state WHERE key='delivery_context'"
         ).fetchone()
         if binding:
             require_context(json.loads(binding[0]), context)
         elif _database_has_evidence(connection):
             require_context(None, context)
-    finally:
-        connection.close()
 
 
 def _database_has_evidence(connection: sqlite3.Connection) -> bool:

plugins/capability/darrow-review/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "darrow-review",
   "description": "Read-only comprehensive code review and fix-scoped repair verification",
-  "version": "0.5.2",
+  "version": "0.5.3",
   "license": "BUSL-1.1",
   "author": {
     "name": "Björn Rochel",

plugins/capability/darrow-review/.codex-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "darrow-review",
-  "version": "0.5.2",
+  "version": "0.5.3",
   "description": "Read-only comprehensive code review and fix-scoped repair verification",
   "author": {
     "name": "Björn Rochel",

plugins/capability/darrow-review/backend/src/darrow_review/provider.py

@@ -51,8 +51,8 @@ def json_object(text: str) -> dict[str, object]:
         )
     except (ValueError, RecursionError) as exc:
         raise ReviewError(f"invalid JSON: {exc}") from exc
-    require(isinstance(value, dict), "JSON root must be an object")
-    assert isinstance(value, dict)
+    if not isinstance(value, dict):
+        raise ReviewError("JSON root must be an object")
     return {str(key): item for key, item in value.items()}
 
 
@@ -64,18 +64,21 @@ def assistant_observation(
         f"assistant record has unexpected agentId on line {line}",
     )
     message = row.get("message")
-    require(isinstance(message, dict), f"assistant record lacks message on line {line}")
-    assert isinstance(message, dict)
+    if not isinstance(message, dict):
+        raise ReviewError(f"assistant record lacks message on line {line}")
     require(
         message.get("role") == "assistant",
         f"assistant record has unexpected message.role on line {line}",
     )
     model, effort = message.get("model"), row.get("effort")
-    require(
-        isinstance(model, str) and isinstance(effort, str) and model and effort,
-        f"assistant record lacks model or effort on line {line}",
-    )
-    return str(model), str(effort)
+    if (
+        not isinstance(model, str)
+        or not isinstance(effort, str)
+        or not model
+        or not effort
+    ):
+        raise ReviewError(f"assistant record lacks model or effort on line {line}")
+    return model, effort
 
 
 def observe_transcript(path: Path, agent: str) -> tuple[str, str]:

plugins/capability/darrow-review/backend/src/darrow_review/report.py

@@ -2,14 +2,15 @@
 
 from __future__ import annotations
 
+from collections import Counter
+
 from . import report_templates as templates
 from .common import read_text
 from .records import Records, validate_result
 from .verification import validate_verification
 
-
-def escape(value: str) -> str:
-    entities = {
+ESCAPES = str.maketrans(
+    {
         "&": "&",
         "<": "&lt;",
         ">": "&gt;",
@@ -22,7 +23,11 @@ def escape(value: str) -> str:
         "|": "&#124;",
         "\\": "&#92;",
     }
-    return "".join(entities.get(char, char) for char in value)
+)
+
+
+def escape(value: str) -> str:
+    return value.translate(ESCAPES)
 
 
 def guidance(fields: list[str], prefix: str = "") -> str:
@@ -163,15 +168,15 @@ def closed_findings(result: Records) -> str:
 
 
 def verification(result: Records) -> str:
-    statuses = [row[2] for row in result.get("attempt")]
+    statuses = Counter(row[2] for row in result.get("attempt"))
     gaps = result.get("evidence_gap")
     return templates.VERIFICATION.substitute(
         title=result.value("outcome").upper(),
         outcome=escape(result.value("outcome")),
         total=len(result.get("original_finding")),
-        resolved=statuses.count("resolved"),
-        unresolved=statuses.count("unresolved"),
-        blocked=statuses.count("blocked"),
+        resolved=statuses["resolved"],
+        unresolved=statuses["unresolved"],
+        blocked=statuses["blocked"],
         regression_count=len(result.get("regression")),
         attempted_findings=attempted_findings(result),
         regressions=repair_regressions(result),

plugins/capability/darrow-review/backend/src/darrow_review/routing.py

@@ -8,6 +8,7 @@
 from pathlib import Path
 
 from .common import (
+    ReviewError,
     new_record,
     package_root,
     read_text,
@@ -70,11 +71,8 @@ def body(self) -> str:
 
 
 def parse_reviewer(value: object, source: str) -> Route:
-    require(
-        isinstance(value, dict),
-        "invalid reviewer configuration: reviewer must be an object",
-    )
-    assert isinstance(value, dict)
+    if not isinstance(value, dict):
+        raise ReviewError("invalid reviewer configuration: reviewer must be an object")
     require(
         set(value) == {"host", "harness", "provider", "model", "effort"},
         "invalid reviewer configuration: each reviewer must contain exactly five fields",
@@ -108,11 +106,8 @@ def catalog(path: Path, source: str) -> dict[str, Route]:
         "invalid reviewer configuration: unknown root field",
     )
     values = config.get("reviewers", [])
-    require(
-        isinstance(values, list),
-        "invalid reviewer configuration: reviewers must be an array",
-    )
-    assert isinstance(values, list)
+    if not isinstance(values, list):
+        raise ReviewError("invalid reviewer configuration: reviewers must be an array")
     require(
         values or source == "repository",
         "invalid reviewer configuration: bundled reviewers array must not be empty",

plugins/foundation/darrow-decisions/.claude-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "darrow-decisions",
   "description": "Capture and query durable repository decisions at their canonical scope",
-  "version": "0.2.1",
+  "version": "0.2.2",
   "license": "BUSL-1.1"
 }

plugins/foundation/darrow-decisions/.codex-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "darrow-decisions",
-  "version": "0.2.1",
+  "version": "0.2.2",
   "description": "Capture and query durable repository decisions at their canonical scope",
   "author": {
     "name": "Björn Rochel",

plugins/foundation/darrow-decisions/backend/src/darrow_decisions/catalog.py

@@ -2,6 +2,7 @@
 
 import os
 import tempfile
+from dataclasses import replace
 from pathlib import Path
 
 from .catalog_format import FORMAT, CatalogRow, checksum, render
@@ -29,7 +30,7 @@ def make_rows(root: Path, records: list[Record]) -> list[CatalogRow]:
     for record in records:
         row = CatalogRow(record, fingerprint(root, record.path), 0, 0)
         crc, size = checksum(row.signature())
-        rows.append(CatalogRow(record, row.fingerprint, crc, size))
+        rows.append(replace(row, crc=crc, size=size))
     return rows