Use agent audit for runtime env

Author: homanp

pithos/cli.py

@@ -16,7 +16,13 @@
 
 from .events import event_mode_from_env, make_event_sink
 from .exec_backend import SANDBOX_MODES
-from .repo_runner import DEFAULT_REPO_VOTES, allocate_repo_run_dir, run_repo_static
+from .env_discovery import env_hint_report_from_agent_audit
+from .repo_runner import (
+    DEFAULT_REPO_VOTES,
+    allocate_repo_run_dir,
+    run_repo_env_audit,
+    run_repo_static,
+)
 from .repo_source import DEFAULT_REPO_CACHE, redact_secrets, resolve_repo_source
 from .runtime_template import generate_runtime_profile_template, write_runtime_profile_template
 from .runtime_verifier import run_runtime_preflight, run_verify_repo
@@ -335,7 +341,23 @@ def _cmd_run(args: argparse.Namespace) -> int:
             allocate_repo_run_dir(args.results_dir, repo_path) if args.execute_app else None
         )
         runtime_preflight = None
+        env_audit_report = None
         if args.execute_app:
+            if args.runtime_profile is None:
+                env_audit = asyncio.run(
+                    run_repo_env_audit(
+                        repo_path=repo_path,
+                        provider=args.provider,
+                        model=args.model,
+                        agent_env=agent_env,
+                        results_dir=args.results_dir,
+                        pi_config_dir=pi_config_dir,
+                        sandbox_mode=args.sandbox_mode,
+                        out_dir=run_out_dir,
+                        event_sink=event_sink,
+                    )
+                )
+                env_audit_report = env_hint_report_from_agent_audit(env_audit)
             runtime_preflight = run_runtime_preflight(
                 repo_path=repo_path,
                 profile_path=args.runtime_profile,
@@ -347,6 +369,7 @@ def _cmd_run(args: argparse.Namespace) -> int:
                 agent_env=agent_env,
                 pi_config_dir=pi_config_dir,
                 sandbox_mode=args.sandbox_mode,
+                env_audit_report=env_audit_report,
                 event_sink=event_sink,
             )
         scan = asyncio.run(
@@ -380,6 +403,7 @@ def _cmd_run(args: argparse.Namespace) -> int:
             pi_config_dir=pi_config_dir,
             sandbox_mode=args.sandbox_mode,
             preflight_result=runtime_preflight,
+            env_audit_report=env_audit_report,
             event_sink=event_sink,
         )
     except KeyboardInterrupt:

pithos/env_discovery.py

@@ -31,7 +31,8 @@
 }
 MAX_SOURCE_FILES = 500
 MAX_FILE_BYTES = 512 * 1024
-REQUIRED_ENV_SOURCES = {"env_example", "github_actions"}
+AGENT_RUNTIME_SOURCE = "agent_runtime"
+REQUIRED_ENV_SOURCES = {"env_example", "github_actions", AGENT_RUNTIME_SOURCE}
 
 _ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
 _ENV_ASSIGN_RE = re.compile(r"^\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=")
@@ -188,6 +189,25 @@ def collect_env_hints(repo: Path) -> EnvHintReport:
     )
 
 
+def env_hint_report_from_agent_audit(audit: dict[str, Any]) -> EnvHintReport:
+    """Normalize agent-classified runtime env audit output into env hints."""
+    hints: list[EnvHint] = []
+    for item in _audit_required_items(audit):
+        name = _audit_item_name(item)
+        if not name or not is_env_name(name):
+            continue
+        hints.append(
+            EnvHint(
+                name=name,
+                source=AGENT_RUNTIME_SOURCE,
+                path=_audit_item_path(item),
+                confidence=_audit_item_confidence(item),
+                evidence=_audit_item_evidence(item),
+            )
+        )
+    return EnvHintReport(hints=_dedupe_hints(hints))
+
+
 def env_names_from_examples(repo: Path) -> list[str]:
     """Return env var names declared in example env files."""
     return sorted({hint.name for hint in _env_example_hints(repo.resolve(), [])})
@@ -299,6 +319,47 @@ def _source_paths(repo: Path) -> Iterable[Path]:
         yield path
 
 
