Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,12 @@ found in a later PowerShell window, confirm that the Python installer's Scripts
directory is on PATH; the `$Scripts` lines above make it available immediately
in the current window.

`argus doctor` is an active repair command. By default it launches an installed
Agent CLI in the real Argus directories with tools enabled, lets the Agent
inspect and fix the machine, then reruns deterministic checks. Use
`argus doctor --advisor none` only when you want diagnostics without an Agent
repair turn.

Until the first versioned PyPI release, refresh the moving GitHub preview with:

```powershell
Expand Down
4 changes: 4 additions & 0 deletions README.zh-CN.md
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,10 @@ argus
一次禁止工具调用的 Agent turn。上面的 `$Scripts` 命令会让当前 PowerShell 立即找到
`argus`;如果新窗口仍找不到,再确认 Python 安装器的 Scripts 目录已加入 PATH。

`argus doctor` 是主动修复命令:默认会在真实 Argus 目录中启动用户电脑上已安装的
Agent CLI,开放工具让 Agent 直接检查并修复机器,然后重新运行确定性检查验收。
只有需要“纯诊断、不启动 Agent 修复”时才使用 `argus doctor --advisor none`。

正式 PyPI 版本发布前,用下面的命令刷新持续更新的 GitHub Preview:

```powershell
Expand Down
15 changes: 13 additions & 2 deletions argus_skill/adapters/agent_cli_backend/_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
import threading
from dataclasses import replace
from pathlib import Path
from typing import Any
from typing import Any, Iterable

from ...core.models import RunnerOptions, RunnerResult
from ...core.secret_guard import known_secret_values
Expand Down Expand Up @@ -94,6 +94,7 @@ def __init__(
default_watchdog_hard_idle_seconds: int = _RUNNER_DEFAULT_HARD_IDLE_SECONDS,
before_exec=None,
event_callback=None,
known_secret_values_override: Iterable[str] | None = None,
) -> None:
deps = load_agent_cli_runtime()
self._deps = deps
Expand Down Expand Up @@ -140,7 +141,11 @@ def __init__(
self._usage_project_root: Path | None = None
self._usage_global_root: Path | None = None
self._usage_mission_id: str | None = None
self._known_secret_values = known_secret_values()
self._known_secret_values_override = tuple(
known_secret_values_override or ()
)
self._known_secret_values: tuple[str, ...] = ()
self._refresh_known_secret_values()

@property
def backend(self) -> str:
Expand Down Expand Up @@ -210,6 +215,12 @@ def _usage_context_snapshot(
self._usage_global_root,
)

def _refresh_known_secret_values(self) -> None:
self._known_secret_values = tuple(dict.fromkeys((
*self._known_secret_values_override,
*known_secret_values(),
)))

def _configured_pricing_model(self, *, profile: str = "") -> str:
"""Read the implicit model from Codex's own config, never another route."""
if not self._is_codex:
Expand Down
3 changes: 1 addition & 2 deletions argus_skill/adapters/agent_cli_backend/_exec.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
from typing import TYPE_CHECKING

from ...core.models import RunnerOptions, RunnerResult
from ...core.secret_guard import known_secret_values
from ._exec_admission import admit
from ._exec_context import _ExecContext
from ._exec_spawn import spawn_and_finish
Expand All @@ -43,7 +42,7 @@ def execute(
run_label: str,
resume_thread_id: str | None = None,
) -> RunnerResult:
backend._known_secret_values = known_secret_values()
backend._refresh_known_secret_values()
# Pin Codex's implicit config model before any accounting or execution.
# The generated command, reservation, and settled usage record therefore
# share one model id instead of independently guessing after the call.
Expand Down
45 changes: 37 additions & 8 deletions argus_skill/apps/cli/_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -784,42 +784,71 @@ def _cmd_doctor(args: argparse.Namespace) -> int:
include_backend=True,
probe_auth=bool(getattr(args, "deep", False)),
)
payload = report.to_jsonable()
payload["verification"] = bool(getattr(args, "verify", False))
if repair_payload is not None:
payload["repair"] = repair_payload
from ...maintenance.advisor import run_doctor_advisor

