diff --git a/potpie/cli/commands/query.py b/potpie/cli/commands/query.py index 6cd8baa4b..60a68358f 100644 --- a/potpie/cli/commands/query.py +++ b/potpie/cli/commands/query.py @@ -21,6 +21,7 @@ from potpie.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 "" diff --git a/potpie/context-engine/domain/agent_context_port.py b/potpie/context-engine/domain/agent_context_port.py index 7c5d1e59e..335334fec 100644 --- a/potpie/context-engine/domain/agent_context_port.py +++ b/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" diff --git a/potpie/context-engine/tests/unit/test_agent_context_port.py b/potpie/context-engine/tests/unit/test_agent_context_port.py index 7b0ddf5d8..60b4eb4d7 100644 --- a/potpie/context-engine/tests/unit/test_agent_context_port.py +++ b/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" diff --git a/tests/unit/test_cli_resolve_intent.py b/tests/unit/test_cli_resolve_intent.py new file mode 100644 index 000000000..fd1297a81 --- /dev/null +++ b/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 potpie.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"