fix: ingest pipeline — input contracts, payload normalization, freshness guards

adapters/code_locator.py

@@ -21,6 +21,19 @@ def get_code_locator():
     return RealCodeLocatorAdapter(repo_path=repo_path)
 
 
+def ensure_code_graph_fresh(repo_path: str | None = None) -> None:
+    """Ensure the code graph index exists and matches HEAD.
+
+    Safe to call multiple times — only rebuilds if stale.
+    Called automatically by tools that depend on the code graph.
+    """
+    repo = repo_path or os.getenv("REPO_PATH", ".")
+    ensure_runtime_env()
+    from code_locator.config import load_config
+    config = load_config()
+    ensure_index_matches_repo(repo, config)
+
+
 # Alias for the CodeIntelligencePort factory (same implementation, named for the port)
 get_code_intelligence = get_code_locator
 

contracts.py

@@ -119,7 +119,76 @@ class DetectDriftResponse(BaseModel):
 # (LinkCommitResponse defined above alongside SearchDecisionsResponse)
 
 
-# ── Tool 5: /ingest ───────────────────────────────────────────────────
+# ── Tool 5: /ingest — INPUT contracts ────────────────────────────────
+
+
+class IngestSpan(BaseModel):
+    """Source excerpt from a meeting, document, or manual input."""
+    text: str = ""
+    source_type: str = "manual"       # transcript | notion | document | manual
+    source_ref: str = ""              # meeting ID, Notion page ID, etc.
+    speakers: list[str] = []
+    meeting_date: str = ""
+
+
+class IngestCodeRegion(BaseModel):
+    """Pre-resolved code region for a mapping."""
+    symbol: str
+    file_path: str
+    start_line: int = 0
+    end_line: int = 0
+    type: str = "function"
+    purpose: str = ""
+
+
+class IngestMapping(BaseModel):
+    """One decision-to-code mapping in the internal pipeline format."""
+    intent: str
+    span: IngestSpan = IngestSpan()
+    symbols: list[str] = []
+    code_regions: list[IngestCodeRegion] = []
+
+
+class IngestDecision(BaseModel):
+    """One decision in the natural LLM-generated format."""
+    id: str = ""
+    title: str = ""
+    description: str = ""
+    status: str = ""
+    participants: list[str] = []
+
+
+class IngestActionItem(BaseModel):
+    owner: str = "unassigned"
+    action: str = ""
+    due: str = ""
+
+
+class IngestPayload(BaseModel):
+    """Ingest input — accepts EITHER mappings (internal) or decisions (natural LLM).
+
+    If ``mappings`` is present, it's used directly (internal pipeline format).
+    If ``decisions`` is present, they are normalized into mappings automatically.
+    """
+    # Common fields
+    repo: str = ""
+    commit_hash: str = ""
+    query: str = ""
+
+    # Internal pipeline format
+    mappings: list[IngestMapping] = []
+
+    # Natural LLM-generated format (normalized into mappings if present)
+    source: str = "manual"
+    title: str = ""
+    date: str = ""
+    participants: list[str] = []
+    decisions: list[IngestDecision] = []
+    action_items: list[IngestActionItem] = []
+    open_questions: list[str] = []
+
+
+# ── Tool 5: /ingest — RESPONSE contracts ─────────────────────────────
 
 
 class IngestStats(BaseModel):

handlers/decision_status.py

@@ -1,25 +1,35 @@
 """Handler for /decision_status MCP tool.
 
 Surfaces implementation status of all tracked decisions.
-Read-only — does NOT auto-trigger link_commit.
+Auto-syncs the ledger to HEAD before returning status.
 
 Phase 0: backed by MockLedgerAdapter fixture data
 Phase 2: backed by SurrealDBLedgerAdapter with real graph traversal
 """
 
 from __future__ import annotations
 
+import logging
 from datetime import datetime, timezone
 
 from adapters.ledger import get_ledger
 from contracts import CodeRegionSummary, DecisionStatusEntry, DecisionStatusResponse
 
+logger = logging.getLogger(__name__)
+
 
 async def handle_decision_status(
     filter: str = "all",
     since: str | None = None,
     ref: str = "HEAD",
 ) -> DecisionStatusResponse:
+    # Auto-sync to HEAD so status reflects current code state
+    try:
+        from handlers.link_commit import handle_link_commit
+        await handle_link_commit(ref)
+    except Exception as exc:
+        logger.warning("[status] auto-sync failed: %s", exc)
+
     ledger = get_ledger()
     decisions_raw = await ledger.get_all_decisions(filter=filter)
 

handlers/ingest.py

@@ -13,7 +13,7 @@
 import re
 
 from adapters.ledger import get_ledger
-from contracts import IngestResponse, IngestStats, SourceCursorSummary
+from contracts import IngestPayload, IngestResponse, IngestStats, SourceCursorSummary
 
 logger = logging.getLogger(__name__)
 
@@ -34,6 +34,73 @@
 })
 
 