advisor = run_doctor_advisor(
report,
context,
requested=str(getattr(args, "advisor", "auto") or "auto"),
probe_auth=bool(getattr(args, "deep", False)),
)
if advisor.get("attempts"):
report = run_full_doctor(
context,
include_backend=True,
probe_auth=bool(getattr(args, "deep", False)),
)
repaired_with_tools = any(
bool(item.get("tool_activity_observed"))
for item in advisor.get("attempts", ())
)
if report.ok and repaired_with_tools and advisor["status"] == "failed":
advisor["status"] = "completed"
advisor["error"] = ""
advisor["analysis"] = (
advisor.get("analysis")
or "Agent repairs passed final deterministic verification."
)
advisor["recovered_by_final_verification"] = True
advisor["verified"] = report.ok
advisor["remaining_findings"] = [
item.code for item in report.findings if not item.ok
]
payload = report.to_jsonable()
payload["verification"] = bool(getattr(args, "verify", False))
if repair_payload is not None:
payload["repair"] = repair_payload
agent_ok = advisor["status"] in {"completed", "disabled"}
payload["deterministic_ok"] = report.ok
payload["ok"] = report.ok and agent_ok
payload["advisor"] = advisor
if bool(getattr(args, "json", False)):
sys.stdout.write(json.dumps(payload, ensure_ascii=False, indent=2) + "\n")
else:
sys.stdout.write(render_full_report(report) + "\n")
if advisor["status"] == "completed":
sys.stdout.write(
f"\nCode Agent analysis ({advisor['backend']}):\n"
f"\nCode Agent repair ({advisor['backend']}):\n"
f"{advisor['analysis'].strip()}\n"
)
elif advisor["status"] == "failed":
sys.stdout.write(
f"\nCode Agent analysis failed ({advisor['backend']}): "
f"\nCode Agent repair failed ({advisor['backend']}): "
f"{advisor['error']}\n"
)
if advisor.get("analysis"):
sys.stdout.write(f"{advisor['analysis'].strip()}\n")
elif advisor["status"] == "unavailable":
sys.stdout.write(
"\nCode Agent analysis unavailable: no supported Agent CLI was "
"\nCode Agent repair unavailable: no supported Agent CLI was "
"found on PATH. Deterministic findings above are still valid.\n"
)
if repair_payload is not None:
sys.stdout.write(
f"safe repair plan {repair_payload['plan_id']}: "
f"{repair_payload['status']}\n"
)
return 0 if report.ok else 3
return 0 if report.ok and agent_ok else 3