+def _audit_required_items(audit: dict[str, Any]) -> Iterable[Any]:
+    for key in ("required_env", "required", "env"):
+        value = audit.get(key)
+        if isinstance(value, list):
+            return value
+    return []
+
+
+def _audit_item_name(item: Any) -> str | None:
+    if isinstance(item, str):
+        return item
+    if isinstance(item, dict):
+        value = item.get("name") or item.get("variable") or item.get("env")
+        return str(value) if value else None
+    return None
+
+
+def _audit_item_path(item: Any) -> str:
+    if not isinstance(item, dict):
+        return "ENV-AUDIT.json"
+    sources = item.get("sources") or item.get("source_files") or item.get("files")
+    if isinstance(sources, list) and sources:
+        return str(sources[0])
+    value = item.get("path") or item.get("source") or "ENV-AUDIT.json"
+    return str(value)
+
+
+def _audit_item_confidence(item: Any) -> str:
+    if not isinstance(item, dict):
+        return "medium"
+    value = str(item.get("confidence") or "medium").lower()
+    return value if value in {"high", "medium", "low"} else "medium"
+
+
+def _audit_item_evidence(item: Any) -> str | None:
+    if not isinstance(item, dict):
+        return None
+    value = item.get("reason") or item.get("evidence") or item.get("why")
+    return str(value)[:240] if value else None
+
+
 def _safe_source_evidence(match: str) -> str:
     return match[:120]
 

pithos/repo_runner.py

@@ -161,6 +161,8 @@ def _json_candidates(text: str) -> list[tuple[Any, str]]:
 
 
 def _expected_json_shape(tag: str) -> type | None:
+    if tag == "env_audit_json":
+        return dict
     if tag == "findings_json":
         return list
     if tag == "triage_json":
@@ -175,6 +177,8 @@ def _matches_expected_json(value: Any, expected: type | None, tag: str) -> bool:
         return all(isinstance(item, dict) for item in value)
     if tag == "triage_json":
         return isinstance(value.get("findings"), list)
+    if tag == "env_audit_json":
+        return isinstance(value.get("required_env"), list)
     return True
 
 
@@ -249,6 +253,49 @@ def _threat_prompt(repo_meta: dict[str, Any], stack: dict[str, Any]) -> str:
 """
 
 
+def _env_audit_prompt(repo_meta: dict[str, Any], stack: dict[str, Any]) -> str:
+    return f"""\
+Determine the environment variables required to run the repository mounted read-only at {REPO_MOUNT}
+as a meaningful local/staging application, not merely to compile a public page.
+
+Repository metadata:
+{json.dumps(repo_meta, indent=2)}
+
+Detected stack hints:
+{json.dumps(stack, indent=2)}
+
+Inspect source and configuration only. Do not execute project code. Decide which env vars a user
+would need to configure for the app's primary runtime functionality to work in local/staging,
+including auth/session, database, first-party integrations, webhooks, billing if it is part of the
+product, scheduled/background workflows, and worker/sandbox callbacks that the app itself invokes.
+Use package scripts, app framework config, middleware, root layouts, server/client initialization
+code, app route handlers, workflow/worker definitions, .env.example, docs, and CI configuration as
+evidence.
+
+Exclude variables only when they are clearly unrelated to running the application: tests/fixtures,
+lint/build tooling, CI/deploy metadata, analytics or observability that safely degrades, source-map
+upload settings, alternate provider fallbacks not used by the primary configured path, or
+vulnerability verification personas.
+
+For each required variable, include:
+- name
+- confidence: high|medium|low
+- reason: one sentence explaining why app runtime needs it
+- sources: list of repo-relative evidence files
+
+Return only:
+<env_audit_json>
+{{ "required_env": [ ... ], "optional_env": [ ... ], "summary": {{ ... }} }}
+</env_audit_json>
+<env_audit_markdown>
+...human-readable markdown...
+</env_audit_markdown>
+
+The <env_audit_json> tag is mandatory. If no variables are required to boot, return
+{{ "required_env": [], "optional_env": [], "summary": {{ "required": 0 }} }}.
+"""
+
+
 def _scan_prompt(threat_model: str, max_findings: int, dependency_advisory_summary: str) -> str:
     return f"""\
 Perform a static vulnerability scan of the repository mounted read-only at {REPO_MOUNT}.
