diff --git a/.env.example b/.env.example index aa772b14..612473cd 100644 --- a/.env.example +++ b/.env.example @@ -122,7 +122,7 @@ WORKSPACE_ROOT=~/code # Path override for the graphify binary. # ALFRED_GRAPHIFY_BIN= -# Path to the graphify graph JSON. +# Checkout-relative path to the graphify graph JSON. # ALFRED_GRAPHIFY_GRAPH=graphify-out/graph.json # Fallback provider when graphify is unavailable (e.g. code-memory). diff --git a/docs/CODE_MEMORY.md b/docs/CODE_MEMORY.md index 208dfc98..dcae0c09 100644 --- a/docs/CODE_MEMORY.md +++ b/docs/CODE_MEMORY.md @@ -393,7 +393,7 @@ Then enable it (or tick it in `alfred batteries` / the desktop battery picker): | `ALFRED_GRAPHIFY_MCP` | `0` (off) | Attach graphify's read-only graph MCP to firings, taking the code-graph slot. | | `ALFRED_GRAPHIFY_FALLBACK` | unset (`code-memory` when enabled through the battery picker) | Explicit engine to use while a repo has no Graphify graph. Set `none` to leave the slot empty instead. | | `ALFRED_GRAPHIFY_BIN` | auto | Override the `graphify-mcp` executable path. Alfred otherwise uses the installed entrypoint. Package installation happens during battery setup, never inside an agent firing. | -| `ALFRED_GRAPHIFY_GRAPH` | `graphify-out/graph.json` | Graph file passed to the MCP server, relative to each firing's repo worktree unless absolute. | +| `ALFRED_GRAPHIFY_GRAPH` | `graphify-out/graph.json` | Graph file passed to the MCP server, relative to each firing's repo worktree. Absolute paths and paths that escape the checkout are rejected. | A firing serves the graph in its own working directory (`graphify-out/graph.json`), so build or update the graph per repo. If that repo has no graph yet, Alfred does diff --git a/docs/CONFIG.md b/docs/CONFIG.md index 0010e306..59de740e 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -126,7 +126,7 @@ are experimental, deep-tuning, or set by Alfred itself at runtime. | `ALFRED_CODE_MEMORY_FETCH_TIMEOUT_S` | int | `120` | internal | Fetch timeout in seconds for the code-memory server. | | `ALFRED_GRAPHIFY_MCP` | bool | `0` | operator | Expose the graphify MCP server. | | `ALFRED_GRAPHIFY_BIN` | path | | operator | Path override for the graphify binary. | -| `ALFRED_GRAPHIFY_GRAPH` | path | `graphify-out/graph.json` | operator | Path to the graphify graph JSON. | +| `ALFRED_GRAPHIFY_GRAPH` | path | `graphify-out/graph.json` | operator | Checkout-relative path to the graphify graph JSON. | | `ALFRED_GRAPHIFY_FALLBACK` | str | | operator | Fallback provider when graphify is unavailable (e.g. code-memory). | | `ALFRED_GRAPH_DENSIFY` | bool | `1` | internal | Enable graph densification projection; on by default. | | `ALFRED_GBRAIN_BIN` | path | | internal | Path to an external graph-brain binary. | diff --git a/docs/DESKTOP_CLIENT.md b/docs/DESKTOP_CLIENT.md index 40ff8adb..e328c4f5 100644 --- a/docs/DESKTOP_CLIENT.md +++ b/docs/DESKTOP_CLIENT.md @@ -58,6 +58,12 @@ The top row should feel like an operations status strip, not a dashboard full of Ask owns plain-language intake and the planning inbox. Plan cards show: +When one selected repository is in scope and its full GitHub slug maps to a +verified local checkout, Ask runs its interrogator from that checkout. Claude +may only read, grep, and glob, and Codex is pinned to its read-only sandbox. A +missing, bare-name, unselected, or multi-repository mapping falls back to +Alfred's state directory instead of granting filesystem scope by guesswork. + - parent issue and Slack thread - readiness verdict - affected repos and rollout order diff --git a/lib/agent_runner/process.py b/lib/agent_runner/process.py index 921e75a5..47aafe85 100644 --- a/lib/agent_runner/process.py +++ b/lib/agent_runner/process.py @@ -68,6 +68,7 @@ CODEX_APPROVAL_POLICY, CODEX_BIN, CODEX_DEFAULT_SANDBOX, + WORKSPACE, ) from .reliability import ( CircuitBreaker, @@ -502,6 +503,85 @@ def _graphify_entrypoint_works(command: str) -> bool: return False +def _git_common_dir(checkout: Path) -> Path | None: + """Return the shared Git directory for a checkout or linked worktree.""" + + try: + completed = subprocess.run( + ["git", "-C", str(checkout), "rev-parse", "--git-common-dir"], + capture_output=True, + text=True, + timeout=5, + check=False, + env={ + key: value + for key, value in os.environ.items() + if key in {"HOME", "PATH", "TMPDIR", "SYSTEMROOT"} + }, + ) + except (OSError, subprocess.TimeoutExpired): + return None + if completed.returncode != 0: + return None + raw = (completed.stdout or "").strip() + if not raw: + return None + common = Path(raw).expanduser() + if not common.is_absolute(): + common = checkout / common + try: + return common.resolve(strict=True) + except (OSError, RuntimeError): + return None + + +def _graphify_checkout_roots(workdir: Path) -> list[Path]: + """Prefer the configured checkout that owns this firing worktree.""" + + try: + checkout = workdir.resolve(strict=True) + except (OSError, RuntimeError): + return [] + roots: list[Path] = [] + try: + from .github import repo_to_local_map + + mappings = repo_to_local_map(os.environ) + except Exception: + mappings = {} + if mappings: + common = _git_common_dir(checkout) + if common is not None: + for raw in mappings.values(): + candidate = Path(str(raw)).expanduser() + if not candidate.is_absolute(): + candidate = WORKSPACE / candidate + try: + candidate = candidate.resolve(strict=True) + except (OSError, RuntimeError): + continue + if candidate in roots or _git_common_dir(candidate) != common: + continue + roots.append(candidate) + if checkout not in roots: + roots.append(checkout) + return roots + + +def _graphify_graph_path(workdir: Path, graph_path: Path) -> Path | None: + """Resolve one checkout-relative graph without crossing repo boundaries.""" + + for root in _graphify_checkout_roots(workdir): + try: + resolved = (root / graph_path).resolve(strict=True) + resolved.relative_to(root) + except (OSError, RuntimeError, ValueError): + continue + if resolved.is_file(): + return resolved + return None + + def _graphify_mcp_server(workdir: Path | None = None) -> dict[str, Any] | None: """Return the ``mcpServers`` entry for graphify, or ``None`` when disabled or the command is not on PATH. @@ -518,16 +598,15 @@ def _graphify_mcp_server(workdir: Path | None = None) -> dict[str, Any] | None: cmd, prefix = invocation graph = os.environ.get("ALFRED_GRAPHIFY_GRAPH", "").strip() or "graphify-out/graph.json" graph_path = Path(graph).expanduser() - resolved_graph = ( - graph_path if graph_path.is_absolute() else (workdir / graph_path if workdir else None) - ) - if resolved_graph is not None and not resolved_graph.is_file(): + if workdir is None or graph_path.is_absolute(): + return None + resolved_graph = _graphify_graph_path(workdir, graph_path) + if resolved_graph is None: return None - graph_arg = str(resolved_graph) if resolved_graph is not None else str(graph_path) return { GRAPHIFY_MCP_SERVER: { "command": cmd, - "args": [*prefix, graph_arg, "--transport", "stdio"], + "args": [*prefix, str(resolved_graph), "--transport", "stdio"], } } @@ -871,6 +950,7 @@ def claude_invoke_streaming( timeout: int = 1200, resume_session: str | None = None, model: str | None = None, + read_only_isolation: bool = False, _auth_retry: bool = False, ) -> ClaudeResult: """Streaming counterpart of :func:`claude_invoke`. Same return shape. @@ -901,23 +981,46 @@ def claude_invoke_streaming( if max_turns is None: max_turns = _CLAUDE_UNLIMITED_TURNS - memory_script = _memory_mcp_script() + memory_script = None if read_only_isolation else _memory_mcp_script() + resolved_tools = ( + allowed_tools + if read_only_isolation + else _with_memory_mcp_tools(allowed_tools, memory_script, workdir) + ) cmd = [ _runtime_cli_bin("CLAUDE_BIN", CLAUDE_BIN), "-p", prompt, - "--allowedTools", - _with_memory_mcp_tools(allowed_tools, memory_script, workdir), - "--max-turns", - str(max_turns), - "--output-format", - "stream-json", - "--verbose", - "--permission-mode", - "bypassPermissions", ] - cmd.extend(_agent_settings_args()) - cmd.extend(_memory_mcp_args(memory_script, workdir)) + if read_only_isolation: + cmd.extend(["--tools", resolved_tools]) + cmd.extend( + [ + "--allowedTools", + resolved_tools, + "--max-turns", + str(max_turns), + "--output-format", + "stream-json", + "--verbose", + "--permission-mode", + "dontAsk" if read_only_isolation else "bypassPermissions", + ] + ) + if read_only_isolation: + cmd.extend( + [ + "--safe-mode", + "--strict-mcp-config", + "--mcp-config", + '{"mcpServers":{}}', + "--no-session-persistence", + ] + ) + if not read_only_isolation: + cmd.extend(_agent_settings_args()) + if memory_script is not None: + cmd.extend(_memory_mcp_args(memory_script, workdir)) if model: cmd.extend(["--model", model]) if resume_session: @@ -1065,6 +1168,7 @@ def _capture_stdout() -> None: timeout=timeout, resume_session=resume_session, model=model, + read_only_isolation=read_only_isolation, _auth_retry=True, ) return result @@ -1149,6 +1253,8 @@ def codex_invoke( sandbox: str | None = None, approval_policy: str | None = None, bypass_approvals_and_sandbox: bool = False, + ignore_user_config: bool = False, + ephemeral: bool = False, add_dirs: list[Path] | None = None, allowed_tools: str | None = None, max_turns: int | None = None, @@ -1207,6 +1313,10 @@ def codex_invoke( "--cd", str(workdir), ] + if ignore_user_config: + cmd.append("--ignore-user-config") + if ephemeral: + cmd.append("--ephemeral") resolved_sandbox = sandbox or CODEX_DEFAULT_SANDBOX if bypass_approvals_and_sandbox: cmd.append("--dangerously-bypass-approvals-and-sandbox") @@ -1688,6 +1798,9 @@ def invoke_agent_engine( codex_add_dirs: list[Path] | None = None, codex_approval_policy: str | None = None, codex_bypass_approvals_and_sandbox: bool = False, + codex_ignore_user_config: bool = False, + codex_ephemeral: bool = False, + claude_read_only_isolation: bool = False, claude_fn: Callable[..., ClaudeResult] | None = None, codex_fn: Callable[..., ClaudeResult] | None = None, on_fallback: Callable[[ClaudeResult], None] | None = None, @@ -1800,6 +1913,7 @@ def _invoke_claude() -> ClaudeResult: max_turns=claude_max_turns, timeout=timeout, model=claude_model, + read_only_isolation=claude_read_only_isolation, ) def _invoke_codex() -> ClaudeResult: @@ -1813,6 +1927,8 @@ def _invoke_codex() -> ClaudeResult: sandbox=codex_sandbox, approval_policy=codex_approval_policy, bypass_approvals_and_sandbox=codex_bypass_approvals_and_sandbox, + ignore_user_config=codex_ignore_user_config, + ephemeral=codex_ephemeral, add_dirs=codex_add_dirs, ) diff --git a/lib/alfred_config.py b/lib/alfred_config.py index 76694a60..4e375618 100644 --- a/lib/alfred_config.py +++ b/lib/alfred_config.py @@ -947,7 +947,7 @@ def V( "path", "graphify-out/graph.json", "memory", - "Path to the graphify graph JSON.", + "Checkout-relative path to the graphify graph JSON.", operator=True, ), V( diff --git a/lib/compose_converse.py b/lib/compose_converse.py index f838a073..cb85cfbe 100644 --- a/lib/compose_converse.py +++ b/lib/compose_converse.py @@ -387,6 +387,7 @@ def build_repo_grounding( *, workspace_root: Path, repo_to_local: dict[str, str] | None = None, + allow_workspace_fallback: bool = True, ) -> str: """Assemble each target repo's CLAUDE.md (multi-repo aware). @@ -397,11 +398,12 @@ def build_repo_grounding( Two path sources, treated differently. A ``repo_to_local`` (GH_REPO_TO_LOCAL) hit is TRUSTED operator config: the operator may legitimately point a repo at - an absolute checkout outside ``workspace_root``, so it is used as-is. With no - mapping we fall back to the raw request slug's bare name as a directory under - ``workspace_root`` - that name is UNTRUSTED, so it is contained: a traversal - slug is dropped and degrades to the "no local checkout" block rather than - becoming an arbitrary-file-read sink (py/path-injection). + an absolute checkout outside ``workspace_root``, so it is used as-is. Unless + ``allow_workspace_fallback`` is false, an unmapped repo falls back to the raw + request slug's bare name as a directory under ``workspace_root``. That name + is UNTRUSTED, so it is contained: a traversal slug is dropped and degrades + to the "no local checkout" block rather than becoming an arbitrary-file-read + sink (py/path-injection). """ repo_to_local = repo_to_local or {} repos = [repo for repo in repos if repo] @@ -425,11 +427,13 @@ def build_repo_grounding( # be an absolute path outside workspace_root, so honor it as-is # (an absolute ``mapped`` wins the join, mirroring the original). repo_dir: Path | None = Path(workspace_root) / mapped - else: + elif allow_workspace_fallback: # UNTRUSTED: the request slug's bare name as a directory under the # workspace. Contain it so a traversal slug cannot escape; an escape # degrades to the same safe fallback a missing checkout gets. repo_dir = _contained_repo_dir(workspace_root, bare) + else: + repo_dir = None if repo_dir is None: blocks.append( f"{header}\n\nNo local checkout or CLAUDE.md available for this " @@ -485,7 +489,11 @@ def _file_tree_summary(repo_dir: Path, *, limit: int = 80) -> str: return "\n".join(lines) -def load_code_map(code_map_path: Path | None) -> str: +def load_code_map( + code_map_path: Path | None, + *, + repos: Iterable[str] | None = None, +) -> str: """Render the code-map-refresh JSON as compact grounding, if present. Reuses whatever ``code-map-refresh`` last wrote (per-repo endpoints, client @@ -500,6 +508,17 @@ def load_code_map(code_map_path: Path | None) -> str: return "A code map exists but could not be read; rely on the repo docs above." if not isinstance(data, dict): return "A code map exists but is malformed; rely on the repo docs above." + repo_filter = {str(repo).strip().casefold() for repo in repos or () if str(repo).strip()} + repo_filter_names = {repo.rsplit("/", 1)[-1] for repo in repo_filter} + + def matches_repo_filter(value: object) -> bool: + if not repo_filter: + return True + normalized = str(value or "").strip().casefold() + if "/" in normalized: + return normalized in repo_filter + return normalized in repo_filter_names + lines: list[str] = [] generated = str(data.get("generated_at") or "").strip() if generated: @@ -509,6 +528,8 @@ def load_code_map(code_map_path: Path | None) -> str: for slug, info in repos.items(): if not isinstance(info, dict): continue + if not matches_repo_filter(slug): + continue endpoints = info.get("endpoints") or [] routes = info.get("routes") or [] calls = info.get("api_calls") or [] @@ -545,7 +566,13 @@ def load_code_map(code_map_path: Path | None) -> str: lines.append(f"- `{slug}`: " + ", ".join(counts)) drift = data.get("contract_drift") if isinstance(drift, list) and drift: - lines.append(f"Contract drift entries: {len(drift)} (advisory).") + filtered_drift = [ + item + for item in drift + if isinstance(item, dict) and matches_repo_filter(item.get("caller")) + ] + if filtered_drift: + lines.append(f"Contract drift entries: {len(filtered_drift)} (advisory).") return "\n".join(lines) or "Code map present but empty." @@ -1042,6 +1069,7 @@ def _is_build_verb_form(token: str) -> bool: "report", "check", "confirm", + "identify", "inspect", "read", "review", @@ -1426,6 +1454,12 @@ def looks_like_read_only_info_request(text: str) -> bool: return False command_index = 0 + if len(tokens) > 2 and tokens[0] == "in" and "/" in tokens[1]: + # Desktop Ask commonly scopes a repository before the actual command: + # "In owner/repo, identify where X is checked." Treat only the canonical + # owner/repo shape as a prefix; ordinary placement requests such as + # "in settings, add a button" must stay build work. + command_index = 2 for prefix in _READ_ONLY_FORMAT_PREFIXES: if tuple(tokens[: len(prefix)]) == prefix: command_index = len(prefix) @@ -1717,8 +1751,12 @@ def summarize(turns: Sequence[condenser.Turn]) -> str: timeout=CONDENSER_TIMEOUT, claude_max_turns=CONDENSER_MAX_TURNS, claude_model=model, + claude_read_only_isolation=True, codex_model=model, codex_timeout=CONDENSER_TIMEOUT, + codex_sandbox="read-only", + codex_ignore_user_config=True, + codex_ephemeral=True, hybrid_fallback_on_provider_failure=True, ) except Exception: @@ -1846,11 +1884,35 @@ def run_turn( return None if not getattr(result, "success", False) or not getattr(result, "result_text", ""): return None - return parse_turn( + turn = parse_turn( result.result_text, base_draft=base_draft, last_user_message=latest_user_message, ) + if turn is not None: + return turn + + # A live engine can occasionally answer a plain read-only question correctly + # while omitting the requested JSON envelope. Preserve that useful answer + # only when the deterministic classifier independently says this is a + # conversation and no structured draft exists. Malformed build output still + # fails closed so prose can never become an executable plan by accident. + content_draft = replace(base_draft, repos=[]) if base_draft.repos else base_draft + if ( + not _draft_has_content(content_draft) + and classify_message_intent(latest_user_message, draft=content_draft) == INTENT_CONVERSATION + ): + reply = str(result.result_text or "").strip() + if reply and not _reply_claims_plan_or_action(reply): + return ConverseTurn( + reply=reply, + draft=content_draft, + readiness=ConverseReadiness(score=0, ready=False), + done=False, + intent=INTENT_CONVERSATION, + action=None, + ) + return None def _condensed_converse_messages( @@ -1898,7 +1960,11 @@ def _invoke_converse( claude_allowed_tools="Read,Grep,Glob", timeout=timeout, claude_max_turns=DEFAULT_MAX_TURNS, + claude_read_only_isolation=True, codex_timeout=timeout, + codex_sandbox="read-only", + codex_ignore_user_config=True, + codex_ephemeral=True, hybrid_fallback_on_provider_failure=True, ) except Exception: diff --git a/lib/server/routes/converse.py b/lib/server/routes/converse.py index aa508aab..49519a14 100644 --- a/lib/server/routes/converse.py +++ b/lib/server/routes/converse.py @@ -7,12 +7,24 @@ from fastapi import APIRouter, Depends, Request, Response from fastapi.responses import JSONResponse +from starlette.concurrency import run_in_threadpool from server import views router = APIRouter() +def _run_compose_converse_guarded( + request: Request, + body: dict[str, Any], +) -> JSONResponse: + """Keep lock ownership and the complete buffered mutation on one worker.""" + + draft_id = views._safe_compose_draft_id(body.get("draft_id")) + with views._compose_turn_guard(request, draft_id): + return views._run_compose_converse(request, body) + + @router.post( "/api/theme-builder/converse", response_class=JSONResponse, @@ -42,7 +54,7 @@ async def api_theme_builder_converse(request: Request) -> JSONResponse: return JSONResponse({"error": "request body must be JSON"}, status_code=400) if not isinstance(body, dict): return JSONResponse({"error": "request body must be a JSON object"}, status_code=400) - return views._run_theme_builder_converse(request, body) + return await run_in_threadpool(views._run_theme_builder_converse, request, body) @router.post( @@ -74,7 +86,7 @@ async def api_onboarding_converse(request: Request) -> JSONResponse: return JSONResponse({"error": "request body must be JSON"}, status_code=400) if not isinstance(body, dict): return JSONResponse({"error": "request body must be a JSON object"}, status_code=400) - return views._run_onboarding_converse(request, body) + return await run_in_threadpool(views._run_onboarding_converse, request, body) @router.post( @@ -104,7 +116,7 @@ async def api_compose_converse(request: Request) -> JSONResponse: return JSONResponse({"error": "request body must be JSON"}, status_code=400) if not isinstance(body, dict): return JSONResponse({"error": "request body must be a JSON object"}, status_code=400) - return views._run_compose_converse(request, body) + return await run_in_threadpool(_run_compose_converse_guarded, request, body) @router.post("/api/compose/converse/stream") diff --git a/lib/server/routes/plans.py b/lib/server/routes/plans.py index 9a68e3b2..7a0b7e50 100644 --- a/lib/server/routes/plans.py +++ b/lib/server/routes/plans.py @@ -5,6 +5,7 @@ import json import logging from dataclasses import replace +from typing import Any from fastapi import APIRouter, Depends, Request from fastapi.responses import JSONResponse @@ -29,6 +30,18 @@ router = APIRouter() +def _run_planning_mutation( + request: Request, + draft_id: str, + function: Any, + *args: Any, +) -> Any: + """Run one planning mutation with lock ownership confined to this worker.""" + + with views._compose_turn_guard(request, draft_id): + return function(*args) + + @router.get("/api/plans", response_class=JSONResponse) async def api_plans(request: Request, limit: int = 50) -> JSONResponse: rows = request.app.state.reader.list_plans(limit=min(max(1, limit), 200)) @@ -97,6 +110,9 @@ async def api_discard_plan(request: Request, plan_id: str) -> JSONResponse: return JSONResponse({"error": "plan not found"}, status_code=404) try: result = await run_in_threadpool( + _run_planning_mutation, + request, + draft_id, views._discard_planning_draft_group, views._state_root(request), draft_id, @@ -176,11 +192,17 @@ async def api_file_plan_issue(request: Request, plan_id: str) -> JSONResponse: same-origin and token-gated like other local mutations, and it is idempotent via the saved draft's ``bridge.issue_url`` field. """ + draft_id = views._safe_planning_draft_id(plan_id) + if draft_id is None: + return JSONResponse({"error": "plan not found"}, status_code=404) try: result = await run_in_threadpool( + _run_planning_mutation, + request, + draft_id, views._file_planning_draft_issue, views._state_root(request), - plan_id, + draft_id, ) except FileNotFoundError: return JSONResponse({"error": "plan not found"}, status_code=404) @@ -205,21 +227,14 @@ async def api_file_plan_issue(request: Request, plan_id: str) -> JSONResponse: return JSONResponse(result) -@router.post( - "/api/plans/draft", - response_class=JSONResponse, - dependencies=[Depends(views.require_mutation_token)], -) -async def api_compose_draft(request: Request) -> JSONResponse: - try: - body = json.loads((await request.body()).decode("utf-8") or "{}") - except (json.JSONDecodeError, UnicodeDecodeError): - return JSONResponse({"error": "request body must be JSON"}, status_code=400) - if not isinstance(body, dict): - return JSONResponse({"error": "request body must be a JSON object"}, status_code=400) +def _run_compose_draft( + request: Request, + body: dict[str, Any], + draft_id: str | None, +) -> JSONResponse: + """Run one blocking draft refinement while its shared draft lock is held.""" text = str(body.get("text") or "").strip() - draft_id = views._safe_compose_draft_id(body.get("draft_id")) prior_payload, prior_path = views._read_compose_draft_payload(request, draft_id) base_draft = views._compose_base_draft(body, prior_payload) @@ -322,3 +337,29 @@ async def api_compose_draft(request: Request) -> JSONResponse: }, } ) + + +def _run_compose_draft_guarded( + request: Request, + body: dict[str, Any], + draft_id: str | None, +) -> JSONResponse: + with views._compose_turn_guard(request, draft_id): + return _run_compose_draft(request, body, draft_id) + + +@router.post( + "/api/plans/draft", + response_class=JSONResponse, + dependencies=[Depends(views.require_mutation_token)], +) +async def api_compose_draft(request: Request) -> JSONResponse: + try: + body = json.loads((await request.body()).decode("utf-8") or "{}") + except (json.JSONDecodeError, UnicodeDecodeError): + return JSONResponse({"error": "request body must be JSON"}, status_code=400) + if not isinstance(body, dict): + return JSONResponse({"error": "request body must be a JSON object"}, status_code=400) + + draft_id = views._safe_compose_draft_id(body.get("draft_id")) + return await run_in_threadpool(_run_compose_draft_guarded, request, body, draft_id) diff --git a/lib/server/setup.py b/lib/server/setup.py index d8c6bd65..ff3cb8ad 100644 --- a/lib/server/setup.py +++ b/lib/server/setup.py @@ -858,7 +858,16 @@ def capability_status( _context_compression_capability(runtime_env), _engineering_skills_capability(runtime_env), ] - counts = { + counts = _capability_summary(capabilities) + return { + "version": 1, + "summary": counts | {"total": len(capabilities)}, + "capabilities": capabilities, + } + + +def _capability_summary(capabilities: list[dict[str, Any]]) -> dict[str, int]: + return { "ready": sum(1 for item in capabilities if item["state"] == "ready"), "actionable": sum( 1 @@ -867,11 +876,85 @@ def capability_status( ), "disabled": sum(1 for item in capabilities if item["state"] == "disabled"), } - return { - "version": 1, - "summary": counts | {"total": len(capabilities)}, - "capabilities": capabilities, + + +def _reconcile_code_graph_coverage( + capability_plane: dict[str, Any], + coverage: dict[str, Any], +) -> None: + """Reconcile Graphify readiness from verified selected-checkout coverage.""" + + capability = _capability_by_key(capability_plane, "code_graph") + detected = capability.get("detected") or {} + if str(detected.get("engine") or "") != "graphify" or not capability.get("installed"): + return + + covered = [str(repo) for repo in coverage.get("covered", []) if str(repo)] + missing = [str(repo) for repo in coverage.get("missing", []) if str(repo)] + graphify_covered, fallback_covered = _graphify_provider_coverage(coverage, covered) + coverage_ready = bool(coverage.get("ready")) + capability["state"] = "ready" if coverage_ready else "needs_index" + capability["detected"] = dict(detected) | { + "coverage_ready": coverage_ready, + "covered": covered, + "missing": missing, + "graphify_covered": graphify_covered, + "fallback_covered": fallback_covered, } + if coverage_ready: + if fallback_covered and graphify_covered: + capability["detail"] = ( + f"Verified code-graph coverage uses Graphify for {', '.join(graphify_covered)} " + f"and the code-memory fallback for {', '.join(fallback_covered)}." + ) + elif fallback_covered: + capability["detail"] = ( + "Selected repositories have verified code-graph coverage through the " + "code-memory fallback; Graphify graphs are not ready." + ) + else: + capability["detail"] = "Graphify covers all verified selected repositories." + elif missing: + capability["detail"] = ( + "Graphify is installed, but verified graphs are missing for selected " + f"repositories: {', '.join(missing)}." + ) + else: + capability["detail"] = ( + "Graphify is installed, but selected repository coverage is not verified." + ) + capability["install_hint"] = "" if coverage_ready else _graphify_index_install_hint() + + capabilities = capability_plane.get("capabilities") or [] + capability_plane["summary"] = _capability_summary(capabilities) | {"total": len(capabilities)} + + +def _graphify_provider_coverage( + coverage: dict[str, Any], + covered: list[str], +) -> tuple[list[str], list[str]]: + detected = coverage.get("detected") or [] + provider_rows = [row for row in detected if isinstance(row, dict) and "provider" in row] + if not provider_rows: + return covered, [] + graphify_covered = [ + str(row.get("repo") or "") + for row in provider_rows + if row.get("provider") == "graphify" and str(row.get("repo") or "") + ] + fallback_covered = [ + str(row.get("repo") or "") + for row in provider_rows + if row.get("provider") == "code-memory" and str(row.get("repo") or "") + ] + return graphify_covered, fallback_covered + + +def _graphify_index_install_hint() -> str: + battery = batteries.battery_by_id("graphify") + if battery is not None and battery.install_hint: + return battery.install_hint + return "Build a graph for each selected repository with `graphify `." # --------------------------------------------------------------------------- # @@ -1008,14 +1091,12 @@ def _code_graph_capability( if graphify and bool(graphify.get("configured")): installed = bool(graphify.get("installed")) graph_present = bool(graphify.get("graph_present")) - if code_ready and not (installed and graph_present): + if not installed and code_ready: if fallback_enabled and not code_memory.get("enabled"): code_memory = dict(code_memory) code_memory["enabled"] = True code_memory["detail"] = ( - "Code-memory fallback is ready while Graphify awaits a per-repo graph." - if installed - else "Code-memory fallback is ready while Graphify is not installed." + "Code-memory fallback is ready while Graphify is not installed." ) graphify = None else: @@ -1040,6 +1121,7 @@ def _code_graph_capability( "docs": graphify.get("docs"), "graph_path": graphify.get("graph_path"), "graph_present": graph_present, + "fallback": graphify.get("fallback"), }, install_hint=( "" @@ -1708,10 +1790,16 @@ def bootstrap_status() -> dict[str, Any]: any_engine = any(e["installed"] for e in engines) repo_checkouts = _selected_repo_local_paths(repos, runtime_env) code_memory = code_memory_status(runtime_env) - code_memory_coverage = _code_memory_coverage( - repos, code_memory, runtime_env, resolved=repo_checkouts - ) capability_plane = capability_status(code_memory, launcher_env=runtime_env) + code_graph = _capability_by_key(capability_plane, "code_graph") + code_memory_coverage = _selected_code_graph_coverage( + repos, + runtime_env, + code_memory=code_memory, + code_graph=code_graph, + resolved=repo_checkouts, + ) + _reconcile_code_graph_coverage(capability_plane, code_memory_coverage) install = install_inventory(repos=repos, env=runtime_env) first_run = first_run_readiness_status( gh=gh, @@ -1721,6 +1809,7 @@ def bootstrap_status() -> dict[str, Any]: queue_missing=queue_missing, install=install, code_memory=code_memory, + code_memory_coverage=code_memory_coverage, capability_plane=capability_plane, runtime_env=runtime_env, repo_checkouts=repo_checkouts, @@ -1764,6 +1853,7 @@ def first_run_readiness_status( queue_missing: list[str], install: dict[str, Any], code_memory: dict[str, Any], + code_memory_coverage: dict[str, Any], capability_plane: dict[str, Any], runtime_env: dict[str, str], repo_checkouts: list[dict[str, Any]], @@ -1778,7 +1868,11 @@ def first_run_readiness_status( _repo_local_paths_readiness_check(repos, runtime_env, resolved=repo_checkouts), _scheduled_fleet_readiness_check(install), _desktop_token_readiness_check(install), - _code_graph_readiness_check(capability_plane, code_memory), + _code_graph_readiness_check( + capability_plane, + code_memory, + coverage=code_memory_coverage, + ), _context_compression_readiness_check(capability_plane), _engineering_skills_readiness_check(capability_plane), _architect_parent_repo_readiness_check(runtime_env), @@ -2298,6 +2392,17 @@ def _local_repo_github_remote( return next((remote for remote in remotes if remote[0] == "origin"), remotes[0]) +def local_repo_matches_github_slug(path: Path, expected_slug: str) -> bool: + """Return whether a checkout exposes a GitHub remote for the exact slug.""" + + _, remote_slug = _local_repo_github_remote( + path, + expected_slug=expected_slug, + deadline=time.monotonic() + 5.0, + ) + return bool(remote_slug) and remote_slug.casefold() == expected_slug.casefold() + + def _code_memory_coverage( repos: list[str], code_memory: dict[str, Any], @@ -2350,6 +2455,126 @@ def _code_memory_coverage( } +def _graphify_coverage( + repos: list[str], + env: dict[str, str], + *, + provider_ready: bool, + resolved: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Prove that every selected checkout owns the Graphify graph it will serve.""" + + configured = Path(env.get("ALFRED_GRAPHIFY_GRAPH") or "graphify-out/graph.json").expanduser() + covered: list[str] = [] + detected: list[dict[str, Any]] = [] + rows = resolved if resolved is not None else _selected_repo_local_paths(repos, env) + for row in rows: + slug = str(row["repo"]).strip().lower() + checkout = Path(str(row.get("path") or "")).expanduser() + graph = configured if configured.is_absolute() else checkout / configured + contained = False + graph_present = False + try: + checkout = checkout.resolve(strict=True) + graph = graph.resolve(strict=True) + if not configured.is_absolute(): + graph.relative_to(checkout) + contained = True + graph_present = graph.is_file() + except (OSError, RuntimeError, ValueError): + pass + ready = ( + provider_ready + and bool(row.get("ready")) + and bool(row.get("identity_matches")) + and contained + and graph_present + ) + if ready: + covered.append(slug) + detected.append( + { + **row, + "graph_path": str(graph), + "graph_within_checkout": contained, + "graph_present": graph_present, + "covered": ready, + } + ) + missing = [repo for repo in repos if repo.strip().lower() not in set(covered)] + return { + "ready": bool(repos) and not missing, + "covered": covered, + "missing": missing, + "detected": detected, + } + + +def _selected_code_graph_coverage( + repos: list[str], + env: dict[str, str], + *, + code_memory: dict[str, Any], + code_graph: dict[str, Any], + resolved: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Verify each repo against the provider Alfred will actually attach.""" + + detected = code_graph.get("detected") or {} + if str(detected.get("engine") or "") != "graphify": + effective_code_memory = dict(code_memory) + effective_code_memory["enabled"] = bool(code_graph.get("enabled")) + return _code_memory_coverage(repos, effective_code_memory, env, resolved=resolved) + + graphify = _graphify_coverage( + repos, + env, + provider_ready=bool(code_graph.get("enabled")) and bool(code_graph.get("installed")), + resolved=resolved, + ) + if str(detected.get("fallback") or "") != "code-memory": + return graphify + + fallback_status = dict(code_memory) + fallback_status["enabled"] = True + fallback = _code_memory_coverage(repos, fallback_status, env, resolved=resolved) + graphify_rows = { + str(row.get("repo") or "").strip().lower(): row for row in graphify["detected"] + } + fallback_rows = { + str(row.get("repo") or "").strip().lower(): row for row in fallback["detected"] + } + graphify_covered = set(graphify["covered"]) + fallback_covered = set(fallback["covered"]) + covered: list[str] = [] + combined: list[dict[str, Any]] = [] + for repo in repos: + slug = repo.strip().lower() + provider = ( + "graphify" + if slug in graphify_covered + else ("code-memory" if slug in fallback_covered else None) + ) + if provider: + covered.append(slug) + combined.append( + { + "repo": slug, + "covered": bool(provider), + "provider": provider, + "graphify": graphify_rows.get(slug), + "fallback": fallback_rows.get(slug), + } + ) + missing = [repo for repo in repos if repo.strip().lower() not in set(covered)] + return { + "ready": bool(repos) and not missing, + "covered": covered, + "missing": missing, + "detected": combined, + } + + def _first_existing_git_repo_candidate( candidates: list[tuple[Path, str]], ) -> tuple[Path, str] | None: @@ -2407,19 +2632,37 @@ def _desktop_token_readiness_check(install: dict[str, Any]) -> dict[str, Any]: def _code_graph_readiness_check( - capability_plane: dict[str, Any], code_memory: dict[str, Any] + capability_plane: dict[str, Any], + code_memory: dict[str, Any], + *, + coverage: dict[str, Any] | None = None, ) -> dict[str, Any]: capability = _capability_by_key(capability_plane, "code_graph") capability_state = str(capability.get("state") or "") - ready = capability_state == "ready" + engine = (capability.get("detected") or {}).get("engine") + coverage_required = coverage is not None and capability_state != "disabled" + coverage_ready = bool(coverage and coverage.get("ready")) + provider_ready = capability_state == "ready" or ( + engine == "graphify" and capability_state == "needs_index" and coverage_ready + ) + ready = provider_ready and (not coverage_required or coverage_ready) disabled = capability_state == "disabled" + missing = [str(repo) for repo in (coverage or {}).get("missing", []) if str(repo)] + if capability_state == "ready" and coverage_required and not coverage_ready: + detail = ( + f"Code graph is installed, but it does not cover selected repositories: {', '.join(missing)}." + if missing + else "Code graph is installed, but selected repository coverage is not verified." + ) + else: + detail = str(capability.get("detail") or code_memory.get("detail") or "") row = _readiness_check( "code_graph", "Code graph memory", category="memory", tier="optional" if disabled else "recommended", ready=ready, - detail=str(capability.get("detail") or code_memory.get("detail") or ""), + detail=detail, action="" if disabled else str(capability.get("install_hint") or "Run `alfred code-memory doctor`."), @@ -2431,9 +2674,12 @@ def _code_graph_readiness_check( "capability_state": capability_state, "enabled": bool(capability.get("enabled")), } - engine = (capability.get("detected") or {}).get("engine") if engine: detected["engine"] = engine + if coverage_required: + detected["coverage_ready"] = coverage_ready + detected["covered"] = list((coverage or {}).get("covered", [])) + detected["missing"] = missing return row | {"detected": detected} diff --git a/lib/server/streaming.py b/lib/server/streaming.py index 7eaaedb7..7deb5890 100644 --- a/lib/server/streaming.py +++ b/lib/server/streaming.py @@ -29,6 +29,7 @@ import asyncio import json +import logging import os import threading from collections.abc import AsyncIterator, Callable, Iterable @@ -37,6 +38,8 @@ from starlette.concurrency import run_in_threadpool +logger = logging.getLogger(__name__) + def _env_float(name: str, default: float) -> float: """Read a float from env, falling back to ``default`` on absence/garbage.""" @@ -286,22 +289,19 @@ async def tail_transcript_sse( async def stream_converse_turn( *, - run_turn: Callable[[], Any], + run_and_reconcile: Callable[[], dict[str, Any] | None], extract_tokens: Callable[[Path], list[str]], transcript_path: Path, - reconcile: Callable[[Any], dict[str, Any]], poll_seconds: float = CONVERSE_POLL_SECONDS, heartbeat_seconds: float = HEARTBEAT_SECONDS, ) -> AsyncIterator[bytes]: """Stream a Compose converse turn token by token, then reconcile. - ``run_turn`` is the blocking call that runs one interrogator turn. It tees - assistant text to ``transcript_path`` via ``claude_invoke_streaming`` and - runs on a worker thread so the event loop stays free. While it runs, this - helper tails ``transcript_path`` with ``extract_tokens`` and emits each - newly seen assistant text fragment as a ``token`` SSE event. When the turn - returns, it emits a single ``result`` event with ``reconcile(turn)``, or an - ``error`` event when the engine returned nothing usable. + ``run_and_reconcile`` owns the complete blocking mutation: it runs one + interrogator turn, persists the result, and returns the response payload. + It continues on its worker even when the SSE consumer disconnects. While + it runs, this helper tails ``transcript_path`` with ``extract_tokens`` and + emits newly seen assistant text fragments as ``token`` events. """ loop = asyncio.get_event_loop() result_box: dict[str, Any] = {} @@ -309,9 +309,10 @@ async def stream_converse_turn( def _worker() -> None: try: - result_box["turn"] = run_turn() - except Exception as exc: - result_box["error"] = str(exc) or exc.__class__.__name__ + result_box["result"] = run_and_reconcile() + except Exception: + logger.exception("streaming Compose turn failed") + result_box["error"] = True finally: done_event.set() @@ -344,13 +345,13 @@ def _worker() -> None: await loop.run_in_executor(None, worker.join, 1.0) if "error" in result_box: - yield _sse("error", {"detail": result_box["error"]}) + yield _sse("error", {"detail": "live_session_unavailable"}) return - turn = result_box.get("turn") - if turn is None: + result = result_box.get("result") + if not isinstance(result, dict): yield _sse("error", {"detail": "live_session_unavailable"}) return - yield _sse("result", reconcile(turn)) + yield _sse("result", result) def _safe_extract(extract_tokens: Callable[[Path], list[str]], transcript_path: Path) -> list[str]: @@ -394,5 +395,29 @@ def assistant_text_fragments(transcript_path: Path) -> list[str]: continue value = block.get("text") if isinstance(value, str) and value: - fragments.append(value) + fragments.append(_visible_converse_text(value)) return fragments + + +def _visible_converse_text(value: str) -> str: + """Hide the model's structured turn envelope from live chat rendering.""" + + candidate = value.strip() + if candidate.startswith("```") and candidate.endswith("```"): + first_newline = candidate.find("\n") + if first_newline == -1: + return value + candidate = candidate[first_newline + 1 : -3].strip() + decoder = json.JSONDecoder() + try: + obj, end = decoder.raw_decode(candidate) + except json.JSONDecodeError: + return value + if candidate[end:].strip(): + return value + if not isinstance(obj, dict) or not {"reply", "draft", "readiness"}.issubset(obj): + return value + reply = obj.get("reply") + if isinstance(reply, str) and reply.strip(): + return reply.strip() + return value diff --git a/lib/server/views.py b/lib/server/views.py index 657b7415..25001629 100644 --- a/lib/server/views.py +++ b/lib/server/views.py @@ -22,7 +22,9 @@ import os import re import secrets -from contextlib import suppress +import threading +from collections.abc import Iterable, Iterator +from contextlib import contextmanager, suppress from dataclasses import asdict, is_dataclass, replace from datetime import UTC, datetime from pathlib import Path @@ -1075,23 +1077,31 @@ def _run_compose_converse(request: Request, body: dict[str, Any]) -> JSONRespons base_draft = _draft_with_selected_setup_scope(base_draft) repos = _compose_context_repos(body, base_draft=base_draft) + explicit_repo = _explicit_conversation_repo(repos, messages) + grounding_repos = [explicit_repo] if explicit_repo else repos + verified_repo_to_local = _compose_verified_repo_to_local(grounding_repos) repo_grounding = cc.build_repo_grounding( - repos, + grounding_repos, workspace_root=_compose_workspace_root(), - repo_to_local=_compose_repo_to_local(), + repo_to_local=verified_repo_to_local, + allow_workspace_fallback=False, ) # Recall fleet lessons only when this turn looks like real work (gated, not # always-on), and append them to the grounding as advisory context. repo_grounding += _converse_memory_grounding(request, messages=messages, base_draft=base_draft) - code_map = cc.load_code_map(_compose_code_map_path()) + code_map = cc.load_code_map(_compose_code_map_path(), repos=grounding_repos) # Plain mode is per-request: the client toggle wins when present, and the # ALFRED_INTAKE_PROFILE server env is only the default when the body omits # the flag. This lets a non-developer flip jargon-free coaching on/off in # the app without restarting the runtime. intake_guidance = cc.intake_guidance_for(_resolve_intake_profile_name(body)) - operational_grounding = _converse_operational_grounding( - request, - conversation_engine=engine, + operational_grounding = ( + _converse_operational_grounding( + request, + conversation_engine=engine, + ) + if not explicit_repo or _conversation_needs_operational_grounding(messages) + else "" ) loader = runtime_facade.prompt_loader() @@ -1130,7 +1140,12 @@ def _run_compose_converse(request: Request, body: dict[str, Any]) -> JSONRespons intake_guidance=intake_guidance, base_draft=base_draft, engine=engine, - workdir=_planning_workdir(request), + workdir=_compose_read_only_workdir( + request, + repos=repos, + messages=messages, + verified_repo_to_local=verified_repo_to_local, + ), on_condense=_converse_condense_recorder(request, draft_id=draft_id), ) if turn is None: @@ -1204,40 +1219,75 @@ def _stream_compose_converse(request: Request, body: dict[str, Any]) -> Any: ), ) - draft_id = _safe_compose_draft_id(body.get("draft_id")) - prior_payload, prior_path = _read_compose_draft_payload(request, draft_id) - base_draft = _converse_base_draft(body, prior_payload) - base_draft = _draft_with_selected_setup_scope(base_draft) + # Pre-mint the firing id so we can tail its transcript while the model runs. + firing_id = cc.converse_firing_id() + transcript = _converse_transcript_path(request, firing_id) - repos = _compose_context_repos(body, base_draft=base_draft) - repo_grounding = cc.build_repo_grounding( - repos, - workspace_root=_compose_workspace_root(), - repo_to_local=_compose_repo_to_local(), + generator = streaming.stream_converse_turn( + run_and_reconcile=lambda: _run_stream_compose_converse( + request, + body, + messages=messages, + engine=engine, + firing_id=firing_id, + ), + extract_tokens=streaming.assistant_text_fragments, + transcript_path=transcript, ) - # Recall fleet lessons only when this turn looks like real work (gated, not - # always-on), and append them to the grounding as advisory context. - repo_grounding += _converse_memory_grounding(request, messages=messages, base_draft=base_draft) - code_map = cc.load_code_map(_compose_code_map_path()) - # Plain mode is per-request: the client toggle wins when present, and the - # ALFRED_INTAKE_PROFILE server env is only the default when the body omits - # the flag. This lets a non-developer flip jargon-free coaching on/off in - # the app without restarting the runtime. - intake_guidance = cc.intake_guidance_for(_resolve_intake_profile_name(body)) - operational_grounding = _converse_operational_grounding( - request, - conversation_engine=engine, + return StreamingResponse( + generator, + media_type="text/event-stream", + headers=_streaming_cors_headers( + request, + { + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + }, + ), ) - loader = runtime_facade.prompt_loader() - if loader is None: # pragma: no cover - loader is always importable - return JSONResponse( - {"error": "compose interrogator prompt loader unavailable"}, - status_code=503, - headers=cors, - ) - try: +def _run_stream_compose_converse( + request: Request, + body: dict[str, Any], + *, + messages: list[Any], + engine: str, + firing_id: str, +) -> dict[str, Any] | None: + """Run, reconcile, and persist one streaming turn under one worker-owned lock.""" + + import compose_converse as cc + + draft_id = _safe_compose_draft_id(body.get("draft_id")) + with _compose_turn_guard(request, draft_id): + prior_payload, prior_path = _read_compose_draft_payload(request, draft_id) + base_draft = _draft_with_selected_setup_scope(_converse_base_draft(body, prior_payload)) + repos = _compose_context_repos(body, base_draft=base_draft) + explicit_repo = _explicit_conversation_repo(repos, messages) + grounding_repos = [explicit_repo] if explicit_repo else repos + verified_repo_to_local = _compose_verified_repo_to_local(grounding_repos) + repo_grounding = cc.build_repo_grounding( + grounding_repos, + workspace_root=_compose_workspace_root(), + repo_to_local=verified_repo_to_local, + allow_workspace_fallback=False, + ) + repo_grounding += _converse_memory_grounding( + request, + messages=messages, + base_draft=base_draft, + ) + code_map = cc.load_code_map(_compose_code_map_path(), repos=grounding_repos) + intake_guidance = cc.intake_guidance_for(_resolve_intake_profile_name(body)) + operational_grounding = ( + _converse_operational_grounding(request, conversation_engine=engine) + if not explicit_repo or _conversation_needs_operational_grounding(messages) + else "" + ) + loader = runtime_facade.prompt_loader() + if loader is None: # pragma: no cover - loader is always importable + raise RuntimeError("compose interrogator prompt loader unavailable") system_prompt = cc.render_system_prompt( prompt_path=_compose_interrogator_prompt_path(), repo_grounding=repo_grounding, @@ -1246,28 +1296,7 @@ def _stream_compose_converse(request: Request, body: dict[str, Any]) -> Any: loader=loader, operational_grounding=operational_grounding, ) - except OSError: - return JSONResponse( - { - "error": "live_session_unavailable", - "detail": ( - "The spec-interrogator prompt could not be loaded. Check the " - "runtime deploy, or use the one-shot plan form." - ), - }, - status_code=503, - headers=cors, - ) - - # Pre-mint the firing id so we can tail its transcript while the model runs. - firing_id = cc.converse_firing_id() - transcript = _converse_transcript_path(request, firing_id) - workdir = _planning_workdir(request) - - on_condense = _converse_condense_recorder(request, draft_id=draft_id) - - def _run() -> Any: - return cc.run_turn( + turn = cc.run_turn( system_prompt=system_prompt, messages=messages, repo_grounding=repo_grounding, @@ -1275,12 +1304,17 @@ def _run() -> Any: intake_guidance=intake_guidance, base_draft=base_draft, engine=engine, - workdir=workdir, + workdir=_compose_read_only_workdir( + request, + repos=repos, + messages=messages, + verified_repo_to_local=verified_repo_to_local, + ), firing_id=firing_id, - on_condense=on_condense, + on_condense=_converse_condense_recorder(request, draft_id=draft_id), ) - - def _reconcile(turn: Any) -> dict[str, Any]: + if turn is None: + return None content_draft = replace(turn.draft, repos=[]) if turn.draft.repos else turn.draft if ( getattr(turn, "intent", "build") == cc.INTENT_CONVERSATION @@ -1298,24 +1332,6 @@ def _reconcile(turn: Any) -> dict[str, Any]: ) return _converse_turn_payload(turn, draft_id=saved_id, saved_path=saved_path) - generator = streaming.stream_converse_turn( - run_turn=_run, - extract_tokens=streaming.assistant_text_fragments, - transcript_path=transcript, - reconcile=_reconcile, - ) - return StreamingResponse( - generator, - media_type="text/event-stream", - headers=_streaming_cors_headers( - request, - { - "Cache-Control": "no-cache, no-transform", - "X-Accel-Buffering": "no", - }, - ), - ) - def _compose_stream_unavailable(request: Request, *, message: str) -> StreamingResponse: """Emit the stream-route no-engine signal without a failing HTTP status. @@ -1450,9 +1466,7 @@ def _save_converse_draft( root = _state_planning_root(request) root.mkdir(parents=True, exist_ok=True) if draft_path is None: - stamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S") - draft_id = f"{_COMPOSE_PREFIX}{stamp}-{_slug(turn.draft.title)}" - draft_path = root / f"{draft_id}.json" + draft_path, draft_id = _new_compose_draft_path(root, turn.draft.title) elif draft_id is None: draft_id = draft_path.stem created_at = ( @@ -1488,9 +1502,7 @@ def _save_converse_draft( "revision_count": len(conversation), "revisions": [message["content"] for message in conversation], } - tmp = draft_path.with_name(f"{draft_path.name}.tmp") - tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") - tmp.replace(draft_path) + _write_json_atomically(draft_path, payload) return draft_path, draft_id @@ -1537,6 +1549,44 @@ def _save_issue_draft(request: Request, draft: IssueDraft, body: str) -> Path: _COMPOSE_PREFIX = "compose-" +_COMPOSE_STATE_LOCKS = tuple(threading.RLock() for _ in range(64)) + + +def _compose_turn_lock(request: Request, draft_id: str | None) -> Any: + """Return the bounded state lock shared by every planning-draft mutation.""" + + del draft_id + key = str(_state_planning_root(request).resolve()) + return _COMPOSE_STATE_LOCKS[hash(key) % len(_COMPOSE_STATE_LOCKS)] + + +@contextmanager +def _compose_turn_guard(request: Request, draft_id: str | None) -> Iterator[None]: + """Acquire and release one planning-state lock on the calling worker thread.""" + + lock = _compose_turn_lock(request, draft_id) + with lock: + yield + + +def _new_compose_draft_path(root: Path, title: str) -> tuple[Path, str]: + """Mint a collision-resistant id for a new Compose draft.""" + + stamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S-%f") + draft_id = f"{_COMPOSE_PREFIX}{stamp}-{secrets.token_hex(3)}-{_slug(title)}" + return root / f"{draft_id}.json", draft_id + + +def _write_json_atomically(path: Path, payload: dict[str, Any]) -> None: + """Replace one JSON file without sharing a temporary path with another writer.""" + + tmp = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp") + try: + tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + tmp.replace(path) + finally: + with suppress(FileNotFoundError): + tmp.unlink() def _safe_compose_draft_id(raw: Any) -> str | None: @@ -1955,9 +2005,7 @@ def _save_compose_draft( root = _state_planning_root(request) root.mkdir(parents=True, exist_ok=True) if draft_path is None: - stamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S") - draft_id = f"{_COMPOSE_PREFIX}{stamp}-{_slug(draft.title)}" - draft_path = root / f"{draft_id}.json" + draft_path, draft_id = _new_compose_draft_path(root, draft.title) elif draft_id is None: draft_id = draft_path.stem created_at = ( @@ -1979,9 +2027,7 @@ def _save_compose_draft( "revision_count": len(revisions), "revisions": revisions, } - tmp = draft_path.with_name(f"{draft_path.name}.tmp") - tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") - tmp.replace(draft_path) + _write_json_atomically(draft_path, payload) return draft_path, draft_id @@ -2115,9 +2161,7 @@ def _file_planning_draft_issue(state_root: Path, plan_id: str) -> dict[str, Any] "source": "native-client", } payload["updated_at"] = now - tmp = path.with_name(f"{path.name}.tmp") - tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") - tmp.replace(path) + _write_json_atomically(path, payload) return { "ok": True, "status": "filed", @@ -2649,6 +2693,154 @@ def _planning_workdir(request: Request) -> Path: return Path(base) +def _explicit_conversation_repo(repos: list[str], messages: Iterable[Any]) -> str: + """Return one selected repo explicitly named in the latest user turn.""" + + text = "" + for message in reversed(list(messages)): + if str(getattr(message, "role", "") or "") == "user": + text = str(getattr(message, "content", "") or "") + break + if not text or re.search(r"\b(compare|comparison|versus|vs\.?|between)\b", text, re.IGNORECASE): + return "" + matches = [ + repo + for repo in repos + if re.search( + rf"(? bool: + """True when the latest user turn asks about live fleet or run state.""" + + for message in reversed(list(messages)): + if str(getattr(message, "role", "") or "") != "user": + continue + text = str(getattr(message, "content", "") or "") + if _CODE_QUESTION_RE.search(text): + return False + if _EXPLICIT_OPERATIONAL_QUERY_RE.search(text): + return True + return bool(_OPERATIONAL_QUERY_RE.search(text)) + return False + + +def _compose_read_only_workdir( + request: Request, + *, + repos: list[str], + messages: Iterable[Any] = (), + verified_repo_to_local: dict[str, str] | None = None, +) -> Path: + """Use one selected, explicitly mapped Git checkout for read-only Ask turns.""" + + fallback = _planning_workdir(request) + repo = repos[0].strip() if len(repos) == 1 else _explicit_conversation_repo(repos, messages) + if not repo: + return fallback + + verified = verified_repo_to_local + candidate = ( + _mapped_verified_checkout(repo, verified) + if verified is not None + else _compose_verified_checkout(repo) + ) + return candidate if candidate is not None else fallback + + +def _compose_verified_repo_to_local(repos: Iterable[str]) -> dict[str, str]: + """Return only exact selected-repo mappings with verified GitHub identity.""" + + verified: dict[str, str] = {} + for repo in repos: + candidate = _compose_verified_checkout(repo) + if candidate is not None: + verified[repo] = str(candidate) + return verified + + +def _compose_verified_checkout(repo: str) -> Path | None: + """Resolve one exact-slug mapping after checkout and remote verification.""" + + repo = repo.strip() + if "/" not in repo: + return None + selected = {item.casefold() for item in _selected_setup_repos()} + if repo.casefold() not in selected: + return None + + mapped = next( + ( + path + for slug, path in _compose_repo_to_local().items() + if "/" in slug and slug.casefold() == repo.casefold() and path + ), + None, + ) + if not mapped: + return None + + candidate = Path(mapped).expanduser() + if not candidate.is_absolute(): + candidate = _compose_workspace_root() / candidate + try: + candidate = candidate.resolve(strict=True) + if not candidate.is_dir() or not (candidate / ".git").exists(): + return None + except (OSError, RuntimeError): + return None + from server import setup as setup_mod + + if not setup_mod.local_repo_matches_github_slug(candidate, repo): + return None + return candidate + + +def _mapped_verified_checkout(repo: str, verified_repo_to_local: dict[str, str]) -> Path | None: + mapped = next( + ( + path + for slug, path in verified_repo_to_local.items() + if slug.casefold() == repo.casefold() and path + ), + None, + ) + return Path(mapped) if mapped else None + + def _repo_from_github_url(url: str) -> str: match = re.search(r"github\.com/([^/\s]+/[^/\s#?]+)(?:/|$)", url) if not match: diff --git a/prompts/spec-interrogator.md b/prompts/spec-interrogator.md index 7dc36578..b9c3d254 100644 --- a/prompts/spec-interrogator.md +++ b/prompts/spec-interrogator.md @@ -93,6 +93,15 @@ never narrate a filing or handoff that you did not actually perform. These are the repositories in scope and what they contain. Treat this as the source of truth for what already exists. +When a conversation turn asks how code in one selected repository works, use +the read-only `Read`, `Grep`, and `Glob` tools to inspect the current checkout +before answering. Follow the relevant definitions and callers far enough to +answer from code, not from filenames or assumptions. The person's question is +already permission to inspect that selected checkout: do not ask them to point +you at a file or approve a read-only lookup. Never edit files or run mutating +commands in this flow. If the answer still cannot be established after +inspection, say exactly what evidence is missing. + ${REPO_GROUNDING} ## Code map diff --git a/tests/test_agent_runner.py b/tests/test_agent_runner.py index 3330fc4d..afa294db 100644 --- a/tests/test_agent_runner.py +++ b/tests/test_agent_runner.py @@ -1077,6 +1077,37 @@ def fake_run(cmd, *, cwd=None, timeout=None, capture=None, env=None, input_text= assert commands[0][commands[0].index("--model") + 1] == "review-model" +def test_codex_invoke_can_isolate_an_ephemeral_turn(tmp_path, monkeypatch): + import agent_runner as ar + from agent_runner import process as process_mod + + root = tmp_path / "codex" + commands = [] + + def fake_run(cmd, *, cwd=None, timeout=None, capture=None, env=None, input_text=None): + commands.append(cmd) + last_path = Path(cmd[cmd.index("--output-last-message") + 1]) + last_path.parent.mkdir(parents=True, exist_ok=True) + last_path.write_text("Grounded answer") + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + monkeypatch.setattr(ar, "CODEX_TRANSCRIPTS_ROOT", root) + monkeypatch.setattr(process_mod, "_popen_run_text", fake_run) + + out = ar.codex_invoke( + "inspect", + workdir=tmp_path, + agent="compose-interrogator", + firing_id="fire-1", + ignore_user_config=True, + ephemeral=True, + ) + + assert out.success is True + assert "--ignore-user-config" in commands[0] + assert "--ephemeral" in commands[0] + + def test_codex_invoke_can_bypass_approvals_and_sandbox(tmp_path, monkeypatch): import agent_runner as ar from agent_runner import process as process_mod diff --git a/tests/test_code_map.py b/tests/test_code_map.py index 1d99cbc9..6c28a888 100644 --- a/tests/test_code_map.py +++ b/tests/test_code_map.py @@ -185,6 +185,60 @@ def test_load_code_map_summarizes_repo_graph(tmp_path: Path) -> None: assert "Contract drift entries: 1" in summary +def test_load_code_map_filters_unselected_repositories(tmp_path: Path) -> None: + from compose_converse import load_code_map + + path = tmp_path / "code-map.json" + path.write_text( + json.dumps( + { + "repos": { + "acme/frontend": {"routes": [{"path": "/"}]}, + "acme/backend": {"endpoints": [{"method": "GET", "path": "/health"}]}, + }, + "contract_drift": [ + {"caller": "acme/frontend", "path": "/missing"}, + {"caller": "acme/backend", "path": "/other"}, + ], + } + ), + encoding="utf-8", + ) + + summary = load_code_map(path, repos=["acme/frontend"]) + + assert "`acme/frontend`: 1 routes" in summary + assert "acme/backend" not in summary + assert "Contract drift entries: 1" in summary + + +def test_load_code_map_matches_selected_slug_to_production_repo_key(tmp_path: Path) -> None: + from compose_converse import load_code_map + + path = tmp_path / "code-map.json" + path.write_text( + json.dumps( + { + "repos": { + "frontend": {"routes": [{"path": "/"}]}, + "backend": {"endpoints": [{"method": "GET", "path": "/health"}]}, + }, + "contract_drift": [ + {"caller": "frontend", "path": "/missing"}, + {"caller": "backend", "path": "/other"}, + ], + } + ), + encoding="utf-8", + ) + + summary = load_code_map(path, repos=["acme/frontend"]) + + assert "`frontend`: 1 routes" in summary + assert "backend" not in summary + assert "Contract drift entries: 1" in summary + + def test_load_code_map_skips_malformed_graph_counts(tmp_path: Path) -> None: from compose_converse import load_code_map diff --git a/tests/test_compose_converse_condense.py b/tests/test_compose_converse_condense.py index 12e10a03..db4c8c38 100644 --- a/tests/test_compose_converse_condense.py +++ b/tests/test_compose_converse_condense.py @@ -135,6 +135,10 @@ def __call__(self, prompt: str, **kwargs: Any) -> tuple[_Result, str]: "agent": agent, "firing_id": kwargs.get("firing_id"), "provider_failover": kwargs.get("hybrid_fallback_on_provider_failure"), + "claude_read_only_isolation": kwargs.get("claude_read_only_isolation"), + "codex_sandbox": kwargs.get("codex_sandbox"), + "codex_ignore_user_config": kwargs.get("codex_ignore_user_config"), + "codex_ephemeral": kwargs.get("codex_ephemeral"), } ) if agent == cc.CONDENSER_AGENT: @@ -178,9 +182,45 @@ def test_short_conversation_runs_once_without_condensing() -> None: assert spy.condenser_calls == [] # no summarizer call assert len(spy.interrogator_calls) == 1 assert spy.interrogator_calls[0]["provider_failover"] is True + assert spy.interrogator_calls[0]["claude_read_only_isolation"] is True + assert spy.interrogator_calls[0]["codex_sandbox"] == "read-only" + assert spy.interrogator_calls[0]["codex_ignore_user_config"] is True + assert spy.interrogator_calls[0]["codex_ephemeral"] is True assert records == [] +def test_read_only_conversation_preserves_plain_engine_answer() -> None: + plain_answer = "The dispatch gate is in `process.py`. No code was changed." + spy = _EngineSpy(interrogator_results=[_Result(success=True, result_text=plain_answer)]) + messages = [ + cc.ConverseMessage( + role="user", + content="In acme/alfred, identify where dispatch readiness is checked. Do not change code.", + ) + ] + + turn = _run(spy, messages) + + assert turn is not None + assert turn.intent == cc.INTENT_CONVERSATION + assert turn.reply == plain_answer + assert turn.draft == IssueDraft(title="") + assert turn.action is None + + +def test_build_turn_rejects_plain_engine_answer() -> None: + spy = _EngineSpy( + interrogator_results=[_Result(success=True, result_text="I will add the feature.")] + ) + + turn = _run( + spy, + [cc.ConverseMessage(role="user", content="Add a dark mode toggle to settings.")], + ) + + assert turn is None + + def test_long_conversation_condenses_prompt_proactively() -> None: spy = _EngineSpy(interrogator_results=[_Result(success=True, result_text=_VALID_TURN_JSON)]) config = condenser.CondenserConfig(keep_first=1, keep_last=3, trigger_turns=8) @@ -192,6 +232,10 @@ def test_long_conversation_condenses_prompt_proactively() -> None: assert turn is not None # Summarizer fired exactly once. assert len(spy.condenser_calls) == 1 + assert spy.condenser_calls[0]["claude_read_only_isolation"] is True + assert spy.condenser_calls[0]["codex_sandbox"] == "read-only" + assert spy.condenser_calls[0]["codex_ignore_user_config"] is True + assert spy.condenser_calls[0]["codex_ephemeral"] is True # The interrogator prompt carries the injected summary block, not every turn. interrogator_prompt = spy.interrogator_calls[0]["prompt"] assert "COMPACT SUMMARY of older turns" in interrogator_prompt @@ -227,6 +271,9 @@ def test_reactive_condense_and_retry_on_overflow() -> None: assert len(spy.interrogator_calls) == 2 # Exactly one summarizer call (the reactive condensation). assert len(spy.condenser_calls) == 1 + assert spy.condenser_calls[0]["claude_read_only_isolation"] is True + assert spy.condenser_calls[0]["codex_ignore_user_config"] is True + assert spy.condenser_calls[0]["codex_ephemeral"] is True # The retry prompt is the condensed one. retry_prompt = spy.interrogator_calls[1]["prompt"] assert "COMPACT SUMMARY of older turns" in retry_prompt diff --git a/tests/test_graphify_mcp_wiring.py b/tests/test_graphify_mcp_wiring.py index 04f1c506..8c588c37 100644 --- a/tests/test_graphify_mcp_wiring.py +++ b/tests/test_graphify_mcp_wiring.py @@ -9,6 +9,7 @@ from __future__ import annotations import json +import subprocess import sys from pathlib import Path from types import SimpleNamespace @@ -188,7 +189,7 @@ def test_graphify_never_resolves_packages_during_a_firing(monkeypatch, tmp_path: assert server is None -def test_graphify_expands_home_relative_graph_path(monkeypatch, tmp_path: Path) -> None: +def test_graphify_rejects_absolute_graph_path(monkeypatch, tmp_path: Path) -> None: home = tmp_path / "home" graph = home / "graphs" / "repo.json" graph.parent.mkdir(parents=True) @@ -205,8 +206,105 @@ def test_graphify_expands_home_relative_graph_path(monkeypatch, tmp_path: Path) server = _proc._graphify_mcp_server(tmp_path) + assert server is None + + +def test_graphify_rejects_graph_path_that_escapes_checkout(monkeypatch, tmp_path: Path) -> None: + checkout = tmp_path / "checkout" + checkout.mkdir() + graph = tmp_path / "graph.json" + graph.write_text('{"nodes": [], "links": []}', encoding="utf-8") + monkeypatch.setenv("ALFRED_GRAPHIFY_MCP", "1") + monkeypatch.setenv("ALFRED_GRAPHIFY_GRAPH", "../graph.json") + monkeypatch.setattr( + _proc.shutil, + "which", + lambda name: "/usr/local/bin/graphify-mcp" if name == "graphify-mcp" else None, + ) + monkeypatch.setattr(_proc, "_graphify_entrypoint_works", lambda command: True) + + assert _proc._graphify_mcp_server(checkout) is None + + +def test_graphify_uses_verified_checkout_graph_from_firing_worktree( + monkeypatch, tmp_path: Path +) -> None: + source = tmp_path / "source" + firing = tmp_path / "firing" + source.mkdir() + firing.mkdir() + graph = source / "graphify-out" / "graph.json" + graph.parent.mkdir() + graph.write_text('{"nodes": [], "links": []}', encoding="utf-8") + shared_git = tmp_path / "shared.git" + monkeypatch.setenv("ALFRED_GRAPHIFY_MCP", "1") + monkeypatch.setenv("ALFRED_GRAPHIFY_GRAPH", "graphify-out/graph.json") + monkeypatch.setenv("ALFRED_REPO_LOCAL_MAP", f"acme/repo={source}") + monkeypatch.setattr( + _proc, + "_git_common_dir", + lambda path: shared_git if path in {source, firing} else None, + ) + monkeypatch.setattr( + _proc.shutil, + "which", + lambda name: "/usr/local/bin/graphify-mcp" if name == "graphify-mcp" else None, + ) + monkeypatch.setattr(_proc, "_graphify_entrypoint_works", lambda command: True) + + server = _proc._graphify_mcp_server(firing) + assert server is not None - assert server["graphify"]["args"][0] == str(graph) + assert server["graphify"]["args"] == [str(graph), "--transport", "stdio"] + + +def test_git_common_dir_matches_linked_firing_worktree(tmp_path: Path) -> None: + source = tmp_path / "source" + firing = tmp_path / "firing" + source.mkdir() + subprocess.run(["git", "init", "-q", str(source)], check=True) + subprocess.run( + [ + "git", + "-C", + str(source), + "worktree", + "add", + "--orphan", + "-b", + "firing", + str(firing), + ], + check=True, + ) + + assert _proc._git_common_dir(source) == _proc._git_common_dir(firing) + + +def test_graphify_ignores_checkout_map_from_another_repository(monkeypatch, tmp_path: Path) -> None: + source = tmp_path / "source" + firing = tmp_path / "firing" + source.mkdir() + firing.mkdir() + graph = source / "graphify-out" / "graph.json" + graph.parent.mkdir() + graph.write_text('{"nodes": [], "links": []}', encoding="utf-8") + monkeypatch.setenv("ALFRED_GRAPHIFY_MCP", "1") + monkeypatch.setenv("ALFRED_GRAPHIFY_GRAPH", "graphify-out/graph.json") + monkeypatch.setenv("ALFRED_REPO_LOCAL_MAP", f"other/repo={source}") + monkeypatch.setattr( + _proc, + "_git_common_dir", + lambda path: tmp_path / ("source.git" if path == source else "firing.git"), + ) + monkeypatch.setattr( + _proc.shutil, + "which", + lambda name: "/usr/local/bin/graphify-mcp" if name == "graphify-mcp" else None, + ) + monkeypatch.setattr(_proc, "_graphify_entrypoint_works", lambda command: True) + + assert _proc._graphify_mcp_server(firing) is None def test_graphify_tool_names_use_server_prefix() -> None: diff --git a/tests/test_server.py b/tests/test_server.py index 8f3391e8..1ec91c61 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -2,10 +2,14 @@ from __future__ import annotations +import asyncio import json import os import re import sys +import threading +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace from datetime import UTC, datetime from pathlib import Path from types import SimpleNamespace @@ -20,11 +24,16 @@ sys.path.insert(0, str(LIB)) import compose_converse as cc # noqa: E402 +import server.routes.converse as converse_routes # noqa: E402 +import server.routes.plans as plan_routes # noqa: E402 +import server.setup as server_setup # noqa: E402 import server.views as server_views # noqa: E402 +from fastapi.responses import JSONResponse, StreamingResponse # noqa: E402 from fastapi.testclient import TestClient # noqa: E402 from fleet_brain import Lesson # noqa: E402 from server import FilesystemReader, create_app # noqa: E402 from spec_helper import IssueDraft # noqa: E402 +from starlette.requests import Request # noqa: E402 def _write_jsonl(path: Path, records: list[dict]) -> None: @@ -49,6 +58,113 @@ def _auth_headers(state: Path, **extra: str) -> dict[str, str]: return headers +@pytest.mark.parametrize( + ("path", "runner_name", "response"), + [ + ( + "/api/theme-builder/converse", + "_run_theme_builder_converse", + {"reply": "Choose a theme.", "action": None}, + ), + ( + "/api/onboarding/converse", + "_run_onboarding_converse", + {"reply": "Connect GitHub.", "action": None, "done": False}, + ), + ( + "/api/compose/converse", + "_run_compose_converse", + { + "reply": "What should change?", + "draft": {}, + "readiness": {"score": 0, "ready": False, "missing": []}, + "done": False, + }, + ), + ], +) +def test_buffered_conversation_routes_offload_blocking_engine_turns( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + path: str, + runner_name: str, + response: dict[str, object], +) -> None: + state = tmp_path / "state" + state.mkdir() + calls: list[tuple[object, tuple[object, ...]]] = [] + + def runner(*_args: object) -> JSONResponse: + return JSONResponse(response) + + async def offload(function: object, *args: object) -> JSONResponse: + calls.append((function, args)) + return function(*args) # type: ignore[operator] + + monkeypatch.setattr(server_views, runner_name, runner) + monkeypatch.setattr(converse_routes, "run_in_threadpool", offload) + client = TestClient(create_app(FilesystemReader(state_root=state))) + + result = client.post( + path, + json={"messages": [{"role": "user", "content": "Hello"}]}, + headers=_auth_headers(state), + ) + + assert result.status_code == 200 + assert result.json() == response + offloaded_function = ( + converse_routes._run_compose_converse_guarded if path == "/api/compose/converse" else runner + ) + runner_calls = [(function, args) for function, args in calls if function is offloaded_function] + assert len(runner_calls) == 1 + _function, args = runner_calls[0] + assert len(args) == 2 + assert args[1] == {"messages": [{"role": "user", "content": "Hello"}]} + + +def test_compose_conversation_keeps_status_responsive_during_engine_turn( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + state = tmp_path / "state" + state.mkdir() + started = threading.Event() + release = threading.Event() + + def slow_turn(*_args: object) -> JSONResponse: + started.set() + release.wait(timeout=3) + return JSONResponse( + { + "reply": "Done.", + "draft": {}, + "readiness": {"score": 0, "ready": False, "missing": []}, + "done": False, + } + ) + + monkeypatch.setattr(server_views, "_run_compose_converse", slow_turn) + app = create_app(FilesystemReader(state_root=state)) + with TestClient(app) as client, ThreadPoolExecutor(max_workers=2) as executor: + turn = executor.submit( + client.post, + "/api/compose/converse", + json={"messages": [{"role": "user", "content": "Hello"}]}, + headers=_auth_headers(state), + ) + assert started.wait(timeout=1) + status_request = executor.submit(client.get, "/api/status") + try: + status = status_request.result(timeout=2) + finally: + release.set() + response = turn.result(timeout=2) + + assert status.status_code == 200 + assert response.status_code == 200 + + def test_json_api_status_firings_and_plans(tmp_path: Path) -> None: state = tmp_path / "state" plans = tmp_path / "architect-plans" @@ -2327,6 +2443,8 @@ def fake_run_turn(*, base_draft, messages, repo_grounding, code_map, **_kw): capture["messages"] = list(messages) capture["repo_grounding"] = repo_grounding capture["code_map"] = code_map + capture["workdir"] = _kw.get("workdir") + capture["system_prompt"] = _kw.get("system_prompt") draft = base_draft if draft_overrides: from dataclasses import replace @@ -2489,6 +2607,7 @@ def test_compose_converse_uses_checkout_map_saved_after_runtime_import( _use_interrogator_prompt(monkeypatch) checkout = tmp_path / "new checkout" checkout.mkdir() + (checkout / ".git").mkdir() (checkout / "CLAUDE.md").write_text("Live checkout instructions", encoding="utf-8") runtime = tmp_path / "runtime" runtime.mkdir() @@ -2499,6 +2618,12 @@ def test_compose_converse_uses_checkout_map_saved_after_runtime_import( ) monkeypatch.setenv("ALFRED_HOME", str(runtime)) monkeypatch.setenv("ALFRED_REPO_LOCAL_MAP", "stale/frontend=/tmp/stale-frontend") + monkeypatch.setattr(server_views, "_selected_setup_repos", lambda: ["acme/frontend"]) + monkeypatch.setattr( + server_setup, + "local_repo_matches_github_slug", + lambda path, slug: path == checkout.resolve() and slug == "acme/frontend", + ) capture: dict = {} _stub_converse_turn(monkeypatch, reply="Grounded.", capture=capture) @@ -2519,6 +2644,354 @@ def test_compose_converse_uses_checkout_map_saved_after_runtime_import( assert "Live checkout instructions" in capture["repo_grounding"] +@pytest.mark.parametrize("path", ["/api/compose/converse", "/api/compose/converse/stream"]) +@pytest.mark.parametrize( + ("mapping_key", "remote_matches", "sentinel", "expect_grounding"), + [ + ("acme/frontend", True, "Verified checkout instructions", True), + ("acme/frontend", False, "Wrong remote instructions", False), + ("frontend", True, "wrong-repo-only.txt", False), + ], +) +def test_compose_converse_grounding_matches_verified_workdir_for_every_transport( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + path: str, + mapping_key: str, + remote_matches: bool, + sentinel: str, + expect_grounding: bool, +) -> None: + monkeypatch.setenv("ALFRED_COMPOSE_CONVERSE_ENGINE", "claude") + _use_interrogator_prompt(monkeypatch) + checkout = tmp_path / "frontend" + (checkout / ".git").mkdir(parents=True) + if sentinel.endswith(".txt"): + (checkout / sentinel).write_text("belongs to another repo", encoding="utf-8") + else: + (checkout / "CLAUDE.md").write_text(sentinel, encoding="utf-8") + monkeypatch.setattr(server_views, "_compose_workspace_root", lambda: tmp_path) + monkeypatch.setattr(server_views, "_selected_setup_repos", lambda: ["acme/frontend"]) + monkeypatch.setattr( + server_views.runtime_facade, + "repo_to_local", + lambda: {mapping_key: str(checkout)}, + ) + monkeypatch.setattr( + server_setup, + "local_repo_matches_github_slug", + lambda candidate, slug: ( + remote_matches and candidate == checkout.resolve() and slug == "acme/frontend" + ), + ) + capture: dict = {} + _stub_converse_turn(monkeypatch, reply="Grounded.", capture=capture) + + state = tmp_path / "state" + state.mkdir() + client = TestClient(create_app(FilesystemReader(state_root=state))) + response = client.post( + path, + json={ + "repos": ["acme/frontend"], + "messages": [{"role": "user", "content": "How does login work?"}], + }, + headers=_auth_headers(state), + ) + + assert response.status_code == 200 + assert (sentinel in capture["repo_grounding"]) is expect_grounding + assert (capture["workdir"] == checkout.resolve()) is expect_grounding + if not expect_grounding: + assert "No local checkout or CLAUDE.md available" in capture["repo_grounding"] + + +def test_compose_converse_runs_from_single_selected_mapped_checkout( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("ALFRED_COMPOSE_CONVERSE_ENGINE", "claude") + _use_interrogator_prompt(monkeypatch) + checkout = tmp_path / "frontend" + (checkout / ".git").mkdir(parents=True) + monkeypatch.setattr(server_views, "_selected_setup_repos", lambda: ["acme/frontend"]) + monkeypatch.setattr( + server_views.runtime_facade, + "repo_to_local", + lambda: {"acme/frontend": str(checkout)}, + ) + monkeypatch.setattr( + server_setup, + "local_repo_matches_github_slug", + lambda path, slug: path == checkout.resolve() and slug == "acme/frontend", + ) + capture: dict = {} + _stub_converse_turn(monkeypatch, reply="Grounded.", capture=capture) + + state = tmp_path / "state" + state.mkdir() + client = TestClient(create_app(FilesystemReader(state_root=state))) + + response = client.post( + "/api/compose/converse", + json={ + "repos": ["acme/frontend"], + "messages": [{"role": "user", "content": "How does login work?"}], + }, + headers=_auth_headers(state), + ) + + assert response.status_code == 200 + assert capture["workdir"] == checkout.resolve() + + +def test_compose_converse_uses_explicit_repo_from_multi_repo_context( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("ALFRED_COMPOSE_CONVERSE_ENGINE", "claude") + _use_interrogator_prompt(monkeypatch) + frontend = tmp_path / "frontend" + backend = tmp_path / "backend" + (frontend / ".git").mkdir(parents=True) + (backend / ".git").mkdir(parents=True) + (frontend / "CLAUDE.md").write_text("Frontend instructions", encoding="utf-8") + (backend / "CLAUDE.md").write_text("Backend instructions", encoding="utf-8") + selected = ["acme/frontend", "acme/backend"] + monkeypatch.setattr(server_views, "_selected_setup_repos", lambda: selected) + monkeypatch.setattr( + server_views.runtime_facade, + "repo_to_local", + lambda: { + "acme/frontend": str(frontend), + "acme/backend": str(backend), + }, + ) + monkeypatch.setattr( + server_setup, + "local_repo_matches_github_slug", + lambda path, slug: path == frontend.resolve() and slug == "acme/frontend", + ) + capture: dict = {} + _stub_converse_turn(monkeypatch, reply="Grounded.", capture=capture) + + state = tmp_path / "state" + state.mkdir() + client = TestClient(create_app(FilesystemReader(state_root=state))) + + response = client.post( + "/api/compose/converse", + json={ + "context_repos": selected, + "messages": [ + { + "role": "user", + "content": "In acme/frontend, how does login work?", + } + ], + }, + headers=_auth_headers(state), + ) + + assert response.status_code == 200 + assert capture["workdir"] == frontend.resolve() + assert "Frontend instructions" in capture["repo_grounding"] + assert "Backend instructions" not in capture["repo_grounding"] + + +@pytest.mark.parametrize( + ("message", "expects_operational"), + [ + ("Why did acme/frontend run fail?", True), + ("In acme/frontend, identify where login is checked.", False), + ("Where is the acme/frontend status endpoint implemented?", False), + ("Where is queued-job code handled in acme/frontend?", False), + ("How does the acme/frontend workflow runner start?", False), + ], +) +def test_compose_converse_adds_live_state_only_to_repo_operational_questions( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + message: str, + expects_operational: bool, +) -> None: + monkeypatch.setenv("ALFRED_COMPOSE_CONVERSE_ENGINE", "claude") + _use_interrogator_prompt(monkeypatch) + selected = ["acme/frontend", "acme/backend"] + monkeypatch.setattr(server_views, "_selected_setup_repos", lambda: selected) + monkeypatch.setattr( + server_views, + "_converse_operational_grounding", + lambda *_args, **_kwargs: "LIVE RUN DATA", + ) + capture: dict = {} + _stub_converse_turn(monkeypatch, reply="Grounded.", capture=capture) + state = tmp_path / "state" + state.mkdir() + client = TestClient(create_app(FilesystemReader(state_root=state))) + + response = client.post( + "/api/compose/converse", + json={ + "context_repos": selected, + "messages": [{"role": "user", "content": message}], + }, + headers=_auth_headers(state), + ) + + assert response.status_code == 200 + assert ("LIVE RUN DATA" in capture["system_prompt"]) is expects_operational + + +def test_compose_converse_keeps_multi_repo_questions_in_fallback_workdir( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + state = tmp_path / "state" + state.mkdir() + request = SimpleNamespace( + app=SimpleNamespace(state=SimpleNamespace(reader=FilesystemReader(state_root=state))) + ) + selected = ["acme/frontend", "acme/backend"] + monkeypatch.setattr(server_views, "_selected_setup_repos", lambda: selected) + messages = [ + SimpleNamespace( + role="user", + content="Compare acme/frontend with acme/backend.", + ) + ] + + assert server_views._compose_read_only_workdir( + request, + repos=selected, + messages=messages, + ) == server_views._planning_workdir(request) + + +def test_compose_converse_does_not_reuse_stale_repo_scope_for_followup( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + state = tmp_path / "state" + state.mkdir() + request = SimpleNamespace( + app=SimpleNamespace(state=SimpleNamespace(reader=FilesystemReader(state_root=state))) + ) + selected = ["acme/frontend", "acme/backend"] + monkeypatch.setattr(server_views, "_selected_setup_repos", lambda: selected) + messages = [ + SimpleNamespace(role="user", content="In acme/frontend, how does login work?"), + SimpleNamespace(role="assistant", content="It uses the session provider."), + SimpleNamespace(role="user", content="What is the fleet doing now?"), + ] + + assert server_views._explicit_conversation_repo(selected, messages) == "" + assert server_views._compose_read_only_workdir( + request, + repos=selected, + messages=messages, + ) == server_views._planning_workdir(request) + + +def test_compose_converse_keeps_implicit_comparison_multi_repo( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + state = tmp_path / "state" + state.mkdir() + request = SimpleNamespace( + app=SimpleNamespace(state=SimpleNamespace(reader=FilesystemReader(state_root=state))) + ) + selected = ["acme/frontend", "acme/backend"] + monkeypatch.setattr(server_views, "_selected_setup_repos", lambda: selected) + messages = [ + SimpleNamespace(role="user", content="In acme/frontend, how does login work?"), + SimpleNamespace(role="assistant", content="It uses the session provider."), + SimpleNamespace(role="user", content="Compare it with acme/backend."), + ] + + assert server_views._explicit_conversation_repo(selected, messages) == "" + assert server_views._compose_read_only_workdir( + request, + repos=selected, + messages=messages, + ) == server_views._planning_workdir(request) + + +@pytest.mark.parametrize( + ("message", "expected"), + [ + ("Why did acme/frontend run fail?", True), + ("What did acme/frontend ship today?", True), + ("When did acme/frontend last run?", True), + ("Did acme/frontend run yesterday?", True), + ("In acme/frontend, identify where readiness is checked.", False), + ("Where is the status endpoint implemented?", False), + ("Why does the status endpoint fail?", False), + ("When does the nightly workflow run?", False), + ("When does cron cleanup run?", False), + ("Where is queued-job code handled?", False), + ("How does the workflow runner start?", False), + ], +) +def test_converse_operational_grounding_gate_uses_latest_turn( + message: str, + expected: bool, +) -> None: + messages = [ + SimpleNamespace(role="user", content="What is the fleet doing?"), + SimpleNamespace(role="assistant", content="Two agents are live."), + SimpleNamespace(role="user", content=message), + ] + + assert server_views._conversation_needs_operational_grounding(messages) is expected + + +def test_compose_converse_rejects_bare_or_unselected_workdir_mappings( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + checkout = tmp_path / "frontend" + (checkout / ".git").mkdir(parents=True) + state = tmp_path / "state" + state.mkdir() + request = SimpleNamespace( + app=SimpleNamespace(state=SimpleNamespace(reader=FilesystemReader(state_root=state))) + ) + monkeypatch.setattr(server_views, "_selected_setup_repos", lambda: ["acme/frontend"]) + monkeypatch.setattr( + server_views.runtime_facade, + "repo_to_local", + lambda: {"frontend": str(checkout), "other/frontend": str(checkout)}, + ) + + fallback = server_views._planning_workdir(request) + + assert server_views._compose_read_only_workdir(request, repos=["acme/frontend"]) == fallback + assert server_views._compose_read_only_workdir(request, repos=["other/frontend"]) == fallback + + +def test_compose_converse_rejects_mapped_checkout_with_wrong_remote( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + checkout = tmp_path / "frontend" + (checkout / ".git").mkdir(parents=True) + state = tmp_path / "state" + state.mkdir() + request = SimpleNamespace( + app=SimpleNamespace(state=SimpleNamespace(reader=FilesystemReader(state_root=state))) + ) + monkeypatch.setattr(server_views, "_selected_setup_repos", lambda: ["acme/frontend"]) + monkeypatch.setattr( + server_views.runtime_facade, + "repo_to_local", + lambda: {"acme/frontend": str(checkout)}, + ) + monkeypatch.setattr( + server_setup, + "local_repo_matches_github_slug", + lambda _path, _slug: False, + ) + + assert server_views._compose_read_only_workdir( + request, repos=["acme/frontend"] + ) == server_views._planning_workdir(request) + + def test_setup_repo_selection_requires_local_checkouts(tmp_path: Path) -> None: state = tmp_path / "state" state.mkdir() @@ -2628,6 +3101,273 @@ def test_compose_converse_iterates_on_same_draft_id( assert sum(1 for row in drafts if row["draft_id"] == draft_id) == 1 +def test_compose_converse_serializes_retries_for_the_same_draft( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ALFRED_COMPOSE_CONVERSE_ENGINE", "claude") + _use_interrogator_prompt(monkeypatch) + state = tmp_path / "state" + state.mkdir() + + _stub_converse_turn( + monkeypatch, + reply="What should happen on failure?", + draft_overrides={"title": "Reliable export"}, + ) + app = create_app(FilesystemReader(state_root=state)) + with TestClient(app) as client: + first = client.post( + "/api/compose/converse", + json={"messages": [{"role": "user", "content": "Make export reliable"}]}, + headers=_auth_headers(state), + ).json() + draft_id = first["draft_id"] + + entered = [threading.Event(), threading.Event()] + release_first = threading.Event() + call_guard = threading.Lock() + base_drafts: list[IssueDraft] = [] + + def concurrent_turn(*, base_draft: IssueDraft, **_kwargs: object): + with call_guard: + call_index = len(base_drafts) + base_drafts.append(base_draft) + entered[call_index].set() + if call_index == 0: + release_first.wait(timeout=3) + draft = replace(base_draft, problem="Exports can fail silently.") + else: + draft = replace(base_draft, desired_behavior="Failed exports can be retried.") + return cc.ConverseTurn( + reply="Captured.", + draft=draft, + readiness=cc.ConverseReadiness(score=60, ready=False, missing=()), + done=False, + ) + + monkeypatch.setattr(cc, "run_turn", concurrent_turn) + request_body = { + "draft_id": draft_id, + "messages": [{"role": "user", "content": "Capture this retry detail"}], + } + with ThreadPoolExecutor(max_workers=2) as executor: + one = executor.submit( + client.post, + "/api/compose/converse", + json=request_body, + headers=_auth_headers(state), + ) + assert entered[0].wait(timeout=1) + two = executor.submit( + client.post, + "/api/compose/converse", + json=request_body, + headers=_auth_headers(state), + ) + assert not entered[1].wait(timeout=0.2) + release_first.set() + first_retry = one.result(timeout=2) + second_retry = two.result(timeout=2) + + assert first_retry.status_code == 200 + assert second_retry.status_code == 200 + assert entered[1].is_set() + assert base_drafts[1].problem == "Exports can fail silently." + saved = json.loads(Path(second_retry.json()["saved_path"]).read_text(encoding="utf-8")) + assert saved["draft"]["problem"] == "Exports can fail silently." + assert saved["draft"]["desired_behavior"] == "Failed exports can be retried." + + +def test_compose_converse_creates_distinct_drafts_under_concurrency( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ALFRED_COMPOSE_CONVERSE_ENGINE", "claude") + _use_interrogator_prompt(monkeypatch) + state = tmp_path / "state" + state.mkdir() + _stub_converse_turn( + monkeypatch, + reply="Captured.", + draft_overrides={"title": "Concurrent request"}, + ) + app = create_app(FilesystemReader(state_root=state)) + + with TestClient(app) as client, ThreadPoolExecutor(max_workers=8) as executor: + futures = [ + executor.submit( + client.post, + "/api/compose/converse", + json={"messages": [{"role": "user", "content": f"Request {index}"}]}, + headers=_auth_headers(state), + ) + for index in range(8) + ] + responses = [future.result(timeout=4) for future in futures] + + assert all(response.status_code == 200 for response in responses) + draft_ids = {response.json()["draft_id"] for response in responses} + assert len(draft_ids) == len(responses) + assert len(list((state / "planning-drafts").glob("compose-*.json"))) == len(responses) + + +def test_compose_stream_persists_after_client_disconnect( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ALFRED_COMPOSE_CONVERSE_ENGINE", "claude") + _use_interrogator_prompt(monkeypatch) + state = tmp_path / "state" + state.mkdir() + app = create_app(FilesystemReader(state_root=state)) + request = Request( + { + "type": "http", + "app": app, + "method": "POST", + "path": "/api/compose/converse/stream", + "query_string": b"", + "headers": [], + "scheme": "http", + "server": ("testserver", 80), + "client": ("testclient", 50000), + } + ) + started = threading.Event() + release = threading.Event() + + def delayed_turn(*, base_draft: IssueDraft, **_kwargs: object): + started.set() + release.wait(timeout=3) + return cc.ConverseTurn( + reply="Captured.", + draft=replace(base_draft, title="Persist disconnected turn"), + readiness=cc.ConverseReadiness(score=50, ready=False, missing=()), + done=False, + ) + + monkeypatch.setattr(cc, "run_turn", delayed_turn) + response = server_views._stream_compose_converse( + request, + {"messages": [{"role": "user", "content": "Persist this work"}]}, + ) + assert isinstance(response, StreamingResponse) + + async def disconnect_after_open() -> None: + iterator = response.body_iterator + first = await anext(iterator) + assert first.startswith(b"event: open") + assert started.wait(timeout=1) + await iterator.aclose() + + asyncio.run(disconnect_after_open()) + release.set() + deadline = threading.Event() + for _ in range(100): + if list((state / "planning-drafts").glob("compose-*.json")): + break + deadline.wait(0.02) + saved = list((state / "planning-drafts").glob("compose-*.json")) + assert len(saved) == 1 + payload = json.loads(saved[0].read_text(encoding="utf-8")) + assert payload["draft"]["title"] == "Persist disconnected turn" + + +def test_planning_state_lock_covers_deduplicated_sibling_drafts(tmp_path: Path) -> None: + state = tmp_path / "state" + state.mkdir() + app = create_app(FilesystemReader(state_root=state)) + request = SimpleNamespace(app=app) + + first = server_views._compose_turn_lock(request, "compose-first") + sibling = server_views._compose_turn_lock(request, "compose-sibling") + + assert first is sibling + + +def test_compose_converse_serializes_with_one_shot_draft_refinement( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ALFRED_COMPOSE_CONVERSE_ENGINE", "claude") + _use_interrogator_prompt(monkeypatch) + state = tmp_path / "state" + state.mkdir() + app = create_app(FilesystemReader(state_root=state)) + + _stub_converse_turn( + monkeypatch, + reply="What fails today?", + draft_overrides={"title": "Shared planning draft"}, + ) + with TestClient(app) as client: + seeded = client.post( + "/api/compose/converse", + json={"messages": [{"role": "user", "content": "Improve retries"}]}, + headers=_auth_headers(state), + ).json() + draft_id = seeded["draft_id"] + + converse_entered = threading.Event() + release_converse = threading.Event() + + def blocked_turn(*, base_draft: IssueDraft, **_kwargs: object): + converse_entered.set() + release_converse.wait(timeout=3) + return cc.ConverseTurn( + reply="Captured.", + draft=replace(base_draft, problem="Retries can lose queued work."), + readiness=cc.ConverseReadiness(score=50, ready=False, missing=()), + done=False, + ) + + original_refine = plan_routes.refine_issue_draft + plan_entered = threading.Event() + plan_bases: list[IssueDraft] = [] + + def tracked_refine(base_draft: IssueDraft, *args: object, **kwargs: object): + plan_bases.append(base_draft) + plan_entered.set() + return original_refine(base_draft, *args, **kwargs) + + monkeypatch.setattr(cc, "run_turn", blocked_turn) + monkeypatch.setattr(plan_routes, "engine_refiner_from_env", lambda **_kwargs: None) + monkeypatch.setattr(plan_routes, "refine_issue_draft", tracked_refine) + + with ThreadPoolExecutor(max_workers=2) as executor: + converse = executor.submit( + client.post, + "/api/compose/converse", + json={ + "draft_id": draft_id, + "messages": [{"role": "user", "content": "Retries lose queued work"}], + }, + headers=_auth_headers(state), + ) + assert converse_entered.wait(timeout=1) + one_shot = executor.submit( + client.post, + "/api/plans/draft", + json={ + "draft_id": draft_id, + "text": "desired behavior: failed work remains retryable", + }, + headers=_auth_headers(state), + ) + assert not plan_entered.wait(timeout=0.2) + release_converse.set() + converse_response = converse.result(timeout=2) + one_shot_response = one_shot.result(timeout=2) + + assert converse_response.status_code == 200 + assert one_shot_response.status_code == 200 + assert plan_entered.is_set() + assert plan_bases[0].problem == "Retries can lose queued work." + saved = json.loads(Path(one_shot_response.json()["saved_path"]).read_text(encoding="utf-8")) + assert saved["draft"]["problem"] == "Retries can lose queued work." + + def _capture_intake_guidance(monkeypatch: pytest.MonkeyPatch, capture: dict) -> None: """Patch run_turn to record the intake_guidance the endpoint passed in. diff --git a/tests/test_setup_status.py b/tests/test_setup_status.py index c2978544..5aa331e0 100644 --- a/tests/test_setup_status.py +++ b/tests/test_setup_status.py @@ -91,6 +91,298 @@ def test_code_memory_coverage_requires_exact_github_identity(tmp_path: Path) -> assert missing_binary["ready"] is False +def test_code_graph_readiness_requires_selected_repo_coverage() -> None: + capability_plane = { + "capabilities": [ + { + "key": "code_graph", + "state": "ready", + "enabled": True, + "detail": "A graph exists.", + "install_hint": "Run `alfred code-memory index`.", + "detected": {}, + } + ] + } + code_memory = {"graph_dir": "/tmp/code-memory", "detail": "A graph exists."} + + readiness = setup_mod._code_graph_readiness_check( + capability_plane, + code_memory, + coverage={ + "ready": False, + "covered": ["acme/api"], + "missing": ["acme/web"], + }, + ) + + assert readiness["ready"] is False + assert readiness["state"] == "actionable" + assert readiness["detail"] == ( + "Code graph is installed, but it does not cover selected repositories: acme/web." + ) + assert readiness["detected"] == { + "capability_state": "ready", + "enabled": True, + "coverage_ready": False, + "covered": ["acme/api"], + "missing": ["acme/web"], + } + + +def test_graphify_coverage_is_scoped_to_remote_verified_checkouts(tmp_path: Path) -> None: + api = tmp_path / "api" + web = tmp_path / "web" + _git_repo_with_origin(api, "acme/api") + _git_repo_with_origin(web, "acme/web") + graph = api / "graphify-out" / "graph.json" + graph.parent.mkdir() + graph.write_text("{}", encoding="utf-8") + rows = [ + setup_mod._inspect_repo_checkout("acme/api", api, "map"), + setup_mod._inspect_repo_checkout("acme/web", web, "map"), + ] + + coverage = setup_mod._graphify_coverage( + ["acme/api", "acme/web"], + {"ALFRED_GRAPHIFY_GRAPH": str(graph)}, + provider_ready=True, + resolved=rows, + ) + + assert coverage["ready"] is False + assert coverage["covered"] == [] + assert coverage["missing"] == ["acme/api", "acme/web"] + assert coverage["detected"][0]["graph_within_checkout"] is False + assert coverage["detected"][1]["graph_within_checkout"] is False + + +def test_graphify_readiness_requires_selected_repo_coverage() -> None: + capability_plane = { + "capabilities": [ + { + "key": "code_graph", + "state": "ready", + "enabled": True, + "installed": True, + "detail": "Graphify is installed.", + "detected": {"engine": "graphify"}, + "install_hint": "stale repair action", + } + ] + } + + readiness = setup_mod._code_graph_readiness_check( + capability_plane, + {}, + coverage={"ready": False, "covered": ["acme/api"], "missing": ["acme/web"]}, + ) + + assert readiness["ready"] is False + assert readiness["state"] == "actionable" + assert readiness["detected"]["coverage_ready"] is False + assert readiness["detected"]["missing"] == ["acme/web"] + + +@pytest.mark.parametrize( + ("coverage", "expected_state", "expected_ready", "expected_actionable"), + [ + ( + {"ready": True, "covered": ["acme/web"], "missing": []}, + "ready", + 2, + 0, + ), + ( + {"ready": False, "covered": [], "missing": ["acme/web"]}, + "needs_index", + 1, + 1, + ), + ], +) +def test_graphify_capability_plane_reconciles_verified_coverage( + coverage: dict[str, object], + expected_state: str, + expected_ready: int, + expected_actionable: int, +) -> None: + capability_plane = { + "version": 1, + "summary": {"ready": 1, "actionable": 1, "disabled": 0, "total": 2}, + "capabilities": [ + { + "key": "code_graph", + "state": "needs_index", + "enabled": True, + "installed": True, + "detail": "Graphify is installed.", + "detected": {"engine": "graphify"}, + }, + {"key": "engineering_skills", "state": "ready"}, + ], + } + + setup_mod._reconcile_code_graph_coverage(capability_plane, coverage) + + code_graph = setup_mod._capability_by_key(capability_plane, "code_graph") + assert code_graph["state"] == expected_state + assert code_graph["detected"]["coverage_ready"] is coverage["ready"] + assert code_graph["detected"]["covered"] == coverage["covered"] + assert code_graph["detected"]["missing"] == coverage["missing"] + assert code_graph["detected"]["graphify_covered"] == coverage["covered"] + assert code_graph["detected"]["fallback_covered"] == [] + assert capability_plane["summary"] == { + "ready": expected_ready, + "actionable": expected_actionable, + "disabled": 0, + "total": 2, + } + if expected_state == "needs_index": + assert "`graphify `" in code_graph["install_hint"] + else: + assert code_graph["install_hint"] == "" + + +def test_graphify_reconciliation_reports_code_memory_fallback_as_provider( + tmp_path: Path, +) -> None: + checkout = tmp_path / "web" + _git_repo_with_origin(checkout, "acme/web") + resolved = [setup_mod._inspect_repo_checkout("acme/web", checkout, "map")] + capability_plane = { + "version": 1, + "summary": {"ready": 1, "actionable": 1, "disabled": 0, "total": 2}, + "capabilities": [ + { + "key": "code_graph", + "state": "needs_index", + "enabled": True, + "installed": True, + "detail": "Graphify is installed.", + "detected": {"engine": "graphify", "fallback": "code-memory"}, + }, + {"key": "engineering_skills", "state": "ready"}, + ], + } + code_graph = setup_mod._capability_by_key(capability_plane, "code_graph") + coverage = setup_mod._selected_code_graph_coverage( + ["acme/web"], + {"ALFRED_GRAPHIFY_GRAPH": "graphify-out/graph.json"}, + code_memory={ + "enabled": False, + "binary": {"resolved": True}, + "index_present": True, + "repos": {"selected": ["acme/web"]}, + }, + code_graph=code_graph, + resolved=resolved, + ) + + setup_mod._reconcile_code_graph_coverage(capability_plane, coverage) + + assert code_graph["state"] == "ready" + assert code_graph["detected"]["graphify_covered"] == [] + assert code_graph["detected"]["fallback_covered"] == ["acme/web"] + assert code_graph["detail"] == ( + "Selected repositories have verified code-graph coverage through the " + "code-memory fallback; Graphify graphs are not ready." + ) + assert "Graphify covers all" not in code_graph["detail"] + + +def test_bootstrap_status_absolute_graph_downgrade_uses_graphify_repair( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + _stub_common(monkeypatch) + _isolate_launcher_env(monkeypatch, tmp_path) + workspace = tmp_path / "workspace" + checkout = workspace / "web" + _git_repo_with_origin(checkout, "octocat/web") + graph = checkout / "graphify-out" / "graph.json" + graph.parent.mkdir() + graph.write_text("{}", encoding="utf-8") + monkeypatch.setenv("WORKSPACE_ROOT", str(workspace)) + monkeypatch.setenv("WORKSPACE_SUBDIR", "") + monkeypatch.setenv("ALFRED_QUEUE_REPOS", "octocat/web") + monkeypatch.setenv("ALFRED_SHIPPED_REPOS", "octocat/web") + monkeypatch.setenv("ALFRED_GRAPHIFY_GRAPH", str(graph)) + monkeypatch.setattr( + setup_mod.batteries, + "manifest", + lambda _env: { + "batteries": [ + { + "id": "graphify", + "configured": True, + "enabled": True, + "installed": True, + "status": "enabled", + } + ] + }, + ) + + payload = setup_mod.bootstrap_status() + + code_graph = setup_mod._capability_by_key(payload["capability_plane"], "code_graph") + first_run = {check["key"]: check for check in payload["first_run"]["checks"]}["code_graph"] + assert code_graph["state"] == "needs_index" + assert code_graph["detected"]["coverage_ready"] is False + assert code_graph["detected"]["missing"] == ["octocat/web"] + assert "`graphify `" in code_graph["install_hint"] + assert first_run["ready"] is False + assert first_run["action"] == code_graph["install_hint"] + assert "code-memory doctor" not in first_run["action"] + + +def test_bootstrap_status_reconciles_verified_graphify_coverage( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + _stub_common(monkeypatch) + _isolate_launcher_env(monkeypatch, tmp_path) + workspace = tmp_path / "workspace" + checkout = workspace / "web" + _git_repo_with_origin(checkout, "octocat/web") + graph = checkout / "graphify-out" / "graph.json" + graph.parent.mkdir() + graph.write_text("{}", encoding="utf-8") + monkeypatch.setenv("WORKSPACE_ROOT", str(workspace)) + monkeypatch.setenv("WORKSPACE_SUBDIR", "") + monkeypatch.setenv("ALFRED_QUEUE_REPOS", "octocat/web") + monkeypatch.setenv("ALFRED_SHIPPED_REPOS", "octocat/web") + monkeypatch.setenv("ALFRED_GRAPHIFY_GRAPH", "graphify-out/graph.json") + monkeypatch.setattr( + setup_mod.batteries, + "manifest", + lambda _env: { + "batteries": [ + { + "id": "graphify", + "configured": True, + "enabled": True, + "installed": True, + "status": "enabled", + } + ] + }, + ) + + payload = setup_mod.bootstrap_status() + + code_graph = setup_mod._capability_by_key(payload["capability_plane"], "code_graph") + first_run = {check["key"]: check for check in payload["first_run"]["checks"]}["code_graph"] + states = [item["state"] for item in payload["capability_plane"]["capabilities"]] + assert code_graph["state"] == "ready" + assert code_graph["detected"]["coverage_ready"] is True + assert first_run["ready"] is True + assert first_run["detected"]["capability_state"] == "ready" + assert payload["capability_plane"]["summary"]["ready"] == states.count("ready") + assert payload["capability_plane"]["summary"]["actionable"] == sum( + state in {"installable", "missing", "needs_index", "available"} for state in states + ) + + def test_bootstrap_status_reports_code_memory_defaults( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: @@ -1118,6 +1410,51 @@ def test_ready_code_memory_wins_while_graphify_is_not_usable( ) +def test_missing_graphify_uses_ready_fallback_for_selected_repo_coverage( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + checkout = tmp_path / "web" + _git_repo_with_origin(checkout, "octocat/web") + resolved = [setup_mod._inspect_repo_checkout("octocat/web", checkout, "map")] + monkeypatch.setattr( + setup_mod.batteries, + "manifest", + lambda _env: { + "batteries": [ + { + "id": "graphify", + "configured": True, + "enabled": False, + "installed": False, + } + ] + }, + ) + code_memory = { + "enabled": False, + "binary": {"resolved": True}, + "index_present": True, + "repos": {"selected": ["octocat/web"]}, + "detail": "Fallback index is ready.", + } + env = {"ALFRED_GRAPHIFY_FALLBACK": "code-memory"} + plane = setup_mod.capability_status(code_memory, launcher_env=env) + code_graph = next(item for item in plane["capabilities"] if item["key"] == "code_graph") + + coverage = setup_mod._selected_code_graph_coverage( + ["octocat/web"], + env, + code_memory=code_memory, + code_graph=code_graph, + resolved=resolved, + ) + + assert code_graph["source"]["source"] == "DeusData/codebase-memory-mcp" + assert code_graph["enabled"] is True + assert coverage["ready"] is True + assert coverage["covered"] == ["octocat/web"] + + def test_relative_graph_is_not_probed_against_setup_server_cwd( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: @@ -1143,6 +1480,106 @@ def test_relative_graph_is_not_probed_against_setup_server_cwd( assert code_graph["detected"]["graph_present"] is False +def test_relative_graph_keeps_graphify_selected_when_fallback_is_ready( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + setup_mod.batteries, + "manifest", + lambda _env: { + "batteries": [ + { + "id": "graphify", + "configured": True, + "enabled": True, + "installed": True, + } + ] + }, + ) + code_memory = { + "enabled": False, + "binary": {"resolved": True}, + "index_present": True, + "detail": "Fallback index is ready.", + } + + payload = setup_mod.capability_status( + code_memory, + launcher_env={ + "ALFRED_GRAPHIFY_GRAPH": "graphify-out/graph.json", + "ALFRED_GRAPHIFY_FALLBACK": "code-memory", + }, + ) + code_graph = next(item for item in payload["capabilities"] if item["key"] == "code_graph") + + assert code_graph["state"] == "needs_index" + assert code_graph["detected"]["engine"] == "graphify" + assert code_graph["detected"]["fallback"] == "code-memory" + + +def test_selected_graphify_coverage_wins_over_missing_fallback_index(tmp_path: Path) -> None: + checkout = tmp_path / "web" + _git_repo_with_origin(checkout, "octocat/web") + graph = checkout / "graphify-out" / "graph.json" + graph.parent.mkdir() + graph.write_text("{}", encoding="utf-8") + resolved = [setup_mod._inspect_repo_checkout("octocat/web", checkout, "map")] + code_graph = { + "enabled": True, + "installed": True, + "detected": {"engine": "graphify", "fallback": "code-memory"}, + } + code_memory = { + "enabled": False, + "binary": {"resolved": True}, + "index_present": True, + "repos": {"selected": []}, + } + + coverage = setup_mod._selected_code_graph_coverage( + ["octocat/web"], + {"ALFRED_GRAPHIFY_GRAPH": "graphify-out/graph.json"}, + code_memory=code_memory, + code_graph=code_graph, + resolved=resolved, + ) + + assert coverage["ready"] is True + assert coverage["covered"] == ["octocat/web"] + assert coverage["missing"] == [] + assert coverage["detected"][0]["provider"] == "graphify" + + +def test_selected_graphify_coverage_uses_explicit_fallback_per_repo(tmp_path: Path) -> None: + checkout = tmp_path / "web" + _git_repo_with_origin(checkout, "octocat/web") + resolved = [setup_mod._inspect_repo_checkout("octocat/web", checkout, "map")] + code_graph = { + "enabled": True, + "installed": True, + "detected": {"engine": "graphify", "fallback": "code-memory"}, + } + code_memory = { + "enabled": False, + "binary": {"resolved": True}, + "index_present": True, + "repos": {"selected": ["octocat/web"]}, + } + + coverage = setup_mod._selected_code_graph_coverage( + ["octocat/web"], + {"ALFRED_GRAPHIFY_GRAPH": "graphify-out/graph.json"}, + code_memory=code_memory, + code_graph=code_graph, + resolved=resolved, + ) + + assert coverage["ready"] is True + assert coverage["covered"] == ["octocat/web"] + assert coverage["detected"][0]["provider"] == "code-memory" + + def test_capability_plane_reports_builtin_context_governor_with_headroom_detected( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: diff --git a/tests/test_spec_interrogator_prompt.py b/tests/test_spec_interrogator_prompt.py index c050d447..c5ec33be 100644 --- a/tests/test_spec_interrogator_prompt.py +++ b/tests/test_spec_interrogator_prompt.py @@ -39,6 +39,14 @@ def test_prompt_keeps_status_answers_tight() -> None: assert "not a per-agent roll call" in text +def test_prompt_requires_read_only_code_inspection_for_repo_questions() -> None: + text = " ".join(_text().split()) + assert "read-only `Read`, `Grep`, and `Glob` tools" in text + assert "answer from code, not from filenames or assumptions" in text + assert "do not ask them to point you at a file or approve a read-only lookup" in text + assert "Never edit files or run mutating commands in this flow" in text + + def test_prompt_keeps_voice_rules() -> None: # The brevity edits must not drop the existing voice rules. text = _text() diff --git a/tests/test_streaming_endpoints.py b/tests/test_streaming_endpoints.py index a43e9bb5..3b7d4a52 100644 --- a/tests/test_streaming_endpoints.py +++ b/tests/test_streaming_endpoints.py @@ -139,6 +139,49 @@ def test_assistant_text_fragments_in_order(tmp_path: Path) -> None: assert streaming.assistant_text_fragments(transcript) == ["Reading ", "the code."] +def test_assistant_text_fragments_hides_structured_turn_envelope(tmp_path: Path) -> None: + transcript = tmp_path / "turn.jsonl" + structured = { + "intent": "conversation", + "reply": "The readiness gate is in process.py.", + "draft": {}, + "readiness": {"score": 0, "ready": False}, + } + transcript.write_text( + _assistant_line("```json\n" + json.dumps(structured) + "\n```") + "\n", + encoding="utf-8", + ) + + assert streaming.assistant_text_fragments(transcript) == [ + "The readiness gate is in process.py." + ] + + +def test_assistant_text_fragments_preserves_plain_answer_with_code_braces( + tmp_path: Path, +) -> None: + transcript = tmp_path / "plain.jsonl" + answer = 'The parser accepts objects such as {"status": "ready"}.' + transcript.write_text(_assistant_line(answer) + "\n", encoding="utf-8") + + assert streaming.assistant_text_fragments(transcript) == [answer] + + +def test_assistant_text_fragments_preserves_embedded_envelope_example( + tmp_path: Path, +) -> None: + transcript = tmp_path / "example.jsonl" + envelope = { + "reply": "Only this sample reply", + "draft": {}, + "readiness": {"score": 0, "ready": False}, + } + answer = "The API returns this example:\n```json\n" + json.dumps(envelope) + "\n```\nUse it." + transcript.write_text(_assistant_line(answer) + "\n", encoding="utf-8") + + assert streaming.assistant_text_fragments(transcript) == [answer] + + def test_tail_offset_poll_fallback_returns_json_snapshot(tmp_path: Path) -> None: state = tmp_path / "state" transcript = _transcript_for(state, "lucius", "poll-1") @@ -222,10 +265,9 @@ def run_turn() -> dict[str, str]: async def _collect() -> str: frames = [] async for frame in streaming.stream_converse_turn( - run_turn=run_turn, + run_and_reconcile=run_turn, extract_tokens=streaming.assistant_text_fragments, transcript_path=transcript, - reconcile=lambda turn: {"reply": turn["reply"]}, poll_seconds=0.02, ): frames.append(frame.decode("utf-8")) @@ -239,6 +281,30 @@ async def _collect() -> str: assert "".join(tokens) == "hello world" +def test_stream_converse_turn_does_not_expose_worker_exception_text(tmp_path: Path) -> None: + transcript = tmp_path / "t.jsonl" + transcript.write_text("", encoding="utf-8") + private_sentinel = "/private/operator/workspace/secret.py" + + def fail() -> dict[str, str]: + raise RuntimeError(private_sentinel) + + async def _collect() -> str: + frames = [] + async for frame in streaming.stream_converse_turn( + run_and_reconcile=fail, + extract_tokens=streaming.assistant_text_fragments, + transcript_path=transcript, + poll_seconds=0.01, + ): + frames.append(frame.decode("utf-8")) + return "".join(frames) + + body = asyncio.run(_collect()) + assert private_sentinel not in body + assert "live_session_unavailable" in body + + def test_stream_converse_turn_emits_heartbeat_while_idle(tmp_path: Path) -> None: transcript = tmp_path / "t.jsonl" transcript.write_text("", encoding="utf-8") @@ -252,10 +318,9 @@ def run_turn() -> dict[str, str]: async def _collect() -> str: frames = [] async for frame in streaming.stream_converse_turn( - run_turn=run_turn, + run_and_reconcile=run_turn, extract_tokens=streaming.assistant_text_fragments, transcript_path=transcript, - reconcile=lambda turn: {"reply": turn["reply"]}, poll_seconds=0.02, heartbeat_seconds=0.05, ): @@ -284,10 +349,9 @@ def run_turn() -> dict[str, str]: async def _collect() -> str: frames = [] async for frame in streaming.stream_converse_turn( - run_turn=run_turn, + run_and_reconcile=run_turn, extract_tokens=streaming.assistant_text_fragments, transcript_path=transcript, - reconcile=lambda turn: {"reply": turn["reply"]}, poll_seconds=0.02, heartbeat_seconds=0, ): diff --git a/tests/unit/agent_runner/test_process.py b/tests/unit/agent_runner/test_process.py index 0c2c17e5..9196c651 100644 --- a/tests/unit/agent_runner/test_process.py +++ b/tests/unit/agent_runner/test_process.py @@ -119,6 +119,7 @@ def test_claude_invoke_streaming_writes_transcript(fresh_agent_runner, monkeypat import agent_runner.process as proc monkeypatch.setenv("ALFRED_CLAUDE_PROXY_SOCKET", "/tmp/socket-that-should-be-ignored") + monkeypatch.setenv("ALFRED_AGENT_HOOKS", "1") captured: dict[str, object] = {} assistant = { @@ -176,6 +177,7 @@ def fake_popen(cmd, **kwargs): assert cmd[cmd.index("--output-format") + 1] == "stream-json" assert "--allowedTools" in cmd assert cmd[cmd.index("--allowedTools") + 1].startswith("Read,Bash") + assert "--settings" in cmd assert captured["timeout"] == 42 assert captured["kwargs"]["cwd"] == "/tmp" assert captured["kwargs"]["env"]["CLAUDE_CONFIG_DIR"] @@ -183,6 +185,70 @@ def fake_popen(cmd, **kwargs): assert transcript.read_text(encoding="utf-8") == stream +def test_claude_invoke_streaming_can_enforce_read_only_isolation(fresh_agent_runner, monkeypatch): + """Compose exposes only read tools and ignores repository customizations.""" + ar = fresh_agent_runner + import agent_runner.process as proc + + monkeypatch.setenv("ALFRED_AGENT_HOOKS", "1") + monkeypatch.delenv("ALFRED_AGENT_NOTIFICATIONS", raising=False) + captured: dict[str, object] = {} + final = { + "type": "result", + "subtype": "success", + "result": "ok", + "num_turns": 1, + "total_cost_usd": 0.0, + "session_id": "isolated", + "stop_reason": "end_turn", + } + + class FakeProc: + returncode = 0 + stdout = io.StringIO(json.dumps(final) + "\n") + stderr = io.StringIO("") + + def wait(self, timeout: int) -> int: + return self.returncode + + def kill(self) -> None: + self.returncode = -9 + + def fake_popen(cmd, **kwargs): + captured["cmd"] = cmd + return FakeProc() + + monkeypatch.setattr(proc.subprocess, "Popen", fake_popen) + monkeypatch.setattr( + proc, + "_memory_mcp_script", + lambda: (_ for _ in ()).throw(AssertionError("memory MCP must stay disabled")), + ) + + result = ar.claude_invoke_streaming( + prompt="inspect", + workdir=Path("/tmp"), + allowed_tools="Read,Grep,Glob", + agent="compose-interrogator", + firing_id="isolated-read", + read_only_isolation=True, + ) + + assert result.success is True + cmd = captured["cmd"] + assert cmd[cmd.index("--tools") + 1] == "Read,Grep,Glob" + assert cmd[cmd.index("--allowedTools") + 1] == "Read,Grep,Glob" + assert cmd[cmd.index("--permission-mode") + 1] == "dontAsk" + assert "--safe-mode" in cmd + assert "--strict-mcp-config" in cmd + assert cmd[cmd.index("--mcp-config") + 1] == '{"mcpServers":{}}' + assert "--no-session-persistence" in cmd + assert "--settings" not in cmd + assert "alfred_hooks.py" not in " ".join(cmd) + assert "--dangerously-skip-permissions" not in cmd + assert "mcp__alfred_memory" not in " ".join(cmd) + + def test_claude_invoke_streaming_surfaces_popen_oserror(fresh_agent_runner, monkeypatch): ar = fresh_agent_runner from pathlib import Path