feat(cli): scope raglogs ask to latest ingestion (#29)

README.md

@@ -564,8 +564,20 @@ raglogs ask "why did login fail?"
 raglogs ask "what changed before latency increased?" --since 2h
 raglogs ask "what happened in billing?" --since 1h
 raglogs ask "why are checkouts failing?" --format json
+raglogs ask "why did login fail?" --ingestion-job <uuid>
+raglogs ask "why did login fail?" --all-ingestions
 ```
 
+Like `explain` / `timeline` / `compare`, `ask` defaults to the latest completed ingestion. Pass `--ingestion-job` to target a specific job, or `--all-ingestions` to search every ingested log.
+
+| Flag | Description |
+|---|---|
+| `--since` | Relative window: `30m`, `1h`, `24h` |
+| `--service` | Filter to one service |
+| `--format` | `text` or `json` |
+| `--ingestion-job` | Scope to a specific ingestion job UUID |
+| `--all-ingestions` | Search all historical ingestions, not just the latest |
+
 **How retrieval works**
 
 1. **Semantic** — if `EMBEDDINGS_PROVIDER` is `openai` or `local` *and* log lines were ingested with `--with-embeddings`, the question is embedded and nearest neighbors are fetched from pgvector (`ASK_SEMANTIC_TOP_K`, `ASK_SEMANTIC_MIN_SIMILARITY`). Paraphrases that keyword search would miss (e.g. "why are payments being declined?" vs "Stripe signature verification failed") can still match.

src/cli/commands/ask.py

@@ -4,32 +4,39 @@
 from rich.console import Console
 from rich.panel import Panel
 
-app = typer.Typer()
 console = Console()
 
 
-@app.callback(invoke_without_command=True)
 def ask_cmd(
-    ctx: typer.Context,
     question: Optional[str] = typer.Argument(None, help="Your question about the logs"),
     since: Optional[str] = typer.Option(None, "--since", help="Time window e.g. 2h"),
     service: Optional[str] = typer.Option(None, "--service", help="Filter by service"),
     fmt: str = typer.Option("text", "--format", help="Output format: text|json"),
+    all_ingestions: bool = typer.Option(
+        False,
+        "--all-ingestions",
+        help="Search all historical ingestion data (not just latest)",
+    ),
+    ingestion_job: Optional[str] = typer.Option(
+        None,
+        "--ingestion-job",
+        help="Scope search to a specific ingestion job UUID",
+    ),
     scope: str = typer.Option(
         "default",
         "--scope",
         help="Isolation scope (CLI default: default)",
     ),
 ):
     """Ask a natural language question about your logs."""
-    if ctx.invoked_subcommand is not None:
-        return
-
     if not question:
         console.print("[red]Error:[/red] Please provide a question.")
         console.print("[dim]Example: raglogs ask 'why did login fail?'[/dim]")
         raise typer.Exit(1)
 
+    import uuid
+
+    from src.core.explain.summarizer import get_latest_ingestion_job_id
     from src.core.retrieval.question_router import answer_question
     from src.db.session import get_db
     from src.utils.time import resolve_window
@@ -45,12 +52,21 @@ def ask_cmd(
     with console.status(f"[cyan]Searching logs for: {question}[/cyan]"):
         try:
             with get_db() as db:
+                job_id = None
+                if ingestion_job:
+                    job_id = uuid.UUID(ingestion_job)
+                elif not all_ingestions:
+                    # Default: scope to latest ingestion job to avoid mixing incidents
+                    job_id = get_latest_ingestion_job_id(db, scope=scope)
+                    if job_id:
+                        console.print(f"[dim]Scoped to latest ingestion: {job_id}[/dim]")
                 result = answer_question(
                     db=db,
                     question=question,
                     window_start=window_start,
                     window_end=window_end,
                     service=service,
+                    ingestion_job_id=job_id,
                     scope=scope,
                 )
         except Exception as e:

src/cli/main.py

@@ -11,7 +11,7 @@ def _build_app() -> typer.Typer:
     from src.cli.commands.clusters import clusters_cmd
     from src.cli.commands.status import status_cmd
     from src.cli.commands.config_cmd import config_cmd
-    from src.cli.commands.ask import app as ask_app
+    from src.cli.commands.ask import ask_cmd
     from src.cli.commands.demo import demo_cmd
     from src.cli.commands.worker import worker_cmd
     from src.cli.commands.timeline import timeline_cmd
@@ -37,7 +37,7 @@ def _build_app() -> typer.Typer:
     _app.command("timeline")(timeline_cmd)
     _app.command("compare")(compare_cmd)
     _app.command("purge")(purge_cmd)
-    _app.add_typer(ask_app, name="ask")
+    _app.command("ask")(ask_cmd)
     _app.add_typer(keys_app, name="keys")
     return _app
 

tests/unit/test_cli_ask.py

@@ -0,0 +1,122 @@
+"""CLI wiring for `raglogs ask` ingestion scoping.
+
+The core retrieval path already accepts `ingestion_job_id`; these tests
+cover Typer flag wiring only (no database).
+"""
+import re
+import uuid
+from unittest.mock import MagicMock, patch
+
+from typer.testing import CliRunner
+
+from src.cli.main import app
+from src.core.retrieval.question_router import AskResult
+
+runner = CliRunner()
+
+_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")
+
+
+def _plain_help(text: str) -> str:
+    """Strip ANSI and whitespace so wrapped flags like `--ingestion-\\njob` match."""
+    return re.sub(r"\s+", "", _ANSI_RE.sub("", text))
+
+
+def _mock_result() -> AskResult:
+    return AskResult(
+        question="why did login fail?",
+        answer_text="Auth token invalid in api.",
+        evidence_items=["12 events: Auth token invalid"],
+        clusters_used=[],
+        total_matches=12,
+        retrieval_mode="keyword",
+    )
+
+
+def _ctx_db() -> MagicMock:
+    mock_db = MagicMock()
+    mock_db.__enter__ = MagicMock(return_value=mock_db)
+    mock_db.__exit__ = MagicMock(return_value=False)
+    return mock_db
+
+
+class TestAskCliIngestionFlags:
+    def test_help_lists_ingestion_flags(self):
+        result = runner.invoke(
+            app,
+            ["ask", "--help"],
+            env={"COLUMNS": "120", "NO_COLOR": "1", "TERM": "dumb"},
+        )
+        assert result.exit_code == 0
+        help_text = _plain_help(result.output)
+        assert "--ingestion-job" in help_text
+        assert "--all-ingestions" in help_text
+
+    def test_defaults_to_latest_completed_ingestion(self):
+        job_id = uuid.uuid4()
+        mock_db = _ctx_db()
+        with patch("src.db.session.get_db", side_effect=lambda: mock_db), \
+             patch(
+                 "src.core.explain.summarizer.get_latest_ingestion_job_id",
+                 return_value=job_id,
+             ) as mock_latest, \
+             patch(
+                 "src.core.retrieval.question_router.answer_question",
+                 return_value=_mock_result(),
+             ) as mock_answer:
+            result = runner.invoke(app, ["ask", "why did login fail?"])
+
+        assert result.exit_code == 0, result.output
+        mock_latest.assert_called_once_with(mock_db, scope="default")
+        assert mock_answer.call_args.kwargs["ingestion_job_id"] == job_id
+
+    def test_ingestion_job_flag_passed_through(self):
+        job_id = str(uuid.uuid4())
+        mock_db = _ctx_db()
+        with patch("src.db.session.get_db", side_effect=lambda: mock_db), \
+             patch(
+                 "src.core.explain.summarizer.get_latest_ingestion_job_id",
+             ) as mock_latest, \
+             patch(
+                 "src.core.retrieval.question_router.answer_question",
+                 return_value=_mock_result(),
+             ) as mock_answer:
+            result = runner.invoke(app, [
+                "ask", "why did login fail?", "--ingestion-job", job_id,
+            ])
+
+        assert result.exit_code == 0, result.output
+        mock_latest.assert_not_called()
+        assert str(mock_answer.call_args.kwargs["ingestion_job_id"]) == job_id
+
+    def test_all_ingestions_skips_scoping(self):
+        mock_db = _ctx_db()
+        with patch("src.db.session.get_db", side_effect=lambda: mock_db), \
+             patch(
+                 "src.core.explain.summarizer.get_latest_ingestion_job_id",
+             ) as mock_latest, \
+             patch(
+                 "src.core.retrieval.question_router.answer_question",
+                 return_value=_mock_result(),
+             ) as mock_answer:
+            result = runner.invoke(app, [
+                "ask", "why did login fail?", "--all-ingestions",
+            ])
+
+        assert result.exit_code == 0, result.output
+        mock_latest.assert_not_called()
+        assert mock_answer.call_args.kwargs["ingestion_job_id"] is None
+
+    def test_invalid_ingestion_job_exits_1(self):
+        mock_db = _ctx_db()
+        with patch("src.db.session.get_db", side_effect=lambda: mock_db), \
+             patch(
+                 "src.core.retrieval.question_router.answer_question",
+             ) as mock_answer:
+            result = runner.invoke(app, [
+                "ask", "why did login fail?", "--ingestion-job", "not-a-uuid",
+            ])
+
+        assert result.exit_code == 1, result.output
+        mock_answer.assert_not_called()
+        assert "Error" in result.output