@@ -354,6 +401,30 @@ def _triage_prompt(
 
 def _repair_prompt(tag: str, previous_text: str, context: dict[str, Any] | None = None) -> str:
     context_text = json.dumps(context or {}, indent=2)
+    if tag == "env_audit_json":
+        return f"""\
+The previous runtime environment audit did not emit the required machine-readable artifact.
+Reformat the previous answer into the exact artifacts below. Use only the previous answer and
+context; do not invent required env vars that were not stated.
+
+Context:
+<context_json>
+{context_text}
+</context_json>
+
+Previous answer:
+<previous_answer>
+{previous_text}
+</previous_answer>
+
+Return only:
+<env_audit_json>
+{{ "required_env": [ ... ], "optional_env": [ ... ], "summary": {{ ... }} }}
+</env_audit_json>
+<env_audit_markdown>
+...human-readable markdown summary...
+</env_audit_markdown>
+"""
     if tag == "findings_json":
         return f"""\
 The previous static vulnerability scan did not emit the required machine-readable artifact.
@@ -406,6 +477,92 @@ def _repair_prompt(tag: str, previous_text: str, context: dict[str, Any] | None
     raise ValueError(f"unsupported repair tag: {tag}")
 
 
+async def run_repo_env_audit(
+    *,
+    repo_path: Path,
+    provider: str,
+    model: str,
+    agent_env: dict[str, str],
+    results_dir: Path,
+    pi_config_dir: Path | None = None,
+    sandbox_mode: str = "docker",
+    out_dir: Path | None = None,
+    event_sink: EventSinkLike = NULL_EVENT_SINK,
+) -> dict[str, Any]:
+    repo_path = repo_path.resolve()
+    backend = create_execution_backend(sandbox_mode)
+    out_dir = out_dir or allocate_repo_run_dir(results_dir, repo_path)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    repo_meta = {"path": str(repo_path), **_git_metadata(repo_path)}
+    stack = _detect_stack(repo_path)
+
+    audit_text, err = await _run_stage(
+        prompt=_env_audit_prompt(repo_meta, stack),
+        provider=provider,
+        model=model,
+        agent_env=agent_env,
+        repo_path=repo_path,
+        pi_config_dir=pi_config_dir,
+        container_name=f"repo_{_safe_repo_name(repo_path)}_env",
+        transcript_path=out_dir / "env_audit_transcript.jsonl",
+        progress_prefix="[repo:env]",
+        primary_tag="env_audit_json",
+        allow_web=False,
+        backend=backend,
+        event_sink=event_sink,
+    )
+    audit_value, _audit_source = _extract_json_artifact(audit_text, "env_audit_json")
+    if audit_value is _MISSING or not isinstance(audit_value, dict):
+        repair_text, repair_err = await _run_stage(
+            prompt=_repair_prompt(
+                "env_audit_json",
+                audit_text,
+                {"repo": repo_meta, "stack": stack, "transport_error": err},
+            ),
+            provider=provider,
+            model=model,
+            agent_env=agent_env,
+            repo_path=repo_path,
+            pi_config_dir=pi_config_dir,
+            container_name=f"repo_{_safe_repo_name(repo_path)}_env_repair",
+            transcript_path=out_dir / "env_audit_repair_transcript.jsonl",
+            progress_prefix="[repo:env:repair]",
+            primary_tag="env_audit_json",
+            allow_web=False,
+            tools=[],
+            backend=backend,
+            event_sink=event_sink,
+        )
+        repaired_value, _repaired_source = _extract_json_artifact(repair_text, "env_audit_json")
+        if repaired_value is _MISSING or not isinstance(repaired_value, dict):
+            errors = [item for item in (err, repair_err, "env_audit: missing artifact") if item]
+            audit_value = {"required_env": [], "optional_env": [], "summary": {}, "errors": errors}
+        else:
+            audit_text = repair_text
+            audit_value = repaired_value
+    elif err:
+        audit_value = {**audit_value, "errors": [err]}
+
+    audit_markdown = parse_xml_tag(audit_text, "env_audit_markdown") or env_audit_markdown(
+        audit_value
+    )
+    _write(
+        out_dir / "ENV-AUDIT.json",
+        json.dumps(audit_value, indent=2),
+        event_sink=event_sink,
+        stage="repo:env",
+        artifact_type="env_audit_json",
+    )
+    _write(
+        out_dir / "ENV-AUDIT.md",
+        audit_markdown,
+        event_sink=event_sink,
+        stage="repo:env",
+        artifact_type="env_audit_markdown",
+    )
+    return audit_value
+
+
 async def _run_stage(
     *,
     prompt: str,
@@ -767,6 +924,27 @@ async def run_repo_static(
     )
 
 
+def env_audit_markdown(audit: dict[str, Any]) -> str:
+    lines = ["# Runtime Environment Audit", ""]
+    required = audit.get("required_env") if isinstance(audit.get("required_env"), list) else []
+    if required:
+        lines += ["## Required", ""]
+        for item in required:
+            if isinstance(item, str):
+                lines.append(f"- `{item}`")
+            elif isinstance(item, dict):
+                name = item.get("name") or item.get("variable") or item.get("env")
+                reason = item.get("reason") or item.get("evidence")
+                suffix = f" - {reason}" if reason else ""
+                lines.append(f"- `{name}`{suffix}")
+    else:
+        lines.append("No required runtime environment variables were identified.")
+    if audit.get("errors"):
+        lines += ["", "## Errors", ""]
+        lines.extend(f"- {error}" for error in audit["errors"])
+    return "\n".join(lines)
+
+
 def _run_markdown(summary: dict[str, Any]) -> str:
     return f"""\
 # Autonomous Repo Static Review