refactor(queue): restore koru.queue lower-layer import contract (ticket-474)

project/ticket-474/README.md

@@ -0,0 +1,31 @@
+# Ticket 474: Restore koru.queue lower-layer import contract (#731)
+
+- **ID**: ticket-474
+- **Owner**: unresolved:human
+- **Status**: IN_PROGRESS
+- **Workflow state**: EDIT
+- **Created**: 2026-10-19
+
+## Goal and scope
+
+Closes semcod/koru#731. `ci.yml` on main had failed for 200+ consecutive runs at the
+"Import contracts" step: `koru.queue is a lower layer: no autonomy or daemon imports` was
+broken through three chains (queue.runner -> activity_log / planfile_queue.application ->
+autonomy.cycle_trace; queue.runners -> task_model_policy -> autonomy.verification_engine).
+
+The stdlib-only helpers those lower layers need move to neutral modules:
+`koru.ticket_logs` (find_ticket_dir, append_ticket_markdown_log,
+append_queue_ticket_markdown_log) and `koru.drive_budget` (per-ticket drive budget).
+`koru.autonomy.cycle_trace` and `koru.autonomy.verification_engine.budget` re-export them, so
+every existing import keeps working; the three lower-layer callers import the neutral modules.
+
+## Acceptance criteria
+
+- [x] AC-01: `lint-imports` (import-linter, `PYTHONPATH=src:packages/koruide/src`): 3 kept, 0 broken.
+- [x] AC-02: `tests/test_queue_layer_contract.py` (AST guard + re-export identity + budget
+  behaviour) and the existing cycle_trace / queue log / retry-guard tests pass (24).
+
+## Tracking boundary
+
+This directory contains the minimal reviewed intent. Optional participant prose
+and raw command logs are not required delivery output.

project/ticket-474/intent.json

@@ -0,0 +1,95 @@
+{
+  "schema": "new-project.intent/v3",
+  "ticket": "ticket-474",
+  "summary": "Restore koru.queue lower-layer import contract (#731)",
+  "workstream": "application",
+  "classification": {
+    "kind": "SERVICE",
+    "priority": "P2",
+    "origin": "health"
+  },
+  "allowedPaths": [
+    "project/ticket-474/**",
+    "TODO.md",
+    "project/TICKETS.md",
+    "src/koru/activity_log.py",
+    "src/koru/autonomy/cycle_trace.py",
+    "src/koru/autonomy/verification_engine/budget.py",
+    "src/koru/bounded_contexts/planfile_queue/application.py",
+    "src/koru/drive_budget.py",
+    "src/koru/task_model_policy.py",
+    "src/koru/ticket_logs.py",
+    "tests/test_queue_layer_contract.py"
+  ],
+  "forbiddenPaths": [
+    "project/ticket-*/user-*.md"
+  ],
+  "stacks": [],
+  "dependsOn": [],
+  "conflictsWith": [],
+  "integrationTicket": null,
+  "delivery": {
+    "acceptedBaseSha": "542339cd8dfcae760a03de346db4616a0eba9f42",
+    "targetBranch": "main",
+    "outcome": "koru.queue no longer imports koru.autonomy; main CI import-contract step passes.",
+    "nonGoals": [
+      "No behaviour change; existing import paths are kept via re-exports."
+    ],
+    "complexity": "M",
+    "estimatedMinutes": 60,
+    "budgets": {
+      "maxImplementationFiles": 8,
+      "maxAffectedComponents": 1,
+      "maxPublicInterfaceChanges": 2,
+      "maxRuntimeDependencies": 0
+    },
+    "architecture": {
+      "status": "accepted",
+      "decision": "Move stdlib-only ticket-log and drive-budget helpers to neutral koru modules; autonomy modules re-export them.",
+      "components": [
+        {
+          "name": "queue-layering",
+          "paths": [
+            "src/koru/ticket_logs.py",
+            "src/koru/drive_budget.py",
+            "src/koru/autonomy/cycle_trace.py",
+            "src/koru/autonomy/verification_engine/budget.py",
+            "src/koru/activity_log.py",
+            "src/koru/bounded_contexts/planfile_queue/application.py",
+            "src/koru/task_model_policy.py",
+            "tests/test_queue_layer_contract.py"
+          ]
+        }
+      ],
+      "responsibilityChanges": false,
+      "interfaceChanges": [
+        "koru.ticket_logs",
+        "koru.drive_budget"
+      ],
+      "dataChanges": [],
+      "ui": {
+        "impact": "none",
+        "states": [],
+        "evidence": []
+      },
+      "rollback": "Revert the commit."
+    },
+    "runtimeDependencies": [],
+    "validation": [
+      {
+        "criterion": "AC-01",
+        "commands": [
+          "PYTHONPATH=src:packages/koruide/src lint-imports"
+        ],
+        "evidence": "project/ticket-474/README.md"
+      },
+      {
+        "criterion": "AC-02",
+        "commands": [
+          "PYTHONPATH=src python3 -m pytest -q -o addopts='' tests/test_queue_layer_contract.py tests/test_queue_log_markdown.py tests/test_cycle_trace.py tests/test_autonomous_retry_guard.py"
+        ],
+        "evidence": "project/ticket-474/README.md"
+      }
+    ]
+  }
+}

