feat(lessons): machine lessons-journal — additive table + observer leaf (ORCH-098)

Step 1 ("Foundation", F2) of the self-improvement epic: formalise free-text
"lessons" from memory/ into a machine-readable `lessons` table — the foundation
for the future retrospective agent (E2), the RICE prioritiser (E3) and Стрим.

- src/lessons.py: pure never-raise observer leaf (record/get/update/snapshot),
  kill-switch only, NO repo scope (observer-only; records about any repo incl.
  enduro; repo cut on the read side). Slug-convention constants.
- src/db.py: additive idempotent `lessons` table in init_db() (+3 indexes);
  nullable attribution columns from the start (NFR-6, _ensure_column forward-safe);
  helpers record_lesson/get_lessons/update_lesson/lessons_snapshot/
  lessons_recent_dup_exists (auto-dedup window).
- 4 auto-detectors (best-effort, source="auto", deduped): gate_failure
  (_handle_qg_failure_rollbacks), merge_hold (_handle_merge_verify HOLD),
  transient_retry (launcher._finalize_transient budget-exhaustion), deploy_degraded
  (post-deploy DEGRADED -> set_repo_freeze).
- src/main.py: GET /lessons, POST /lessons, POST /lessons/{id} + read-only
  `lessons` block in GET /queue; off-switch -> {"enabled": false}.
- src/config.py: lessons_enabled / lessons_query_limit_default / lessons_dedup_window_s.
- tests/test_lessons.py: TC-01..TC-12 (unit + integration), all green.
- Docs: CLAUDE.md, docs/architecture/README.md (component + schema + API), CHANGELOG.

Invariant: the journal is an OBSERVER, not a Quality Gate — STAGE_TRANSITIONS /
QG_CHECKS / check_* / machine-verdict / existing table schemas are byte-for-byte
untouched; enduro not affected. never-raise on every public fn + injection.

Refs: ORCH-098
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/main.py

@@ -1,4 +1,4 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from contextlib import asynccontextmanager
 import logging
 from .db import init_db
@@ -213,6 +213,7 @@ async def queue():
     from . import labels
     from . import cancel
     from . import bug_fast_track
+    from . import lessons
     from .disk_watchdog import disk_watchdog
     from .build_cache_pruner import build_cache_pruner
     return {
@@ -248,6 +249,10 @@ async def queue():
         # kill-switch, label, scope, bug-task counts + the structural savings metric
         # (architecture stages skipped). Additive block; never-raise.
         "bug_fast_track": bug_fast_track.snapshot(),
+        # ORCH-098 (FR-4 / AC-4): lessons-journal observability (read-only) —
+        # kill-switch + counts by type/status + last N lessons. Additive block;
+        # never-raise (snapshot() returns {"enabled": ...} minimum on error).
+        "lessons": lessons.snapshot(),
         # ORCH-063 (FR-6 / AC-7): disk-watchdog observability (read-only) —
         # enabled, threshold, interval, last measurement per host-path. Additive
         # block; never-raise (status() returns {"enabled": ...} minimum on error).
@@ -390,3 +395,82 @@ async def bug_fast_track_escalate(work_item: str = ""):
         except Exception:
             pass
     return {"ok": True, "work_item": work_item, "track": "full", "was": prev_track}
+
+
+# ---------------------------------------------------------------------------
+# ORCH-098 (FR-4 / FR-5, ADR-001 D5): machine lessons-journal endpoints.
+# Read-only fetch + manual record + re-classify. All never-raise; with the
+# kill-switch off they return {"enabled": false} (style of /metrics, AC-7).
+# ---------------------------------------------------------------------------
+@app.get("/lessons")
+async def lessons_list(
+    type: str = "", status: str = "", repo: str = "", work_item: str = "",
+    limit: int | None = None,
+):
+    """ORCH-098: read-only lessons fetch with optional filters (type / status / repo
+    / work_item / limit). Always 200; reading never mutates. ``lessons_enabled=False``
+    -> ``{"enabled": false}``."""
+    from . import lessons
+    from .config import settings
+    if not getattr(settings, "lessons_enabled", True):
+        return {"enabled": False, "lessons": []}
+    rows = lessons.get(
+        lesson_type=(type or None), status=(status or None), repo=(repo or None),
+        work_item_id=(work_item or None), limit=limit,
+    )
+    return {"enabled": True, "lessons": rows}
+
+
+@app.post("/lessons")
+async def lessons_create(request: Request):
+    """ORCH-098: manually record a lesson (``source="manual"``, never deduped). JSON
+    body: ``lesson_type`` (required) + optional context / analysis / attribution
+    fields. Returns ``{"id": <int>}`` or ``{"enabled": false}`` /
+    ``{"error": ...}``."""
+    from . import lessons
+    from .config import settings
+    if not getattr(settings, "lessons_enabled", True):
+        return {"enabled": False}
+    try:
+        body = await request.json()
+    except Exception:  # noqa: BLE001 - malformed body
+        body = {}
+    if not isinstance(body, dict):
+        body = {}
+    lesson_type = body.get("lesson_type")
+    if not lesson_type:
+        return {"ok": False, "error": "missing 'lesson_type'"}
+    # Only forward known fields; source is forced to "manual" (operator/Стрим).
+    allowed = (
+        "work_item_id", "task_id", "stage", "agent", "repo", "root_cause",
+        "suggestion", "status", "related_task", "attribution", "target_repo",
+        "target_domain", "detail",
+    )
+    kwargs = {k: body[k] for k in allowed if k in body}
+    new_id = lessons.record(lesson_type, source="manual", **kwargs)
+    return {"id": new_id}
+
+
+@app.post("/lessons/{lesson_id}")
+async def lessons_update(lesson_id: int, request: Request):
+    """ORCH-098: re-classify / re-status an existing lesson (status / attribution /
+    target_* / related_task / root_cause / suggestion). Lets a human / the
+    retrospective agent classify an auto-recorded ``unknown``. Returns
+    ``{"ok": bool}`` or ``{"enabled": false}``."""
+    from . import lessons
+    from .config import settings
+    if not getattr(settings, "lessons_enabled", True):
+        return {"enabled": False}
+    try:
+        body = await request.json()
+    except Exception:  # noqa: BLE001 - malformed body
+        body = {}
+    if not isinstance(body, dict):
+        body = {}
+    allowed = (
+        "status", "attribution", "target_repo", "target_domain", "related_task",
+        "root_cause", "suggestion", "stage", "agent", "repo", "detail",
+    )
+    kwargs = {k: body[k] for k in allowed if k in body}
+    ok = lessons.update(lesson_id, **kwargs)
+    return {"ok": ok}