+def _normalize_payload(payload: dict) -> dict:
+    """Validate and normalize ingest payload using Pydantic contracts.
+
+    1. Validates the raw dict against IngestPayload (fails fast on bad types)
+    2. If ``mappings`` is already present, returns as-is (internal format)
+    3. If ``decisions``/``action_items``/``open_questions`` present, converts to mappings
+    """
+    validated = IngestPayload.model_validate(payload)
+
+    # Already has mappings — convert back to dict and return
+    if validated.mappings:
+        return validated.model_dump()
+
+    mappings: list[dict] = []
+    source_meta = {
+        "source_type": validated.source,
+        "source_ref": validated.title,
+        "speakers": validated.participants,
+        "meeting_date": validated.date,
+    }
+
+    for d in validated.decisions:
+        text = d.description or d.title
+        if not text:
+            continue
+        mappings.append({
+            "intent": text,
+            "span": {
+                **source_meta,
+                "text": text,
+                "source_ref": d.id or source_meta["source_ref"],
+                "speakers": d.participants or source_meta["speakers"],
+            },
+            "symbols": [],
+            "code_regions": [],
+        })
+
+    for a in validated.action_items:
+        text = f"[Action: {a.owner}] {a.action}"
+        mappings.append({
+            "intent": text,
+            "span": {**source_meta, "text": text},
+            "symbols": [],
+            "code_regions": [],
+        })
+
+    for q in validated.open_questions:
+        text = f"[Open Question] {q}"
+        mappings.append({
+            "intent": text,
+            "span": {**source_meta, "text": text},
+            "symbols": [],
+            "code_regions": [],
+        })
+
+    if not mappings:
+        logger.warning(
+            "[ingest] payload validated but produced 0 mappings: %s",
+            list(payload.keys()),
+        )
+        return validated.model_dump()
+
+    result = validated.model_dump()
+    result["mappings"] = mappings
+    return result
+
+
 def _regions_from_symbol_ids(symbol_ids: list[int], db, description: str) -> list[dict]:
     """Resolve a list of symbol IDs to code_region dicts."""
     regions = []
@@ -79,8 +146,9 @@ def _auto_ground_via_search(mappings: list[dict], repo: str) -> tuple[list[dict]
         db_path = str(os.path.join(repo, ".bicameral", "code-graph.db"))
 
     try:
-        from adapters.code_locator import get_code_locator
+        from adapters.code_locator import get_code_locator, ensure_code_graph_fresh
         from code_locator.indexing.sqlite_store import SymbolDB
+        ensure_code_graph_fresh(repo)
         locator = get_code_locator()
         db = SymbolDB(db_path)
     except Exception as exc:
@@ -196,7 +264,9 @@ def _resolve_symbols_to_regions(payload: dict, repo: str) -> dict:
         db_path = str(_os.path.join(repo, ".bicameral", "code-graph.db"))
 
     try:
+        from adapters.code_locator import ensure_code_graph_fresh
         from code_locator.indexing.sqlite_store import SymbolDB
+        ensure_code_graph_fresh(repo)
         db = SymbolDB(db_path)
     except Exception as exc:
         logger.warning("[ingest] cannot open symbol DB at %s: %s", db_path, exc)
@@ -249,12 +319,20 @@ async def handle_ingest(
     if hasattr(ledger, "connect"):
         await ledger.connect()
 
+    payload = _normalize_payload(payload)
     repo = str(payload.get("repo") or os.getenv("REPO_PATH", "."))
     payload = _resolve_symbols_to_regions(payload, repo)
     mappings, grounding_deferred = _auto_ground_via_search(payload.get("mappings") or [], repo)
     payload = {**payload, "mappings": mappings}
     result = await ledger.ingest_payload(payload)
 
+    # Sync ledger to HEAD and re-ground any previously ungrounded intents
+    try:
+        from handlers.link_commit import handle_link_commit
+        await handle_link_commit("HEAD")
+    except Exception as exc:
+        logger.warning("[ingest] post-ingest link_commit failed: %s", exc)
+
     cursor_summary = None
     source_type = str(((payload.get("mappings") or [{}])[0].get("span") or {}).get("source_type", "manual"))
     last_source_ref = _derive_last_source_ref(payload)

server.py

@@ -73,7 +73,7 @@ async def list_tools() -> list[Tool]:
             description=(
                 "Surface implementation status of all tracked decisions for the repo. "
                 "Shows which decisions are reflected in code, drifted, pending, or ungrounded. "
-                "Read-only — does not trigger a ledger sync. Slash alias: /bicameral:status"
+                "Auto-syncs the ledger to HEAD before returning status. Slash alias: /bicameral:status"
             ),
             inputSchema={
                 "type": "object",
@@ -171,8 +171,10 @@ async def list_tools() -> list[Tool]:
         Tool(
             name="bicameral.ingest",
             description=(
-                "Ingest a normalized source payload into the decision ledger and advance a source cursor. "
-                "Use this after Slack/Notion/source sync to make new decisions visible to status/search. "
+                "Ingest decisions into the ledger. Accepts two payload formats: "
+                "(1) Internal: {mappings: [{intent, span, symbols, code_regions}]} "
+                "(2) Natural: {decisions: [{title, description}], action_items: [...], open_questions: [...]} "
+                "Auto-grounds decisions to code via BM25. Ensures code graph freshness before grounding. "
                 "Slash alias: /bicameral:ingest"
             ),
             inputSchema={