src/koru/activity_log.py

@@ -390,7 +390,7 @@ def _append_activity_to_ticket_markdown_log(
     target_ticket_dir: Path | None = None
     if ticket_id:
         try:
-            from koru.autonomy.cycle_trace import find_ticket_dir
+            from koru.ticket_logs import find_ticket_dir
             target_ticket_dir = find_ticket_dir(project_root, str(ticket_id))
         except ImportError:
             pass

src/koru/autonomy/cycle_trace.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import re
 from collections.abc import Callable
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -138,127 +137,11 @@ def record_decision_trace(
     append_ticket_markdown_log(project, waiting_ticket, record)
 
 
-def find_ticket_dir(project: Path, ticket_id: str) -> Path | None:
-    """Resolve only explicit Wellmanifest IDs; Planfile IDs are a separate namespace.
-
-    Never infer ownership from a shared numeric suffix or choose an arbitrary
-    slugged directory when several tickets match.
-    """
-    raw = ticket_id.strip().lower()
-    if not re.fullmatch(r"ticket-[0-9]+(?:--?[a-z0-9]+(?:-[a-z0-9]+)*)?", raw):
-        return None
-    project_dir = project / "project"
-    if not project_dir.is_dir():
-        return None
-    exact = project_dir / raw
-    if exact.is_dir() and not exact.is_symlink():
-        return exact
-    if not re.fullmatch(r"ticket-[0-9]+", raw):
-        return None
-    matches = [
-        path for path in project_dir.glob(f"{raw}-*")
-        if path.is_dir() and not path.is_symlink()
-    ]
-    return matches[0] if len(matches) == 1 else None
-
-
-def append_ticket_markdown_log(
-    project: Path,
-    waiting_ticket: str,
-    record: Any,
-) -> None:
-    """Append decision trace to project/ticket-*/koru.log.md with markdown codeblocks."""
-    ticket_dir = find_ticket_dir(project, waiting_ticket)
-    if not ticket_dir:
-        return
-
-    log_file = ticket_dir / "koru.log.md"
-    timestamp = getattr(record, "at", "")
-    cycle = getattr(record, "cycle", 0)
-
-    # Format trace as markdown with yaml codeblock
-    content_blocks = [
-        f"### Cycle {cycle} (`{timestamp}`)\n\n",
-        "```yaml\n",
-        f"uri: koru://cycle/{cycle}/decision/{record.action}\n",
-        f"observed: {record.observed}\n",
-        f"decided: {record.decided}\n",
-        f"action: {record.action}\n",
-        f"evidence: {record.evidence}\n",
-    ]
-    if getattr(record, "skip_because", ""):
-        content_blocks.append(f"because: {record.skip_because}\n")
-    if getattr(record, "next_step", ""):
-        content_blocks.append(f"next: {record.next_step}\n")
-    content_blocks.append("```\n\n")
-
-    # Add alternating NL/DSL trace block
-    content_blocks.extend([
-        "```yaml\n",
-        f'NL: "Cykl {cycle}: {record.decided} ({record.observed})"\n',
-        f'DSL: "action: {record.action} [evidence: {record.evidence}]"\n',
-        "```\n\n",
-    ])
-
-    try:
-        if not log_file.exists():
-            header = f"# Koru Autonomy Log: `{ticket_dir.name}`\n\n"
-            log_file.write_text(header + "".join(content_blocks), encoding="utf-8")
-        else:
-            with log_file.open("a", encoding="utf-8") as f:
-                f.write("".join(content_blocks))
-    except OSError:
-        pass
-
-
-def append_queue_ticket_markdown_log(
-    project: Path,
-    result: Any,
-) -> None:
-    """Append queue execution event to project/ticket-*/koru.log.md if project has wellmanifest layout."""
-    ticket_id = getattr(result, "ticket_id", None)
-    if not ticket_id:
-        return
-    ticket_dir = find_ticket_dir(project, str(ticket_id))
-    if not ticket_dir:
-        return
-
-    log_file = ticket_dir / "koru.log.md"
-    from datetime import UTC, datetime
-
-    timestamp = datetime.now(UTC).replace(microsecond=0).isoformat()
-    status = getattr(result, "status", "unknown")
-    executor = getattr(result, "executor_kind", "unknown")
-    exit_code = getattr(result, "exit_code", 0)
-    message = getattr(result, "message", "")
-
-    content_blocks = [
-        f"### Queue Run (`{timestamp}`)\n\n",
-        "```yaml\n",
-        f"uri: koru://queue/task/{ticket_id}/{status}\n",
-        f"ticket: {ticket_id}\n",
-        f"executor: {executor}\n",
-        f"status: {status}\n",
-        f"exit_code: {exit_code}\n",
-        "```\n\n",
-    ]
-    if message:
-        content_blocks.extend([
-            "```text\n",
-            f"{message}\n",
-            "```\n\n",
-        ])
-
-    try:
-        if not log_file.exists():
-            header = f"# Koru Autonomy Log: `{ticket_dir.name}`\n\n"
-            log_file.write_text(header + "".join(content_blocks), encoding="utf-8")
-        else:
-            with log_file.open("a", encoding="utf-8") as f:
-                f.write("".join(content_blocks))
-    except OSError:
-        pass
-
+from koru.ticket_logs import (  # noqa: E402 - re-exported for existing callers
+    append_queue_ticket_markdown_log,
+    append_ticket_markdown_log,
+    find_ticket_dir,
+)
 
 __all__ = [
     "append_queue_ticket_markdown_log",

src/koru/autonomy/verification_engine/budget.py

@@ -1,75 +1,9 @@
-"""Per-ticket drive budget persistence (failures and reserved attempts)."""
-
-import sqlite3
-from contextlib import closing
-from pathlib import Path
-
-
-def _drive_budget_db(project: Path) -> sqlite3.Connection:
-    path = project / ".planfile" / ".koru" / "drive-budget.sqlite"
-    path.parent.mkdir(parents=True, exist_ok=True)
-    db = sqlite3.connect(path, timeout=5)
-    try:
-        db.execute("CREATE TABLE IF NOT EXISTS failures (ticket TEXT PRIMARY KEY, count INTEGER NOT NULL)")
-        db.execute("CREATE TABLE IF NOT EXISTS attempts (ticket TEXT PRIMARY KEY, count INTEGER NOT NULL)")
-        return db
-    except sqlite3.Error:
-        db.close()
-        raise
-
-
-def record_unsuccessful_drive(project: Path, ticket: str) -> None:
-    """Persist failure count independently of waiting streak and loop restarts."""
-    if not ticket:
-        return
-    with closing(_drive_budget_db(project)) as db, db:
-        db.execute(
-            "INSERT INTO failures VALUES (?, 1) ON CONFLICT(ticket) DO UPDATE SET count=count+1",
-            (ticket,),
-        )
-
-
-def drive_budget_exhausted(project: Path, ticket: str, limit: int = 3) -> bool:
-    """Storage errors close admission; failures need an explicit ticket repair."""
-    if not ticket:
-        return False
-    try:
-        with closing(_drive_budget_db(project)) as db, db:
-            counts = _drive_budget_counts(db, ticket)
-        return max(counts) >= limit
-    except (OSError, sqlite3.Error):
-        return True
-
-
-def _drive_budget_counts(db: sqlite3.Connection, ticket: str) -> tuple[int, int]:
-    counts = []
-    for table in ("failures", "attempts"):
-        row = db.execute(f"SELECT count FROM {table} WHERE ticket=?", (ticket,)).fetchone()
-        counts.append(int(row[0]) if row else 0)
-    return counts[0], counts[1]
-
-
-def reserve_drive_attempt(project: Path, ticket: str, limit: int = 3) -> bool:
-    """Reserve before a paid shell call, atomically; crashes cannot erase attempts.
-
-    The protected finalizer owns completion. Unverified shell calls share the
-    admission budget with failed GUI drives; separate counts avoid double debit.
-    """
-    if not ticket:
-        return True
-    try:
-        with closing(_drive_budget_db(project)) as db, db:
-            db.execute("BEGIN IMMEDIATE")
-            if max(_drive_budget_counts(db, ticket)) >= limit:
-                return False
-            db.execute(
-                "INSERT INTO attempts VALUES (?, 1) ON CONFLICT(ticket) DO UPDATE SET count=count+1",
-                (ticket,),
-            )
-        return True
-    except (OSError, sqlite3.Error):
-        return False
-
-
-# ---------------------------------------------------------------------------
-# Verdict logic (pure heuristics, no LLM)
+"""Re-export of koru.drive_budget for existing callers (semcod/koru#731)."""
+
+from koru.drive_budget import (  # noqa: F401
+    _drive_budget_counts,
+    _drive_budget_db,
+    drive_budget_exhausted,
+    record_unsuccessful_drive,
+    reserve_drive_attempt,
+)

src/koru/bounded_contexts/planfile_queue/application.py

@@ -92,7 +92,7 @@ def run_next_task(self, command: RunNextPlanfileTaskCommand) -> QueueRunResult:
             payload=payload,
             aggregate_id=aggregate_id,
         )
-        from koru.autonomy.cycle_trace import append_queue_ticket_markdown_log
+        from koru.ticket_logs import append_queue_ticket_markdown_log
 
         append_queue_ticket_markdown_log(command.project, result)
         return result