def _cmd_repair(args: argparse.Namespace) -> int:
Expand Down
17 changes: 14 additions & 3 deletions argus_skill/apps/cli/_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -474,7 +474,7 @@ def build_parser() -> argparse.ArgumentParser:
subparsers = parser.add_subparsers(dest="command")
doctor_parser = subparsers.add_parser(
"doctor",
help="Run read-only Argus diagnostics",
help="Diagnose and repair Argus with an installed Code Agent",
)
doctor_parser.add_argument(
"--json",
Expand All @@ -498,9 +498,20 @@ def build_parser() -> argparse.ArgumentParser:
)
doctor_parser.add_argument(
"--advisor",
choices=("auto", "none", "copilot", "codex", "claude", "opencode", "pi", "grok"),
choices=(
"auto",
"none",
"copilot",
"codex",
"claude",
"opencode",
"pi",
"grok",
"qoder",
"dsh",
),
default="auto",
help="ask an installed Code Agent to interpret sanitized findings (default: auto)",
help="ask an installed Code Agent to inspect and repair Argus (default: auto)",
)
repair_parser = subparsers.add_parser(
"repair",
Expand Down
163 changes: 123 additions & 40 deletions argus_skill/core/agent_probe.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
"""Read-only Agent CLI probe shared by setup and Doctor."""
"""Agent CLI turns used by setup verification and Doctor repair."""
from __future__ import annotations

import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Sequence


@dataclass(frozen=True)
Expand All @@ -13,6 +15,64 @@ class AgentProbeResult:
ok: bool
output: str = ""
error: str = ""
tool_activity_observed: bool = False


def _probe_result(
result: Any,
*,
backend: str,
executable: str,
reject_tool_activity: bool,
require_tool_activity: bool = False,
) -> AgentProbeResult:
output = str(getattr(result, "last_agent_message", "") or "").strip()
if not output:
messages = list(getattr(result, "agent_messages", None) or ())
output = next(
(str(message).strip() for message in reversed(messages) if str(message).strip()),
"",
)
exit_code = int(getattr(result, "exit_code", 1) or 0)
fatal_error = str(getattr(result, "fatal_error", "") or "").strip()
turn_completed = getattr(result, "turn_completed", None)
completion_ok = (
bool(turn_completed)
if turn_completed is not None
else exit_code == 0 and not fatal_error
)
tool_activity = bool(getattr(result, "tool_activity_observed", False))
ok = (
exit_code == 0
and completion_ok
and bool(output)
and not (reject_tool_activity and tool_activity)
and not (require_tool_activity and not tool_activity)
)
error = ""
if not ok:
if reject_tool_activity and tool_activity:
error = "Agent used a tool during the tool-free verification turn"
elif require_tool_activity and not tool_activity:
error = "Agent returned without inspecting or repairing with tools"
else:
error = fatal_error
if not error:
stderr = list(getattr(result, "stderr_lines", None) or ())
error = str(stderr[-1]).strip() if stderr else ""
if not error:
error = (
f"Agent CLI exited {getattr(result, 'exit_code', 'unknown')} "
"without a completed assistant reply"
)
return AgentProbeResult(
backend=backend,
executable=executable,
ok=ok,
output=output,
error=error,
tool_activity_observed=tool_activity,
)


def run_read_only_agent_prompt(
Expand Down Expand Up @@ -71,48 +131,71 @@ def run_read_only_agent_prompt(
error=f"{type(exc).__name__}: {exc}",
)

output = str(getattr(result, "last_agent_message", "") or "").strip()
if not output:
messages = list(getattr(result, "agent_messages", None) or ())
output = next(
(str(message).strip() for message in reversed(messages) if str(message).strip()),
"",
)
exit_code = int(getattr(result, "exit_code", 1) or 0)
fatal_error = str(getattr(result, "fatal_error", "") or "").strip()
turn_completed = getattr(result, "turn_completed", None)
completion_ok = (
bool(turn_completed)
if turn_completed is not None
else exit_code == 0 and not fatal_error
)
ok = (
exit_code == 0
and completion_ok
and bool(output)
and not bool(getattr(result, "tool_activity_observed", False))
return _probe_result(
result,
backend=backend,
executable=executable,
reject_tool_activity=True,
)
error = ""
if not ok:
if bool(getattr(result, "tool_activity_observed", False)):
error = "Agent used a tool during the tool-free verification turn"
else:
error = fatal_error
if not error:
stderr = list(getattr(result, "stderr_lines", None) or ())
error = str(stderr[-1]).strip() if stderr else ""
if not error:
error = (
f"Agent CLI exited {getattr(result, 'exit_code', 'unknown')} "
"without a completed assistant reply"
)
return AgentProbeResult(


def run_agent_repair_prompt(
*,
backend: str,
executable: str,
prompt: str,
working_dir: Path,
add_dirs: Sequence[Path] = (),
known_secret_values: Sequence[str] = (),
model: str = "",
run_label: str = "doctor-repair",
) -> AgentProbeResult:
"""Run one installed Agent with tools enabled so it can repair Argus."""
from ..adapters.agent_cli_backend import AgentCliBackend
from .models import RunnerOptions
from .run_gateway import run_exec

try:
runner = AgentCliBackend(
backend=backend,
runner_bin=executable,
default_watchdog_soft_idle_seconds=30,
default_watchdog_stalled_idle_seconds=120,
default_watchdog_hard_idle_seconds=600,
known_secret_values_override=known_secret_values,
)
result = run_exec(
runner,
prompt=prompt,
resume_thread_id=None,
options=RunnerOptions(
model=model or None,
working_dir=str(working_dir),
add_dirs=[str(path) for path in add_dirs] or None,
dangerous_yolo=True,
full_auto=True,
skip_git_repo_check=True,
),
run_label=run_label,
)
except (OSError, RuntimeError, ValueError, subprocess.SubprocessError) as exc:
return AgentProbeResult(
backend=backend,
executable=executable,
ok=False,
error=f"{type(exc).__name__}: {exc}",
)
return _probe_result(
result,
backend=backend,
executable=executable,
ok=ok,
output=output,
error=error,
reject_tool_activity=False,
require_tool_activity=True,
)


__all__ = ["AgentProbeResult", "run_read_only_agent_prompt"]
__all__ = [
"AgentProbeResult",
"run_agent_repair_prompt",
"run_read_only_agent_prompt",
]
Loading
Loading