Fix(cli): detect resolve intent and report intent_source

potpie/context-engine/adapters/inbound/cli/commands/query.py

@@ -21,6 +21,7 @@
 from adapters.inbound.cli.telemetry.usage_events import (
     capture_usage_command_succeeded,
 )
+from domain.agent_context_port import detect_context_intent
 from domain.ports.agent_context import RecordRequest, ResolveRequest, SearchRequest
 
 
@@ -30,11 +31,33 @@ def _split(value: str | None) -> tuple[str, ...]:
     return tuple(v.strip() for v in value.split(",") if v.strip())
 
 
+def _select_intent(explicit: str | None, task: str | None) -> tuple[str, str]:
+    """Resolve the effective intent and how it was chosen.
+
+    Precedence, mirroring the ``(value, resolved_via)`` pattern used by
+    ``resolve_pot_id`` / ``resolve_pot_scope``:
+
+    - ``--intent`` supplied     → (that value, ``"explicit"``)
+    - detected from the task    → (detected intent, ``"detected"``)
+    - nothing matched           → (``"unknown"``, ``"default"``)
+    """
+    if explicit:
+        return explicit, "explicit"
+    detected = detect_context_intent(task)
+    if detected:
+        return detected, "detected"
+    return "unknown", "default"
+
+
 def register(root: typer.Typer) -> None:
     @root.command()
     def resolve(
         task: str = typer.Argument(..., help="The task to pull context for."),
-        intent: str = typer.Option("feature", "--intent"),
+        intent: str = typer.Option(
+            None,
+            "--intent",
+            help="Override intent. Omit to detect it from the task (falls back to 'unknown').",
+        ),
         include: str = typer.Option(
             None, "--include", help="Comma-separated include families."
         ),
@@ -47,13 +70,15 @@ def resolve(
         with contract():
             host = get_host()
             pot_id = resolve_pot_id(host, pot)
+            resolved_intent, intent_source = _select_intent(intent, task)
             env = host.agent_context.resolve(
                 ResolveRequest(
                     pot_id=pot_id,
                     task=task,
-                    intent=intent,
+                    intent=resolved_intent,
                     include=_split(include),
                     mode=mode,
+                    metadata={"intent_source": intent_source},
                 )
             )
             _capture_context_activation(command="resolve", item_count=len(env.items))
@@ -128,6 +153,7 @@ def _envelope_payload(env) -> dict[str, object]:
     return {
         "pot_id": env.pot_id,
         "intent": env.intent,
+        "intent_source": dict(env.metadata).get("intent_source", "default"),
         "overall_confidence": env.overall_confidence,
         "items": [
             {"include": i.include, "score": i.score, "payload": dict(i.payload)}
@@ -146,8 +172,10 @@ def _envelope_payload(env) -> dict[str, object]:
 
 
 def _envelope_human(env) -> str:
+    intent_source = dict(env.metadata).get("intent_source", "default")
     lines = [
-        f"pot={env.pot_id} intent={env.intent} confidence={env.overall_confidence} items={len(env.items)}"
+        f"pot={env.pot_id} intent={env.intent} (source={intent_source}) "
+        f"confidence={env.overall_confidence} items={len(env.items)}"
     ]
     for item in env.items[:10]:
         fact = dict(item.payload).get("fact") or dict(item.payload).get("summary") or ""

potpie/context-engine/domain/agent_context_port.py

@@ -17,6 +17,7 @@
 
 import hashlib
 import json
+import re
 from typing import Any
 
 from domain.ontology import (
@@ -172,6 +173,119 @@
 }
 
 
+# --- Deterministic intent detection -----------------------------------------
+# A dependency-free keyword/phrase matcher over the free-text task, mirroring
+# the "no-guess" discipline of ``domain.ontology_classifier``: when nothing
+# matches confidently it returns ``None`` (the caller surfaces that as the
+# neutral ``unknown`` intent) rather than guessing an intent. The tables are
+# seeded from the natural-language trigger phrases in each recipe's ``"when"``
+# field (see :data:`CONTEXT_RESOLVE_RECIPES`). ``unknown`` is intentionally
+# absent — it is the fallback, never a detected result.
+#
+# Detection runs *before* ``normalize_context_intent`` and does not modify the
+# intent vocabulary, recipes, or default-include map.
+_INTENT_KEYWORDS: dict[str, tuple[str, ...]] = {
+    "security": (
+        "vulnerability",
+        "vulnerabilities",
+        "cve",
+        "exploit",
+        "security",
+        "audit",
+        "hardening",
+        "injection",
+        "auth bypass",
+    ),
+    "debugging": (
+        "bug",
+        "error",
+        "errors",
+        "crash",
+        "crashing",
+        "incident",
+        "failing",
+        "failure",
+        "exception",
+        "traceback",
+        "stack trace",
+        "flaky",
+        "500",
+        "throwing",
+        "broken",
+    ),
+    "operations": (
+        "deploy",
+        "deployment",
+        "runbook",
+        "production",
+        "rollback",
+        "outage",
+        "on-call",
+        "oncall",
+        "alert",
+    ),
+    "refactor": (
+        "refactor",
+        "technical debt",
+        "tech debt",
+        "migrate",
+        "migration",
+        "cleanup",
+        "clean up",
+        "restructure",
+    ),
+    "test": ("test", "tests", "pytest", "coverage", "test suite"),
+    "review": (
+        "review",
+        "pull request",
+        "what changed",
+        "recently",
+        "recent changes",
+        "latest changes",
+        "risky",
+    ),
+    "docs": ("documentation", "docs", "readme", "document"),
+    "onboarding": (
+        "onboard",
+        "onboarding",
+        "unfamiliar",
+        "getting started",
+        "new to",
+        "up to speed",
+    ),
+    "planning": ("roadmap", "sprint", "architecture", "design doc", "milestone"),
+    "feature": ("feature", "implement", "add support", "endpoint", "behavior change"),
+}
+
+
+_INTENT_DETECTION_PRIORITY: tuple[str, ...] = tuple(_INTENT_KEYWORDS)
+
+_INTENT_MATCHERS: dict[str, re.Pattern[str]] = {
+    intent: re.compile(
+        r"|".join(r"\b" + re.escape(kw) + r"\b" for kw in keywords),
+        re.IGNORECASE,
+    )
+    for intent, keywords in _INTENT_KEYWORDS.items()
+}
+
+
+def detect_context_intent(task: str | None) -> str | None:
+    """Map a free-text task to a canonical intent, or ``None`` if unsure.
+
+    Deterministic, case-insensitive, word-boundary keyword match. When the
+    task triggers more than one intent, :data:`_INTENT_DETECTION_PRIORITY`
+    breaks the tie (most specific / highest-risk first). Returns ``None`` when
+    nothing matches — favouring "no guess" over a wrong guess — which the CLI
+    surfaces as the neutral ``unknown`` intent with ``intent_source=default``.
+    """
+    if not task or not task.strip():
+        return None
+    for intent in _INTENT_DETECTION_PRIORITY:
+        if _INTENT_MATCHERS[intent].search(task):
+            return intent
+    return None
+
+
 def normalize_context_intent(intent: str | None) -> str:
     value = (intent or "unknown").strip().lower()
     return value if value in CONTEXT_INTENTS else "unknown"

potpie/context-engine/tests/unit/test_agent_context_port.py

@@ -7,7 +7,9 @@
 from domain.agent_context_port import (
     CONTEXT_INTENTS,
     CONTEXT_RESOLVE_RECIPES,
+    DEFAULT_INTENT_INCLUDES,
     context_recipe_for_intent,
+    detect_context_intent,
 )
 
 pytestmark = pytest.mark.unit
@@ -35,3 +37,84 @@ def test_context_recipe_for_intent_returns_curated_not_generic() -> None:
         assert recipe["mode"] == curated["mode"]
         assert recipe["source_policy"] == curated["source_policy"]
         assert recipe["include"] == curated["include"]
+
+
+@pytest.mark.parametrize(
+    ("task", "expected"),
+    [
+        ("the payment webhook is throwing a 500 error and crashing", "debugging"),
+        ("investigate the failing checkout incident", "debugging"),
+        ("deploy the auth service to production", "operations"),
+        ("refactor the retry queue and clean up tech debt", "refactor"),
+        ("add write coverage in the pytest suite", "test"),
+        ("review this pull request for risky changes", "review"),
+        ("what changed recently in the auth service?", "review"),
+        ("run a security audit for the injection vulnerability", "security"),
+        ("update the readme documentation", "docs"),
+        ("getting started in an unfamiliar repo", "onboarding"),
+        ("plan the sprint roadmap and architecture", "planning"),
+        ("implement a new feature endpoint", "feature"),
+    ],
+)
+def test_detect_context_intent_maps_representative_tasks(
+    task: str, expected: str
+) -> None:
+    assert detect_context_intent(task) == expected
+
+
+def test_detect_context_intent_returns_only_canonical_intents() -> None:
+    """A detected intent is always a real, curated intent (never 'unknown')."""
+    detected = detect_context_intent("deploy to production")
+    assert detected in CONTEXT_INTENTS
+    assert detected != "unknown"
+
+
+def test_detect_recent_change_selects_timeline_bearing_intent() -> None:
+    """Recent-change phrasing must route to an intent whose defaults include timeline."""
+    detected = detect_context_intent("what changed recently in billing?")
+    assert detected is not None
+    assert "timeline" in DEFAULT_INTENT_INCLUDES[detected]
+
+
+@pytest.mark.parametrize("task", ["", "   ", None, "the quick brown fox jumps"])
+def test_detect_context_intent_returns_none_when_unsure(task) -> None:
+    assert detect_context_intent(task) is None
+
+
+@pytest.mark.parametrize(
+    ("task", "expected", "also_matches"),
+    [
+        # security beats debugging + operations
+        (
+            "audit the vulnerability behind the failing production deploy",
+            "security",
+            ("debugging", "operations"),
+        ),
+        # debugging beats operations
+        ("the production deploy is throwing an error", "debugging", ("operations",)),
+        # operations beats review
+        ("review the runbook before the deployment", "operations", ("review",)),
+        # review beats feature
+        ("review the new endpoint implementation", "review", ("feature",)),
+    ],
+)
+def test_detect_context_intent_multi_match_resolves_by_priority(
+    task: str, expected: str, also_matches: tuple[str, ...]
+) -> None:
+    """Regression: when a task matches several intents, the documented
+    priority order (keyword-table order, most specific / highest-risk first)
+    must decide the winner deterministically."""
+    from domain.agent_context_port import _INTENT_MATCHERS
+
+    # Guard the fixture itself: each lower-priority intent really does match
+    # the phrase on its own, so the test exercises a genuine tie-break.
+    for other in also_matches:
+        assert _INTENT_MATCHERS[other].search(task), (
+            f"fixture phrase no longer matches '{other}'; tie-break not exercised"
+        )
+    assert detect_context_intent(task) == expected
+
+
+def test_detect_context_intent_does_not_false_match_substrings() -> None:
+    """Word-boundary matching: 'latest' must not trigger the 'test' intent."""
+    assert detect_context_intent("show me the latest greatest release") != "test"

potpie/context-engine/tests/unit/test_cli_resolve_intent.py

@@ -0,0 +1,62 @@
+"""CLI ``resolve`` intent selection + intent_source surfacing (issue #996)."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from adapters.inbound.cli.commands.query import (
+    _envelope_human,
+    _envelope_payload,
+    _select_intent,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def _fake_env(*, intent: str, intent_source: str) -> SimpleNamespace:
+    return SimpleNamespace(
+        pot_id="pot_1",
+        intent=intent,
+        overall_confidence="low",
+        items=(),
+        coverage=(),
+        unsupported_includes=(),
+        metadata={"intent_source": intent_source},
+    )
+
+
+def test_select_intent_explicit_flag_wins() -> None:
+    assert _select_intent("debugging", "some unrelated task") == (
+        "debugging",
+        "explicit",
+    )
+
+
+def test_select_intent_detects_from_task() -> None:
+    assert _select_intent(None, "the service is throwing a 500 error") == (
+        "debugging",
+        "detected",
+    )
+
+
+def test_select_intent_falls_back_to_unknown_default() -> None:
+    assert _select_intent(None, "the quick brown fox") == ("unknown", "default")
+
+
+def test_envelope_payload_exposes_intent_source() -> None:
+    payload = _envelope_payload(_fake_env(intent="debugging", intent_source="detected"))
+    assert payload["intent"] == "debugging"
+    assert payload["intent_source"] == "detected"
+
+
+def test_envelope_human_header_shows_source() -> None:
+    header = _envelope_human(_fake_env(intent="unknown", intent_source="default"))
+    assert "intent=unknown (source=default)" in header
+
+
+def test_envelope_payload_defaults_source_when_metadata_absent() -> None:
+    env = _fake_env(intent="feature", intent_source="detected")
+    env.metadata = {}
+    assert _envelope_payload(env)["intent_source"] == "default"