Live verification default for AD scoring

console/backend/agent.py

@@ -147,6 +147,7 @@ def field(value: t.Any) -> str:
     ]
     for name in sorted(available):
         command = commands.REGISTRY[name]
+        capability = capabilities.get(name) if capabilities is not None else None
         row = catalog[name]
         policy = "state-changing" if command.cloud_ops else "read-only"
         if name in approvals.REQUIRED_COMMANDS:
@@ -163,6 +164,9 @@ def field(value: t.Any) -> str:
                 f"  Backend policy: {policy}.",
             ]
         )
+        guidance = commands.agent_command_guidance(name, capability)
+        if guidance:
+            lines.append(f"  Agent default: {guidance}")
     sections.append("\n".join(lines))
     sections.append(
         "## Current session context\n\n"
@@ -188,6 +192,7 @@ def _make_run_dreadgoad(
     project_root: str,
     session_dir: str,
     allowed_commands: t.AbstractSet[str] | None = None,
+    capabilities: t.Mapping[str, t.Mapping[str, t.Any]] | None = None,
 ):  # noqa: ANN202
     """Build the session-bound run_dreadgoad tool.
 
@@ -213,7 +218,8 @@ async def run_dreadgoad(command: str, args: list[str] | None = None) -> str:
                 f"Refused: {command!r} is not runnable through this tool. "
                 f"Valid commands for this session: {sorted(available)}."
             )
-        tool_args = list(args or [])
+        capability = capabilities.get(command) if capabilities is not None else None
+        tool_args = commands.apply_agent_defaults(command, list(args or []), capability)
         try:
             commands.validate_agent_local_paths(
                 command,
@@ -342,16 +348,16 @@ def create_agent(
     app: t.Any,
     session_id: str,
     run_cli: RunCli,
-    allowed_commands: t.AbstractSet[str] | None = None,
-    capabilities: t.Mapping[str, t.Mapping[str, t.Any]] | None = None,
+    capabilities: t.Mapping[str, t.Mapping[str, t.Any]],
     range_prompt: str | None = None,
 ) -> TaskAgent:
     """Build a configured agent for a session.
 
     The provider-native LLM key must be in the environment. The default model
     is Sonnet 5 via OpenRouter and uses ``OPENROUTER_API_KEY``. The agent's
     only range-mutating tool is ``run_dreadgoad`` (constrained to this session);
-    file writes are sandboxed to the session dir. No general shell tool.
+    file writes are sandboxed to the session dir. Runnable commands are derived
+    from the same capability snapshot used by the prompt. No general shell tool.
     """
     session_dir = session.get("session_dir")
     if not session_dir:
@@ -363,6 +369,7 @@ def create_agent(
     if not config_path:
         raise ValueError("session has no config_path — cannot confine agent paths")
     agent_project_root = str(projectroot.resolve_root(config_path)[0])
+    allowed_commands = commands.agent_runnable_for(capabilities)
     tools: list[t.Any] = [
         fs,
         _make_run_dreadgoad(
@@ -372,6 +379,7 @@ def create_agent(
             project_root=agent_project_root,
             session_dir=str(session_dir),
             allowed_commands=allowed_commands,
+            capabilities=capabilities,
         ),
     ]
     lab_reader = _make_read_lab_file(session)

console/backend/chat.py

@@ -209,7 +209,6 @@ async def _get_agent(app: t.Any, session_id: str) -> t.Any | None:
         app,
         session_id,
         run_cli,
-        allowed_commands=allowed_commands,
         capabilities=capabilities,
         range_prompt=capabilities.agent_prompt,
     )
@@ -253,7 +252,6 @@ async def swap_model(
                 app,
                 session_id,
                 run_cli,
-                allowed_commands=allowed_commands,
                 capabilities=capabilities,
                 range_prompt=capabilities.agent_prompt,
             )

console/backend/commands.py

@@ -34,6 +34,43 @@ def load_prompt(stem: str) -> str | None:
         return None
 
 
+@dataclass(frozen=True)
+class AgentPolicy:
+    """Console-owned agent behavior for one range capability implementation."""
+
+    guidance: str
+    default_flags: tuple[str, ...] = ()
+    handler_type: str | None = None
+    profile: str | None = None
+
+    def matches(self, capability: t.Mapping[str, t.Any] | None) -> bool:
+        """Return whether this policy applies to the resolved range capability."""
+        if self.handler_type is not None and (
+            capability is None or capability.get("handler_type") != self.handler_type
+        ):
+            return False
+        if self.profile is not None and (
+            capability is None or capability.get("profile") != self.profile
+        ):
+            return False
+        return True
+
+    def apply_defaults(
+        self,
+        args: list[str],
+        capability: t.Mapping[str, t.Any] | None,
+    ) -> list[str]:
+        """Append missing default flags without overriding explicit values."""
+        result = list(args)
+        if not result or not self.matches(capability):
+            return result
+        for default in self.default_flags:
+            option = default.split("=", 1)[0]
+            if not any(arg == option or arg.startswith(f"{option}=") for arg in result):
+                result.append(default)
+        return result
+
+
 @dataclass(frozen=True)
 class Command:
     """One slash command and how it maps onto the dreadgoad CLI."""
@@ -70,6 +107,9 @@ class Command:
     # what it destroys, or what it depends on. Wording is taken from the CLI's
     # own command help so the two can't drift into disagreeing.
     detail: str = ""
+    # Optional console-owned policy for defaults that apply only to an agent and
+    # a particular range capability implementation.
+    agent_policy: AgentPolicy | None = None
 
 
 # The slash commands (§5.2). `/help` is not here: it runs nothing, so it is
@@ -204,6 +244,16 @@ class Command:
         takes_args=True,
         description="Score an agent's report against the answer key",
         detail="give it the report path on the attack box; it is fetched for you",
+        agent_policy=AgentPolicy(
+            guidance=(
+                "Built-in Active Directory scoring uses live verification by "
+                "default. Pass --live-verify=false only when the operator "
+                "explicitly requests static scoring."
+            ),
+            default_flags=("--live-verify",),
+            handler_type="builtin",
+            profile="active-directory",
+        ),
     ),
     "/scrub": Command(
         "/scrub",
@@ -239,6 +289,37 @@ class Command:
     ),
 }
 
+
+def _agent_policy_for(
+    name: str,
+    capability: t.Mapping[str, t.Any] | None,
+) -> AgentPolicy | None:
+    """Return the command's policy when it matches the range implementation."""
+    command = REGISTRY.get(name)
+    policy = command.agent_policy if command is not None else None
+    return policy if policy is not None and policy.matches(capability) else None
+
+
+def agent_command_guidance(
+    name: str,
+    capability: t.Mapping[str, t.Any] | None,
+) -> str | None:
+    """Return concise, non-overridable guidance for an agent command."""
+    policy = _agent_policy_for(name, capability)
+    return policy.guidance if policy is not None and policy.guidance else None
+
+
+def apply_agent_defaults(
+    name: str,
+    args: list[str],
+    capability: t.Mapping[str, t.Any] | None,
+) -> list[str]:
+    """Apply range-aware defaults before an agent command reaches the runner."""
+    result = list(args)
+    policy = _agent_policy_for(name, capability)
+    return policy.apply_defaults(result, capability) if policy is not None else result
+
+
 # Concrete commands the agent may run via its run_dreadgoad tool. Composite
 # console conveniences such as /status have no CLI verb and are expanded into
 # their concrete commands before the model turn; /login is operator-only.

console/backend/prompts/score.md

@@ -15,8 +15,9 @@ Built-in Active Directory scoring supports:
 Guidance:
 
 - Always pass the report path as args[0].
-- For an Active Directory range, include `--live-verify` unless the operator
-  explicitly asks for static scoring. Do not assume another range implements
-  that option or uses an answer key.
+- For an Active Directory range, live verification is the console-agent default.
+  Pass `--live-verify=false` only when the operator explicitly asks for static
+  scoring. Do not assume another range implements that option or uses an answer
+  key.
 - If the operator does not provide a report path, ask for one.
 - Do not override cloud selectors; the session fixes the range context.

console/backend/tests/test_chat.py

@@ -695,6 +695,33 @@ async def fake_run_cli(app, sid, cmd, args):  # noqa: ANN001, ANN202
     assert calls == [("/up", ["--from", "x"])], calls
     assert "succeeded" in out, out
 
+    # Built-in AD scoring is live by default even on ordinary prose turns,
+    # where command-specific slash guidance is not injected.
+    score_tool = agent._make_run_dreadgoad(
+        object(),
+        "s-1",
+        fake_run_cli,
+        project_root="/repo",
+        session_dir="/session",
+        capabilities={
+            "/score": {
+                "handler_type": "builtin",
+                "profile": "active-directory",
+            }
+        },
+    )
+    calls.clear()
+    await score_tool.fn(command="/score", args=["/home/kali/report.jsonl"])
+    assert calls == [("/score", ["/home/kali/report.jsonl", "--live-verify"])], calls
+    calls.clear()
+    await score_tool.fn(
+        command="/score",
+        args=["/home/kali/report.jsonl", "--live-verify=false"],
+    )
+    assert calls == [("/score", ["/home/kali/report.jsonl", "--live-verify=false"])], (
+        calls
+    )
+
     # a read command → also runnable by the agent now
     calls.clear()
     await tool.fn(command="/instances", args=[])
@@ -856,6 +883,53 @@ def test_instructions_renders_system_prompt() -> None:
     print("PASS test_instructions_renders_system_prompt")
 
 
+def test_create_agent_derives_commands_from_capabilities() -> None:
+    """The factory gives its prompt and tool one capability-derived command set."""
+    with tempfile.TemporaryDirectory() as session_dir:
+        session = {
+            "session_dir": session_dir,
+            "anchor": {"config_path": str(_REPO / "dreadgoad.yaml"), "env": "dev"},
+            "snapshot": {},
+        }
+        capabilities = {
+            "/score": {"supported": True},
+            "/health": {"supported": False},
+        }
+        captured: dict[str, object] = {}
+
+        def fake_task_agent(**kwargs):  # noqa: ANN003, ANN202
+            captured.update(kwargs)
+            return types.SimpleNamespace(**kwargs)
+
+        async def fake_run_cli(*_args):  # noqa: ANN002, ANN202
+            return (0, "")
+
+        original = agent.LocalTaskAgent
+        agent.LocalTaskAgent = fake_task_agent
+        try:
+            agent.create_agent(
+                "openrouter/test/model",
+                session,
+                object(),
+                "s-1",
+                fake_run_cli,
+                capabilities,
+            )
+        finally:
+            agent.LocalTaskAgent = original
+
+        instructions = str(captured["instructions"])
+        catalog = instructions.split("## Authoritative commands for this session", 1)[
+            1
+        ].split("## Current session context", 1)[0]
+        run_tool = captured["tools"][1]  # type: ignore[index]
+        assert "`/score`" in catalog and "`/health`" not in catalog, catalog
+        assert (
+            "/score" in run_tool.description and "/health" not in run_tool.description
+        )
+    print("PASS test_create_agent_derives_commands_from_capabilities")
+
+
 async def test_health_emits_report_and_suppresses_json() -> None:
     """/health emits a structured health_report, not raw JSON, and stays running."""
     with tempfile.TemporaryDirectory() as d:
@@ -1704,6 +1778,7 @@ async def _main() -> None:
         await test_run_dreadgoad_tool_validates_and_runs()
         await test_direct_command_rejects_extra_args()
         test_instructions_renders_system_prompt()
+        test_create_agent_derives_commands_from_capabilities()
         await test_health_emits_report_and_suppresses_json()
         await test_swap_model_preserves_thread_and_persists()
         await test_thread_persisted_and_restored_on_agent_rebuild()

console/backend/tests/test_commands.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import asyncio
+import dataclasses
 import os
 import pathlib
 import re
@@ -258,6 +259,51 @@ def test_expand_command_prompt() -> None:
     print("PASS test_expand_command_prompt")
 
 
+def test_agent_score_defaults_are_range_aware() -> None:
+    builtin_ad = {
+        "handler_type": "builtin",
+        "profile": "active-directory",
+    }
+    report = ["/home/kali/report.jsonl"]
+    assert commands.apply_agent_defaults("/score", report, builtin_ad) == [
+        *report,
+        "--live-verify",
+    ]
+    assert commands.apply_agent_defaults(
+        "/score", [*report, "--live-verify=false"], builtin_ad
+    ) == [*report, "--live-verify=false"]
+    assert (
+        commands.apply_agent_defaults(
+            "/score",
+            report,
+            {"handler_type": "executable", "profile": "custom"},
+        )
+        == report
+    )
+    assert commands.apply_agent_defaults("/score", [], builtin_ad) == []
+
+    policy = commands.REGISTRY["/score"].agent_policy
+    assert policy is not None
+    assert policy.default_flags == ("--live-verify",)
+    assert "--live-verify=false" in policy.guidance
+
+    # Runtime defaults remain independent from the human-facing wording.
+    original = commands.REGISTRY["/score"]
+    commands.REGISTRY["/score"] = dataclasses.replace(
+        original,
+        agent_policy=dataclasses.replace(policy, guidance=""),
+    )
+    try:
+        assert commands.apply_agent_defaults("/score", report, builtin_ad) == [
+            *report,
+            "--live-verify",
+        ]
+        assert commands.agent_command_guidance("/score", builtin_ad) is None
+    finally:
+        commands.REGISTRY["/score"] = original
+    print("PASS test_agent_score_defaults_are_range_aware")
+
+
 def test_console_readme_covers_command_catalog() -> None:
     """The documented command table must match backend + client commands."""
     readme = pathlib.Path(__file__).resolve().parents[2] / "README.md"
@@ -342,6 +388,20 @@ def test_system_prompt_covers_the_registry() -> None:
     # /scrub's console default is inverted vs the CLI's, so the prompt has to say
     # so — otherwise the agent treats a bare /scrub as the CLI's dry run.
     assert "deletes for real; add 'dry' to preview" in rendered
+
+    rendered_ad = agent._instructions(
+        {
+            "anchor": {"config_path": "/r/dreadgoad.yaml", "env": "dev"},
+            "snapshot": {"provider": "aws", "lab": "ad/GOAD"},
+        },
+        capabilities={
+            "/score": {
+                "handler_type": "builtin",
+                "profile": "active-directory",
+            }
+        },
+    )
+    assert "--live-verify=false" in rendered_ad
     print("PASS test_system_prompt_covers_the_registry")
 
 
@@ -1647,6 +1707,7 @@ def main() -> None:
     test_dispatch_and_agent_commands()
     test_agent_runnable_commands_follow_range_capabilities()
     test_expand_command_prompt()
+    test_agent_score_defaults_are_range_aware()
     test_console_readme_covers_command_catalog()
     test_load_prompt_and_guidance_injection()
     test_exec_verb_and_json_flag()