diff --git a/anton/core/llm/prompts.py b/anton/core/llm/prompts.py index 989d3d80..318eb4ef 100644 --- a/anton/core/llm/prompts.py +++ b/anton/core/llm/prompts.py @@ -259,20 +259,30 @@ - Throwaway files inside the scratchpad's own working directory. WORKFLOW: -1. NEW artifact: call `create_artifact(name, description, type, primary?)` \ -→ use the returned `` for every subsequent write. -2. EDITING an existing artifact: call `list_artifacts` to find it, then \ -`open_artifact(slug)` to get the folder path. Do NOT call `create_artifact` \ -again — that creates a duplicate. -3. If you discover the entry-point filename only later (or change it), call \ -`update_artifact(slug, primary=...)` so the renderer opens the right file. -4. AFTER FINISHING — reference the artifact in your final message. Once the \ -artifact's files are written, tell the user what was created and point to it by \ -`name` and `slug`, and include the primary file's path \ -(`/`) so it is clickable/openable in a plain CLI. NEVER \ -end with only a description of the content and no pointer to the result. (For \ -fullstack apps, prefer the `url` returned by `launch_backend` as the primary \ -pointer — see the BACKEND & FULLSTACK section.) +1. REGISTER FIRST, always: call `create_artifact(name, description, type, \ +primary?)`. It claims the folder and returns ``. +2. THEN, for `html-app`, `fullstack-stateless-app` and \ +`fullstack-stateful-app` — call `generate_artifact(slug, context)` and let it \ +produce every file. It runs a verified pipeline (data check → tech spec → API \ +spec → code generation with static verification → launch and health check) and \ +for fullstack apps it launches the backend itself, so you do NOT call \ +`launch_backend` afterwards. Do NOT write these files yourself in the \ +scratchpad — the pipeline's checks are what keep the result openable and \ +deployable. +3. For the other types (`document`, `dataset`, `image`, `mixed`) there is no \ +generator: write the files yourself into ``. +4. EDITING an existing artifact: call `list_artifacts` to find it, then \ +`open_artifact(slug)` to get the folder path, then edit its files yourself. Do \ +NOT call `create_artifact` again — that creates a duplicate. `generate_artifact` \ +builds from scratch, so it is not the tool for a small edit. +5. If you discover the entry-point filename only later (or change it), call \ +`update_artifact(slug, primary=...)` so the renderer opens the right file. \ +(`generate_artifact` does this for you.) +6. AFTER FINISHING — reference the artifact in your final message. Tell the user \ +what was created and point to it by `name` and `slug`, and include the primary \ +file's path (`/`) so it is clickable/openable in a plain \ +CLI. NEVER end with only a description of the content and no pointer to the \ +result. For fullstack apps, prefer the `url` the launch step returned. """ @@ -310,14 +320,18 @@ VISUALIZATIONS_HTML_OUTPUT_FORMAT_PROMPT = """\ Present analysis results as HTML dashboards/reports — the user has proactive \ dashboards enabled. Narrate the key insights in chat first (per the workflow \ -above), then build the visualization as a self-contained HTML artifact. - -MANDATORY: BEFORE writing any dashboard, chart, or report HTML, call \ -`recall_skill("build-html-dashboard")` and follow the loaded output contract \ -(artifact registration, file layout, charting library, theme, data embedding). \ -Do NOT build dashboard HTML from memory of those rules — recall the skill in \ -every conversation that produces one. Recalling it too often is fine; \ -skipping it is not.\ +above), then produce the visualization as an artifact. + +Normal path: `create_artifact(type="html-app", …)` then \ +`generate_artifact(slug, context)`. The generator writes the dashboard through a \ +verified pipeline and its own output contract — you do NOT recall a skill or \ +write the HTML yourself for this. + +Building it BY HAND is the exception — editing an existing dashboard, or \ +`generate_artifact` returned an error and the user asked you to continue \ +manually. Only then call `recall_skill("build-html-dashboard")` first and follow \ +the loaded output contract (charting library, theme, file layout, large-dataset \ +handling). Never write dashboard HTML by hand from memory of those rules.\ """ @@ -331,33 +345,35 @@ - Use markdown tables for tabular data. Keep columns aligned and readable. - Use bold/headers for section structure. Use bullet points for lists. - For large datasets, summarize the top N and offer to show more. -- When the user EXPLICITLY asks for a chart, dashboard, plot, or HTML visualization, \ -THEN build it as a self-contained HTML file with inlined CSS, JS, and data. \ -Register the artifact FIRST via `create_artifact(type="html-app", \ -primary="dashboard.html", ...)` and write into the returned `` — \ -see the ARTIFACTS section above for the full contract. \ +- When the user EXPLICITLY asks for a chart, dashboard, plot, or HTML \ +visualization, THEN produce it as an artifact: `create_artifact(type="html-app", \ +primary="dashboard.html", ...)` followed by `generate_artifact(slug, context)` — \ +see the ARTIFACTS section above. If you end up building it BY HAND instead \ +(editing an existing dashboard, or the generator failed and the user asked you \ +to continue), call `recall_skill("build-html-dashboard")` first and follow the \ +loaded output contract. \ Fallback only if `create_artifact` is unavailable: save to `{output_dir}` \ -(create it if needed). \ -MANDATORY: call `recall_skill("build-html-dashboard")` BEFORE writing the HTML \ -and follow the loaded output contract (charting library, theme, file layout, \ -large-dataset handling). Recalling it too often is fine; skipping it is not.\ +(create it if needed).\ """ BACKEND_GENERATION_PROMPT = """\ BACKEND & FULLSTACK APPLICATION GENERATION: -Building a backend service, API, or fullstack web app (artifact types \ -`fullstack-stateless-app` / `fullstack-stateful-app`, launched via \ -`launch_backend`) follows a STRICT contract: a canonical FastAPI+Mangum \ -backend.py template, SECRETS handling, the `/api/*` route prefix, the \ -`static/` frontend layout, requirements.txt, and a launch/preview workflow. \ -The full procedure is NOT in this prompt — it lives in the \ -`build-fullstack-backend` skill. MANDATORY: call \ -`recall_skill("build-fullstack-backend")` BEFORE registering a fullstack \ -artifact or writing any backend code. Code written without it WILL fail \ -launch and deployment. If there is any chance the task involves a backend, \ -recall the skill first — recalling it too often is fine; skipping it is not.\ +Normal path: register the artifact (`fullstack-stateless-app` — prefer this — or \ +`fullstack-stateful-app`), then call `generate_artifact(slug, context)`. The \ +generator writes `backend.py`, `requirements.txt` and `static/index.html` \ +against a hard contract, verifies the backend by importing it and checking its \ +routes, then launches it and health-checks `/api/health`. You do NOT recall a \ +skill, write backend code, or call `launch_backend` yourself on this path. + +Backend code written BY HAND must follow that same strict contract — the \ +canonical FastAPI+Mangum `backend.py` template, `SECRETS` read at point of use, \ +the `/api/*` route prefix, the `static/` frontend layout, `requirements.txt`, \ +and the launch/preview workflow — or it WILL fail launch and deployment. The \ +full procedure is not in this prompt: if you are editing an existing fullstack \ +artifact, or the generator failed and the user asked you to continue manually, \ +call `recall_skill("build-fullstack-backend")` BEFORE touching any backend code.\ """ CONSOLIDATION_PROMPT = """\ diff --git a/anton/core/llm/serialize.py b/anton/core/llm/serialize.py new file mode 100644 index 00000000..45be59f9 --- /dev/null +++ b/anton/core/llm/serialize.py @@ -0,0 +1,68 @@ +"""Shared serialization of LLM messages/responses to plain JSON-able dicts. + +Used by the optional debug loggers (`anton._llm_history_hook` and the +`generate_artifact` step trace). Kept provider-agnostic and dependency-free +so any logger can import it without pulling in the provider classes. +""" + +from __future__ import annotations + +from typing import Any + + +def serialize_content(content: Any) -> Any: + """Normalize a message `content` field to JSON-able form. + + Non-list content (plain strings) passes through untouched. List content + is mapped block-by-block; unknown block shapes pass through unchanged so + nothing is silently dropped. + """ + if not isinstance(content, list): + return content + blocks: list[Any] = [] + for block in content: + if not isinstance(block, dict): + blocks.append(block) + continue + t = block.get("type") + if t == "text": + blocks.append({"type": "text", "text": block.get("text", "")}) + elif t == "tool_use": + blocks.append({ + "type": "tool_use", + "id": block.get("id"), + "name": block.get("name"), + "input": block.get("input"), + }) + elif t == "tool_result": + blocks.append({ + "type": "tool_result", + "tool_use_id": block.get("tool_use_id"), + "content": block.get("content"), + }) + else: + blocks.append(block) + return blocks + + +def serialize_messages(messages: list[dict]) -> list[dict]: + """Normalize a list of chat messages to `[{role, content}]`.""" + return [ + {"role": m.get("role", "?"), "content": serialize_content(m.get("content", ""))} + for m in messages + ] + + +def serialize_response(response) -> dict: + """Normalize an `LLMResponse` to a JSON-able dict.""" + return { + "content": response.content or "", + "tool_calls": [ + {"name": tc.name, "input": tc.input} for tc in response.tool_calls + ], + "usage": { + "input_tokens": response.usage.input_tokens, + "output_tokens": response.usage.output_tokens, + }, + "stop_reason": response.stop_reason, + } diff --git a/anton/core/memory/builtin_skills/build-fullstack-backend/SKILL.md b/anton/core/memory/builtin_skills/build-fullstack-backend/SKILL.md index d7ba130f..36c52fd7 100644 --- a/anton/core/memory/builtin_skills/build-fullstack-backend/SKILL.md +++ b/anton/core/memory/builtin_skills/build-fullstack-backend/SKILL.md @@ -1,12 +1,13 @@ --- name: build-fullstack-backend -description: 'MANDATORY reading before writing ANY backend, API, server, or fullstack - application code (create_artifact types fullstack-stateless-app / fullstack-stateful-app, - or anything that will be launched with launch_backend). Contains the complete hard - contract: the canonical FastAPI+Mangum backend.py template, SECRETS handling, /api/* - route prefix rules, static/ frontend layout, requirements.txt, and the launch/preview - workflow. Building a backend without recalling this skill first WILL break launch - and deployment. When in doubt, recall it.' +description: 'ONLY for writing backend or fullstack code BY HAND. NOT needed on the + normal path — create_artifact(fullstack-stateless-app or fullstack-stateful-app) + followed by generate_artifact writes the backend, verifies it by importing it and + launches it. Recall this when editing an existing fullstack artifact, or when + generate_artifact failed and the user asked you to continue manually: it is the hard + contract — canonical FastAPI+Mangum backend.py template, SECRETS handling, /api/* + route prefix rules, static/ frontend layout, requirements.txt, launch and preview. + Covers backend, API and server code.' metadata: display_name: Backend & fullstack app generation provenance: builtin diff --git a/anton/core/memory/builtin_skills/build-html-dashboard/SKILL.md b/anton/core/memory/builtin_skills/build-html-dashboard/SKILL.md index bbef3ef3..22925c5d 100644 --- a/anton/core/memory/builtin_skills/build-html-dashboard/SKILL.md +++ b/anton/core/memory/builtin_skills/build-html-dashboard/SKILL.md @@ -1,11 +1,11 @@ --- name: build-html-dashboard -description: 'MANDATORY reading before building ANY HTML dashboard, chart, plot, interactive - report, or browser-based visualization (create_artifact type="html-app", or the - frontend of a fullstack app). Contains the full HTML output contract: self-contained - file rules, Apache ECharts setup, dark theme, layout/design standards, and large-dataset - handling. Recall it BEFORE writing the first line of dashboard HTML. When in doubt, - recall it.' +description: 'ONLY for writing dashboard or chart HTML BY HAND. NOT needed on the normal + path — create_artifact(type="html-app") followed by generate_artifact writes the + dashboard itself and carries this contract internally. Recall this when editing an + existing html-app, or when generate_artifact failed and the user asked you to continue + manually: it is the output contract — self-contained file rules, Apache ECharts setup, + dark theme, layout/design standards, large-dataset handling.' metadata: display_name: HTML dashboard & visualization output format provenance: builtin diff --git a/anton/core/session.py b/anton/core/session.py index 77eece9a..2fa34356 100644 --- a/anton/core/session.py +++ b/anton/core/session.py @@ -57,6 +57,7 @@ from anton.core.tools.registry import ToolRegistry from anton.core.tools.tool_defs import ( CREATE_ARTIFACT_TOOL, + GENERATE_ARTIFACT_TOOL, LAUNCH_BACKEND_TOOL, LIST_ARTIFACTS_TOOL, MEMORIZE_TOOL, @@ -943,6 +944,7 @@ def _build_core_tools(self) -> None: self.tool_registry.register_tool(OPEN_ARTIFACT_TOOL) self.tool_registry.register_tool(UPDATE_ARTIFACT_METADATA_TOOL) self.tool_registry.register_tool(LAUNCH_BACKEND_TOOL) + self.tool_registry.register_tool(GENERATE_ARTIFACT_TOOL) async def close(self) -> None: """Clean up scratchpads and other resources.""" diff --git a/anton/core/tools/generate_artifact/__init__.py b/anton/core/tools/generate_artifact/__init__.py new file mode 100644 index 00000000..25368f5a --- /dev/null +++ b/anton/core/tools/generate_artifact/__init__.py @@ -0,0 +1,14 @@ +"""Deterministic generation FSM behind the ``generate_artifact`` tool. + +Public entry point: ``generate(session, artifact_type, artifact_path, +context, slug)``. The outer tool handler validates input and reads artifact +metadata; everything below this surface is provider-agnostic. ``generate`` +builds a ``GenState`` and hands off to ``orchestrator.run``, which walks the +graph nodes (data-sufficiency loop → tech spec → api spec → backend/frontend +generation with verification → launch & health check). Nodes reach real data +via the `scratchpad` sub-tool, guided by the brief's ``## Data`` section. +""" + +from .engine import generate, MAX_ROUNDS + +__all__ = ["generate", "MAX_ROUNDS"] diff --git a/anton/core/tools/generate_artifact/debug_trace.py b/anton/core/tools/generate_artifact/debug_trace.py new file mode 100644 index 00000000..dcec486b --- /dev/null +++ b/anton/core/tools/generate_artifact/debug_trace.py @@ -0,0 +1,107 @@ +"""Optional, generation-scoped step trace for the `generate_artifact` tool. + +Activated by the `ANTON_DEBUG_ARTIFACT_GENERATE_TOOL` env var: when its value +is a non-empty path, `make_trace()` returns a `GenTrace` that appends one +JSON-lines event per step to that file; otherwise a `NullTrace` whose methods +are all no-ops, so call sites never branch. + +The trace is stored on `GenState` and every FSM node logs with an explicit +`node` label. That explicitness is deliberate: the backend and frontend +sub-loops run under `asyncio.gather`, so an ambient "current node" would race. + +Every write is best-effort — the logger must never raise into generation code. +""" + +from __future__ import annotations + +import datetime +import json +import os +from pathlib import Path +from typing import Any + + +class NullTrace: + """No-op trace: identical surface to `GenTrace`, does nothing.""" + + def run_start(self, **_: Any) -> None: ... + def node(self, *_: Any, **__: Any) -> None: ... + def llm_call(self, **_: Any) -> None: ... + def verdict(self, **_: Any) -> None: ... + def scratchpad(self, **_: Any) -> None: ... + def file_written(self, **_: Any) -> None: ... + def verifier(self, **_: Any) -> None: ... + def run_result(self, **_: Any) -> None: ... + + +class GenTrace: + """Append-only JSON-lines writer for one process's generation runs.""" + + def __init__(self, path: str) -> None: + self._path = Path(path) + + def _emit(self, event: str, payload: dict) -> None: + try: + rec = { + "ts": datetime.datetime.now().isoformat(timespec="seconds"), + "event": event, + **payload, + } + with open(self._path, "a", encoding="utf-8") as f: + f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n") + except Exception: + # Best-effort: never let the logger break generation. + pass + + def run_start(self, *, slug, artifact_type, artifact_path, brief, is_fullstack) -> None: + self._emit("run_start", { + "slug": slug, + "artifact_type": artifact_type, + "artifact_path": str(artifact_path), + "brief": brief, + "is_fullstack": is_fullstack, + }) + + def node(self, node, outcome, detail="") -> None: + self._emit("node", {"node": node, "outcome": outcome, "detail": detail}) + + def llm_call(self, *, node, method, system, messages, + response=None, value=None, attempt=None, round=None) -> None: + from anton.core.llm.serialize import serialize_messages, serialize_response + + if response is not None: + resp = serialize_response(response) + else: + resp = {"structured": value} + self._emit("llm_call", { + "node": node, + "method": method, + "attempt": attempt, + "round": round, + "system": system, + "messages": serialize_messages(messages), + "response": resp, + }) + + def verdict(self, *, node, schema, value) -> None: + self._emit("verdict", {"node": node, "schema": schema, "value": value}) + + def scratchpad(self, *, node, input, output) -> None: + self._emit("scratchpad", {"node": node, "input": input, "output": output}) + + def file_written(self, *, node, path) -> None: + self._emit("file_written", {"node": node, "path": path}) + + def verifier(self, *, node, ok, errors, warnings) -> None: + self._emit("verifier", { + "node": node, "ok": ok, "errors": errors, "warnings": warnings, + }) + + def run_result(self, *, ok, result=None, error=None) -> None: + self._emit("run_result", {"ok": ok, "result": result, "error": error}) + + +def make_trace() -> GenTrace | NullTrace: + """Return a `GenTrace` when the debug env var names a file, else `NullTrace`.""" + path = os.environ.get("ANTON_DEBUG_ARTIFACT_GENERATE_TOOL", "").strip() + return GenTrace(path) if path else NullTrace() diff --git a/anton/core/tools/generate_artifact/engine.py b/anton/core/tools/generate_artifact/engine.py new file mode 100644 index 00000000..62d363a6 --- /dev/null +++ b/anton/core/tools/generate_artifact/engine.py @@ -0,0 +1,438 @@ +"""Async tool-call loop(s) that drive the inner generation LLM. + +For html-app: single loop. +For fullstack-stateless-app and fullstack-stateful-app: + 1. One-shot planning call → OpenAPI specification (JSON, kept in memory). + 2. asyncio.gather → backend loop + frontend loop in parallel. + +The sub-generator reaches real data itself through the `scratchpad` sub-tool, +guided by the free-form `## Data` section of the brief (which names the +scratchpads/cells the main agent already used). The engine no longer fabricates +test data or pre-pickles variables. + +The loop protocol is Anthropic tool-use / tool-result blocks, which both +providers Anton ships (AnthropicProvider, OpenAIProvider) accept on input. +""" + +from __future__ import annotations + +import asyncio +import json +from pathlib import Path +from typing import TYPE_CHECKING + +from . import sub_tools +from .prompts import ( + build_api_spec_prompt, + build_backend_kickoff, + build_backend_system_prompt, + build_frontend_kickoff, + build_frontend_system_prompt, + build_subagent_system_prompt, + build_user_kickoff, +) + +if TYPE_CHECKING: + from anton.chat_session import ChatSession + + +# Higher than the old 12 because the sub-generator also spends rounds on +# scratchpad calls (pulling/rebuilding data) on top of writing files, and higher +# than 16 because a file is now built in chunks (`mode="a"`) rather than in one +# call — head, sections, scripts and closing tags each cost a round unless the +# model batches them. One shared cap: it also gates `fetch_data_sample`, and a +# runaway there is bounded by DATA_LOOP_MAX anyway. +MAX_ROUNDS = 20 + + +# Two distinct rejections, because the causes differ and the model must react +# differently. Both wordings point at the only working way out — chunked writing +# (see _WRITE_DISCIPLINE in prompts.py and the design spec, 3.1). +# +# CRITICAL in _TRUNCATED_MSG: state that the EARLIER calls in the same reply did +# land. Truncation is streaming — only the last block is incomplete, the rest +# arrived in full. Without saying so, the model re-sends the chunks it already +# wrote and mode="a" duplicates them in the file. +_TRUNCATED_MSG = ( + "Error: THIS tool call was cut off by the output limit and wrote nothing. " + "The earlier tool calls in this same reply DID take effect — do NOT re-send " + "them, or `mode=\"a\"` will duplicate their content. Continue from where the " + "file now ends, appending with `mode=\"a\"` and keeping each call's " + "`content` well under ~6 KB." +) + +_NO_CONTENT_MSG = ( + "Error: `content` was not delivered, so nothing was written. Re-emit this " + "chunk with a non-empty `content`, well under ~6 KB, appending with " + "`mode=\"a\"` if the file already has earlier chunks." +) + + +def _output_token_cap(session) -> int | None: + """The client's effective output cap, or None when it cannot be read. + + `LLMClient` exposes no public accessor — only the private `_max_tokens` + (`anton/core/llm/client.py:58`). In tests the session is an `AsyncMock` where + the attribute exists and is truthy but is not a number, so the type check is + mandatory: without it the detection would fire on every test. + """ + cap = getattr(getattr(session, "_llm", None), "_max_tokens", None) + return cap if isinstance(cap, int) and cap > 0 else None + + +def _response_is_truncated(response, cap: int | None) -> bool: + """The reply hit the output cap, so its last tool call was not delivered. + + `stop_reason` cannot be relied on: the gateway reports `'stop'` on truncation + too (stage-1a measurement, findings 2-3). `output_tokens` is sometimes None — + then the signal is unknown and we do NOT flag: a false rejection is worse than + a miss. + """ + if not cap: + return False + used = getattr(getattr(response, "usage", None), "output_tokens", None) + return isinstance(used, int) and used >= cap + + +# --------------------------------------------------------------------------- +# Public entry point +# --------------------------------------------------------------------------- + + +async def generate( + *, + session: "ChatSession", + artifact_type: str, + artifact_path: Path, + context: str, + slug: str, + primary: str | None = None, +) -> dict | str: + """Drive the artifact-generation FSM to populate ``artifact_path``. + + Returns a result dict (with a step ``trace``) on success, or a single + error string naming the node where the machine stopped. + """ + from .orchestrator import run + from .state import GenState + from .debug_trace import make_trace + + if artifact_type not in ( + "html-app", "fullstack-stateless-app", "fullstack-stateful-app" + ): + return f"Error: unsupported artifact type: {artifact_type!r}" + + trace = make_trace() + trace.run_start( + slug=slug, + artifact_type=artifact_type, + artifact_path=artifact_path, + brief=context, + is_fullstack=artifact_type != "html-app", + ) + + state = GenState( + session=session, + artifact_type=artifact_type, + artifact_path=artifact_path, + slug=slug, + brief=context, + is_fullstack=artifact_type != "html-app", + primary=primary, + trace_log=trace, + ) + result = await run(state) + if isinstance(result, str): + trace.run_result(ok=False, error=result) + else: + trace.run_result(ok=True, result=result) + return result + + +# --------------------------------------------------------------------------- +# Pre-generation steps +# --------------------------------------------------------------------------- + + +async def _generate_api_spec( + session: "ChatSession", + context: str, + *, + stateless: bool = False, + trace=None, + node_label: str = "make_api_spec", +) -> str: + """One-shot planning call → OpenAPI specification (JSON). + + The model is asked for an OpenAPI document as JSON. We validate the + response by parsing it with ``json.loads``; if parsing succeeds the spec + is considered valid and the (normalized) JSON string is returned. + """ + system, user = build_api_spec_prompt(context, stateless=stateless) + response = await session._llm.plan( + system=system, + messages=[{"role": "user", "content": user}], + ) + if trace is not None: + trace.llm_call( + node=node_label, + method="plan", + system=system, + messages=[{"role": "user", "content": user}], + response=response, + ) + spec = _strip_code_fence((response.content or "").strip()) + if not spec: + return "Error: API spec generation returned empty response." + try: + parsed = json.loads(spec) + except json.JSONDecodeError as exc: + return f"Error: API spec is not valid JSON: {exc}" + return json.dumps(parsed, indent=2, ensure_ascii=False) + + +def _strip_code_fence(text: str) -> str: + """Strip a leading/trailing markdown code fence if present. + + Models often wrap JSON in ```json ... ``` despite being asked for raw JSON. + """ + if not text.startswith("```"): + return text + lines = text.splitlines() + lines = lines[1:] # drop opening ```json / ``` line + if lines and lines[-1].strip() == "```": + lines = lines[:-1] + return "\n".join(lines).strip() + + +# --------------------------------------------------------------------------- +# Generic bounded tool-call loop +# --------------------------------------------------------------------------- + + +async def _run_loop( + *, + session: "ChatSession", + system: str, + kickoff: str, + artifact_path: Path, + node_label: str, + attempt: int | None = None, + trace=None, + step_injections: list[tuple[str, str]] | None = None, + require_files: bool = True, +) -> dict | str: + """Run one bounded sub-agent tool-call loop. + + ``step_injections`` is an optional list of ``(trigger_filename, message)`` + pairs. When a ``write_file`` call successfully writes ``trigger_filename``, + ``message`` is appended to the tool-result content so the model receives + the next-step instruction in the same turn. Each trigger fires at most once. + + Returns a result dict ``{files_written, rounds_used, summary, + scratchpad_execs}`` on success, or a plain error string on failure. + ``scratchpad_execs`` records every scratchpad ``exec`` the loop made — + ``{name, code, output}`` per call — so callers can hand the exact + data-access code that ran to later FSM steps. + """ + tools = sub_tools.tool_schemas() + cap = _output_token_cap(session) + messages: list[dict] = [{"role": "user", "content": kickoff}] + + files_written: list[str] = [] + scratchpad_execs: list[dict] = [] + finished_summary: str | None = None + injected: set[str] = set() + + for round_idx in range(MAX_ROUNDS): + # First round: use the planning model for highest-quality initial generation. + # Subsequent rounds (retries, read_file refinements) use the coding model. + llm_call = session._llm.plan if round_idx == 0 else session._llm.code + response = await llm_call( + system=system, + messages=messages, + tools=tools, + ) + + if trace is not None: + trace.llm_call( + node=node_label, + method="plan" if round_idx == 0 else "code", + system=system, + messages=messages, + response=response, + attempt=attempt, + round=round_idx, + ) + + # Truncation is streaming: only the LAST block of the reply is + # incomplete, the earlier ones arrived in full. Rejecting them all would + # throw away finished work every round, and if the model consistently + # spends its whole output budget (measured: output_tokens at the cap in + # roughly 20 of 32 rounds) nothing would ever be written across all + # rounds. That would be the same failure this work fixes, only with a + # clearer error text. + truncated_tc_id = ( + response.tool_calls[-1].id + if _response_is_truncated(response, cap) and response.tool_calls + else None + ) + + if not response.tool_calls: + tail = (response.content or "").strip() + return ( + f"generator stopped without writing files " + f"(round {round_idx + 1}/{MAX_ROUNDS}). " + f"Last output: {tail[:300]!r}" + ) + + assistant_blocks: list[dict] = [] + if response.content: + assistant_blocks.append({"type": "text", "text": response.content}) + for tc in response.tool_calls: + assistant_blocks.append( + { + "type": "tool_use", + "id": tc.id, + "name": tc.name, + "input": tc.input, + } + ) + messages.append({"role": "assistant", "content": assistant_blocks}) + + result_blocks: list[dict] = [] + for tc in response.tool_calls: + if tc.parse_error: + result_blocks.append( + { + "type": "tool_result", + "tool_use_id": tc.id, + "content": ( + "Error: malformed tool input — re-emit with valid " + f"JSON. ({tc.parse_error})" + ), + } + ) + continue + + name = tc.name + inp = tc.input or {} + + if name == "finish": + summary = str(inp.get("summary") or "").strip() + finished_summary = summary or "(no summary)" + result_blocks.append( + {"type": "tool_result", "tool_use_id": tc.id, "content": "ok"} + ) + elif name == "write_file": + # `inp.get("content")` without a "" default: a missing key must + # be distinguishable from a deliberately empty string, or the + # error text is guessing. Both forms are rejected either way, but + # with different messages — their causes differ. + content = inp.get("content") + if tc.id == truncated_tc_id: + result_blocks.append( + { + "type": "tool_result", + "tool_use_id": tc.id, + "content": _TRUNCATED_MSG, + } + ) + continue + if not content: + result_blocks.append( + { + "type": "tool_result", + "tool_use_id": tc.id, + "content": _NO_CONTENT_MSG, + } + ) + continue + res = sub_tools.write_file( + artifact_path, + inp.get("path", ""), + content, + mode=inp.get("mode", "w"), + ) + msg = res["message"] + if res.get("ok"): + written = res["written"] + if written not in files_written: + files_written.append(written) + if trace is not None: + trace.file_written(node=node_label, path=written) + for trigger, inject_msg in (step_injections or []): + if written == trigger and inject_msg not in injected: + injected.add(inject_msg) + msg = f"{msg}\n\n{inject_msg}" + result_blocks.append( + { + "type": "tool_result", + "tool_use_id": tc.id, + "content": msg, + } + ) + elif name == "read_file": + res = sub_tools.read_file(artifact_path, inp.get("path", "")) + result_blocks.append( + { + "type": "tool_result", + "tool_use_id": tc.id, + "content": res["message"], + } + ) + elif name == "scratchpad": + # Full scratchpad access: the sub-generator pulls or rebuilds + # the data described in the brief's `## Data` section. Lazy + # import avoids a tool_handlers <-> generate_artifact cycle. + from anton.core.tools.tool_handlers import handle_scratchpad + + content = await handle_scratchpad(session, inp) + if inp.get("action") == "exec": + scratchpad_execs.append( + { + "name": str(inp.get("name") or ""), + "code": str(inp.get("code") or ""), + "output": content if isinstance(content, str) else str(content), + } + ) + if trace is not None: + trace.scratchpad(node=node_label, input=inp, output=content) + result_blocks.append( + { + "type": "tool_result", + "tool_use_id": tc.id, + "content": content, + } + ) + else: + result_blocks.append( + { + "type": "tool_result", + "tool_use_id": tc.id, + "content": ( + f"Error: unknown sub-tool `{name}`. " + "Use write_file, read_file, or finish." + ), + } + ) + + messages.append({"role": "user", "content": result_blocks}) + + if finished_summary is not None: + break + else: + return ( + f"generator exceeded round budget ({MAX_ROUNDS}) after writing " + f"{len(files_written)} file(s): {files_written}." + ) + + if require_files and not files_written: + return "generator finished without writing any files." + + return { + "files_written": files_written, + "rounds_used": round_idx + 1, + "summary": finished_summary, + "scratchpad_execs": scratchpad_execs, + } diff --git a/anton/core/tools/generate_artifact/orchestrator.py b/anton/core/tools/generate_artifact/orchestrator.py new file mode 100644 index 00000000..d3b01308 --- /dev/null +++ b/anton/core/tools/generate_artifact/orchestrator.py @@ -0,0 +1,828 @@ +"""Deterministic FSM that drives artifact generation (see design spec). + +`run(state)` walks the graph nodes. Diamond nodes call +`session._llm.generate_object(...)`; the data-fetch and code-generation nodes +reuse `engine._run_loop`; verification uses `verifiers`. +""" +from __future__ import annotations + +import inspect +import re + +from . import engine, prompts +from .prompts import HTML_APP_DEFAULT_PRIMARY +from .state import ( + DATA_LOOP_MAX, + DataVerdict, + FetchVerdict, + GenState, + RequiredData, + VerifyResult, +) + + +async def _decide(state: GenState, schema, prompt_pair, node: str) -> object: + system, user = prompt_pair + result = await state.session._llm.generate_object( + schema, system=system, messages=[{"role": "user", "content": user}] + ) + value = result.model_dump() if hasattr(result, "model_dump") else result + state.trace_log.llm_call( + node=node, method="generate_object", + system=system, messages=[{"role": "user", "content": user}], + value=value, + ) + state.trace_log.verdict(node=node, schema=getattr(schema, "__name__", str(schema)), value=value) + return result + + +# Caps for the exec-code record appended to data_notes: per-cell code, per-cell +# output snippet, and the whole section. Oldest cells are dropped first — the +# most recent ones are the ones that worked. +EXEC_CODE_MAX = 2000 +EXEC_OUTPUT_MAX = 300 +EXEC_NOTES_MAX = 8000 + + +def _render_exec_notes( + execs: list[dict], *, header: str = "### Code executed while fetching" +) -> str: + """Deterministic record of the Python the fetch step ran. + + Appended to data_notes so later steps (tech spec, backend generation) see + the exact working data-access code instead of relying on the model's + `finish` summary to mention it. + """ + blocks: list[str] = [] + for e in execs: + code = (e.get("code") or "").strip() + if not code: + continue + if len(code) > EXEC_CODE_MAX: + code = code[:EXEC_CODE_MAX] + "\n# … truncated …" + out = " ".join((e.get("output") or "").split()) + if len(out) > EXEC_OUTPUT_MAX: + out = out[:EXEC_OUTPUT_MAX] + " …" + block = f"Scratchpad `{e.get('name')}`:\n```python\n{code}\n```" + if out: + block += f"\nOutput: {out}" + blocks.append(block) + dropped = 0 + while blocks and sum(len(b) for b in blocks) > EXEC_NOTES_MAX: + blocks.pop(0) + dropped += 1 + if not blocks: + return "" + if dropped: + header += f" (first {dropped} cell(s) omitted for size)" + return header + "\n" + "\n\n".join(blocks) + + +async def _fetch_data_sample(state: GenState) -> str: + """Run a scratchpad loop that pulls a data sample; return its summary.""" + result = await engine._run_loop( + session=state.session, + system=prompts.build_fetch_data_system_prompt( + state.artifact_path, + datasource_context=_datasource_context(state.session), + public_sources=_public_data_sources_skill(state.session), + ), + kickoff=prompts.build_fetch_data_kickoff(state), + artifact_path=state.artifact_path, + require_files=False, + node_label="fetch_data_sample", + trace=state.trace_log, + ) + if isinstance(result, str): + # Loop failed — surface as a note so the next is_data_enough sees it. + return f"(data fetch step reported: {result})" + summary = result.get("summary") or "(fetch produced no summary)" + exec_notes = _render_exec_notes(result.get("scratchpad_execs") or []) + if exec_notes: + summary += "\n\n" + exec_notes + return summary + + +def _pads_dict(session) -> dict: + """Live scratchpad runtimes, or {} when this is not a real manager. + + The `isinstance(..., dict)` check does double duty. The real + `ScratchpadManager.pads` is a genuine dict (`backends/manager.py:33` returns + `self._pads`), while in tests the session is an `AsyncMock` whose `pads` is a + mock. Rejecting it by type means we never call `.keys()` on it, so we never + create an un-awaited coroutine — one would surface as a RuntimeWarning at GC + time and pollute the output of every test using a mock session. See + `_list_connections`, where such a coroutine has to be closed by hand; here it + simply never comes into existence. + """ + try: + pads = session._scratchpads.pads + except Exception: # noqa: BLE001 + return {} + return pads if isinstance(pads, dict) else {} + + +def _pads_named_in_brief(session, brief: str) -> list[str]: + """Names of live scratchpads that the brief mentions. + + Matching runs from the live pads to the brief text, not the other way round. + Persistent pads cannot be enumerated (`Cell` does not store its pad name, + `base.py:15-26`), and on this branch `get_or_create` does NOT materialise a + pad from replayed cells: `ChatSessionConfig.cells` is declared as `None` + (`session.py:261`) and no production caller sets it. So we work only with + what is already live in the process — and a false match costs a few tokens in + the notes rather than a venv provisioning. + + Word boundaries are mandatory: a pad named `dash` would otherwise match the + word "dashboard", which a dashboard brief is guaranteed to contain. + """ + matched = [ + name + for name in _pads_dict(session) + if isinstance(name, str) + and name.strip() + and re.search(rf"\b{re.escape(name)}\b", brief) + ] + return sorted(matched) + + +def _live_pad_names(session) -> list[str]: + """Live pad names — used only for the skip record. Never raises.""" + return sorted(str(k) for k in _pads_dict(session)) + + +def _inspect_named_scratchpads(state: GenState) -> None: + """Read the cells of pads named in the brief into `data_notes`. + + Reads `pad.cells` directly (`base.py:52`) rather than going through `dump`: + that returns a ready-made markdown string from `render_notebook()`, which + `_render_exec_notes` cannot consume, and it carries every cell's full code + with no truncation. `pad.cells` maps into the required shape as-is, so the + existing caps come for free. + + The step is synchronous and side-effect free: no `await`, no `get_or_create`. + """ + matched = _pads_named_in_brief(state.session, state.brief) + if not matched: + live = _live_pad_names(state.session) + detail = ( + "no scratchpad named in the brief matched the live ones: " + + ", ".join(live) + if live + else "no live scratchpads in this session" + ) + state.record("inspect_scratchpads", "skipped", detail) + return + + pads = _pads_dict(state.session) + execs: list[dict] = [] + for name in matched: + for cell in getattr(pads[name], "cells", None) or []: + code = getattr(cell, "code", "") or "" + if not code.strip(): + continue + execs.append( + { + "name": name, + "code": code, + # stdout or the error text: a failed cell would otherwise + # land in the notes as code with no result — indistinguishable + # from a successful one that printed nothing. "What was tried + # and what did not work" is exactly what the inspection needs. + "output": getattr(cell, "stdout", "") or getattr(cell, "error", "") or "", + } + ) + + # Own header: the default "### Code executed while fetching" would be a lie + # here — nothing was fetched, these are the main agent's cells. It would also + # contradict the `is_data_enough` instruction, which teaches the node to count + # this data as available "regardless of who obtained it". + notes = _render_exec_notes( + execs, + header="### Cells the main agent already ran in: " + ", ".join(matched), + ) + if not notes: + state.record( + "inspect_scratchpads", "empty", + "pads named in the brief have no cells: " + ", ".join(matched), + ) + return + + state.data_notes = (state.data_notes + "\n\n" + notes).strip() + state.record( + "inspect_scratchpads", "done", + f"{len(execs)} cell(s) from {', '.join(matched)}", + ) + + +async def _data_phase(state: GenState) -> str | None: + """is_data_enough ↔ define_required_data → is_possible_to_fetch → fetch. + + Before the first `is_data_enough`, inspect the pads named in the brief: what + the outer agent already gathered does not need gathering again. The phase + itself is not trimmed — only its input changes. + """ + _inspect_named_scratchpads(state) + last_reasoning = "" + for _ in range(DATA_LOOP_MAX + 1): + verdict: DataVerdict = await _decide( + state, DataVerdict, prompts.build_data_enough_prompt(state), "is_data_enough" + ) + if verdict.enough: + state.record("is_data_enough", "yes", verdict.reasoning) + return None + state.record("is_data_enough", "no", verdict.reasoning) + + if state.data_iterations >= DATA_LOOP_MAX: + break + + required: RequiredData = await _decide( + state, RequiredData, prompts.build_required_data_prompt(state), "define_required_data" + ) + required_text = "\n".join( + f"- {it.name} — from {it.where} ({it.why})" for it in required.items + ) or required.reasoning + state.record("define_required_data", "done", required_text) + last_reasoning = required.reasoning + + can: FetchVerdict = await _decide( + state, FetchVerdict, prompts.build_can_fetch_prompt(state, required_text), "is_possible_to_fetch" + ) + if not can.possible: + state.record("is_possible_to_fetch", "no", can.reasoning) + state.error = f"not enough data: {can.reasoning}" + return state.error + state.record("is_possible_to_fetch", "yes", can.reasoning) + last_reasoning = can.reasoning + + notes = await _fetch_data_sample(state) + state.data_iterations += 1 + state.data_notes = (state.data_notes + "\n\n" + notes).strip() + state.record("fetch_data_sample", "done", notes[:200]) + + state.error = ( + f"not enough data: the data loop did not converge within {DATA_LOOP_MAX} " + f"iterations. Last assessment: {last_reasoning}" + ) + return state.error + + +# --------------------------------------------------------------------------- +# Spec + generation nodes +# --------------------------------------------------------------------------- + +from . import verifiers +from .state import GEN_VERIFY_MAX_RETRIES + + +async def _write_tech_spec(state: GenState) -> str | None: + system, user = prompts.build_tech_spec_prompt(state) + resp = await state.session._llm.plan( + system=system, messages=[{"role": "user", "content": user}] + ) + state.trace_log.llm_call( + node="make_tech_spec", method="plan", + system=system, messages=[{"role": "user", "content": user}], + response=resp, + ) + body = (getattr(resp, "content", "") or "").strip() + if not body: + state.error = "make_tech_spec: model returned an empty specification." + return state.error + (state.artifact_path / "spec.md").write_text(body, encoding="utf-8") + if "spec.md" not in state.internal_files: + state.internal_files.append("spec.md") + state.record("make_tech_spec", "done", "wrote spec.md") + return None + + +def _spec_context(state: GenState) -> str: + """Brief + gathered data + tech spec — the shared context handed to + api-spec and generation nodes (explicit inter-node data flow).""" + parts = [state.brief.strip()] + if state.data_notes.strip(): + parts.append("## Data\n" + state.data_notes.strip()) + spec_path = state.artifact_path / "spec.md" + if spec_path.is_file(): + parts.append("## Technical specification\n" + spec_path.read_text(encoding="utf-8")) + journal = state.journal() + if journal: + parts.append("## Progress journal (steps completed so far)\n" + journal) + return "\n\n".join(parts) + + +async def _make_api_spec(state: GenState) -> str | None: + stateless = state.artifact_type == "fullstack-stateless-app" + spec_or_err = await engine._generate_api_spec( + state.session, _spec_context(state), stateless=stateless, + trace=state.trace_log, node_label="make_api_spec", + ) + if spec_or_err.startswith("Error:"): + state.error = f"make_api_spec: {spec_or_err}" + return state.error + state.api_spec = spec_or_err + (state.artifact_path / "openapi.json").write_text(spec_or_err, encoding="utf-8") + if "openapi.json" not in state.internal_files: + state.internal_files.append("openapi.json") + state.record("make_api_spec", "done", "wrote openapi.json") + return None + + +def _vault(session): + """The session's vault, or a local one. Mirrors _map_datasources and handle_update_artifact.""" + from anton.core.datasources.data_vault import LocalDataVault + + return getattr(session, "_data_vault", None) or LocalDataVault() + + +def _list_connections(vault) -> list[dict]: + """`list_connections()` as a list, or [] when this is not a real vault. + + A returned coroutine is closed explicitly: in tests the session is an + `AsyncMock` whose `_data_vault` is truthy while `list_connections()` yields a + coroutine. Swallowing the TypeError is not enough — an abandoned coroutine + surfaces as a RuntimeWarning at GC time, outside any `catch_warnings` block. + """ + try: + raw = vault.list_connections() + except Exception: # noqa: BLE001 + return [] + if inspect.iscoroutine(raw): + raw.close() + return [] + try: + return [c for c in raw if isinstance(c, dict)] + except TypeError: + return [] + + +def _datasource_context(session) -> str: + """The `## Connected Data Sources` section (DS_* names, no values), or "".""" + try: + vault = _vault(session) + if not _list_connections(vault): + return "" + from anton.utils.datasources import build_datasource_context + + return build_datasource_context(vault) or "" + except Exception: # noqa: BLE001 + return "" + + +_PUBLIC_SOURCES_SKILL = "public-data-sources" + + +def _public_data_sources_skill(session) -> str: + """Body of the built-in public-data-sources skill, or "". + + The catalog must not be copied into the prompt — it already lives in + `builtin_skills/public-data-sources/SKILL.md` and is maintained there. The + sub-generator has no `recall_skill`, so the body is fed in directly from the + same `SkillStore` the main agent uses. + + In tests the session is an `AsyncMock`, and `store.load(...)` returns a + **coroutine**. Type-checking `declarative_md` alone is not enough: an + abandoned coroutine surfaces as a RuntimeWarning at GC time, in the output of + an arbitrary test rather than the one that created it. So it is closed + explicitly — the same treatment as in `_list_connections`. (In `_pads_dict` no + coroutine appears at all, because the mock is rejected by type before any + call; here the call is unavoidable.) + """ + try: + store = getattr(session, "_skill_store", None) + skill = store.load(_PUBLIC_SOURCES_SKILL) if store is not None else None + if inspect.iscoroutine(skill): + skill.close() + return "" + body = getattr(skill, "declarative_md", None) + return body if isinstance(body, str) else "" + except Exception: # noqa: BLE001 + return "" + + +def _known_connection_hints(session) -> list[str]: + """` → DS___` for every connection. Never raises. + + Hands over the ready env prefix, not just the slug: turning `-` + into `DS__` sanitises special characters (`prod-db.eu` → + `PROD_DB_EU`, see `_slug_env_prefix`). Asking the model to derive it itself + means asking it to repeat exactly the normalisation it already got wrong — + otherwise it would not have reached this error. + """ + try: + from anton.core.artifacts.models import DatasourceRef + + hints = [] + for c in _list_connections(_vault(session)): + if not (c.get("engine") and c.get("name")): + continue + ref = DatasourceRef(engine=c["engine"], name=c["name"]) + hints.append(f"{ref.slug} → {ref.env_prefix}__") + return sorted(hints) + except Exception: # noqa: BLE001 + return [] + + +def _map_datasources(session, ds_keys: list[str]) -> tuple[list, list[str]]: + """Map DS_* env keys used by backend.py to vault `DatasourceRef`s. + + Returns `(refs, unmapped)`. A key maps when some connection's + `DS__` prefix matches `DS_..._`. Mirrors the vault + lookup already used by `handle_update_artifact` (tool_handlers.py): the + session attribute is `_data_vault`, falling back to `LocalDataVault()`. + """ + from anton.core.artifacts.models import DatasourceRef + from anton.core.datasources.data_vault import LocalDataVault, _slug_env_prefix + + vault = getattr(session, "_data_vault", None) or LocalDataVault() + conns = [ + (c["engine"], c["name"]) + for c in vault.list_connections() + if c.get("engine") and c.get("name") + ] + refs: list = [] + unmapped: list[str] = [] + for key in ds_keys: + match = next( + ((e, n) for (e, n) in conns if key.startswith(_slug_env_prefix(e, n) + "__")), + None, + ) + if match is None: + unmapped.append(key) + else: + ref = DatasourceRef(engine=match[0], name=match[1]) + if ref.slug not in {r.slug for r in refs}: + refs.append(ref) + return refs, unmapped + + +async def _declare_datasources(state: GenState, refs: list) -> None: + """Persist the mapped `DatasourceRef`s into artifact metadata.""" + if not refs: + return + from anton.core.tools.tool_handlers import _artifact_store + + store = _artifact_store(state.session) + if store is not None: + store.update(state.slug, datasources=refs) + state.record("declare_datasources", "done", ", ".join(r.slug for r in refs)) + + +async def _gen_verify_backend(state: GenState, extra_context: str = "") -> str | None: + stateless = state.artifact_type == "fullstack-stateless-app" + system = prompts.build_backend_system_prompt( + state.artifact_path, + stateless=stateless, + datasource_context=_datasource_context(state.session), + ) + verdict = None # guards the terminal message when no attempt ever verified + last_loop_error: str | None = None # see the comment in _gen_verify_frontend + extra = ("\n\n" + extra_context) if extra_context else "" + for attempt in range(GEN_VERIFY_MAX_RETRIES + 1): + kickoff = prompts.build_backend_kickoff(_spec_context(state), state.api_spec or "{}") + extra + result = await engine._run_loop( + session=state.session, system=system, kickoff=kickoff, + artifact_path=state.artifact_path, + node_label="generate_backend", attempt=attempt, trace=state.trace_log, + step_injections=[( + "backend.py", + "backend.py written. Now write requirements.txt listing EVERY " + "package imported in backend.py (one per line). Then call finish.", + )], + ) + if isinstance(result, str): + extra = f"\n\n## Previous attempt failed\n{result}\nFix it and try again." + state.record("generate_backend", "error", result) + last_loop_error = result + continue + for f in result["files_written"]: + if f not in state.files_written: + state.files_written.append(f) + state.record("generate_backend", "done", result.get("summary", "")) + + verdict, ds_keys = await verifiers.verify_backend( + scratchpad_pool=state.session._scratchpads, + slug=state.slug, artifact_path=state.artifact_path, + ) + state.trace_log.verifier( + node="verify_backend", ok=verdict.ok, + errors=list(verdict.errors), warnings=list(verdict.warnings), + ) + if verdict.ok: + # Spec: DS_* keys with no matching vault connection are errors. + refs, unmapped = _map_datasources(state.session, ds_keys) + if unmapped: + known = _known_connection_hints(state.session) + msg = ( + "backend reads DS_* env keys with no matching vault " + "connection: " + ", ".join(unmapped) + + ". Available connections and their env-var namespaces: " + + ("; ".join(known) or "(none)") + + ". Use one of those namespaces verbatim." + ) + state.record("verify_backend", "fail", msg) + verdict.errors.append(msg) + extra = "\n\n## Verification failed — fix these\n- " + msg + continue + state.record("verify_backend", "ok", "; ".join(verdict.warnings)) + await _declare_datasources(state, refs) + return None + state.record("verify_backend", "fail", "; ".join(verdict.errors)) + extra = ( + "\n\n## Verification failed — fix these\n" + + "\n".join(f"- {e}" for e in verdict.errors) + + ("\nWarnings:\n" + "\n".join(f"- {w}" for w in verdict.warnings) if verdict.warnings else "") + ) + detail = ( + "; ".join(verdict.errors) + if verdict is not None + else (last_loop_error or "generation did not produce a verifiable backend") + ) + state.error = "Backend verification failed after retry: " + detail + return state.error + + +def _read_frontend_html(state: GenState, written: list[str]) -> str | None: + if state.is_fullstack: + entry = state.artifact_path / "static" / "index.html" + return entry.read_text(encoding="utf-8") if entry.is_file() else None + # html-app: pick ONLY among what this run actually wrote. `primary` is the + # expectation, `written` is the fact, and the fact is what must be verified: + # otherwise, with no primary set, candidate #1 becomes the default + # `dashboard.html`, and a leftover file of that name from a previous + # generation would shadow the fresh `report.html`. Within what was written, + # primary takes priority — for the case where the loop wrote several .html. + written_html = [rel for rel in written if rel.endswith(".html")] + target = state.primary or HTML_APP_DEFAULT_PRIMARY + for rel in [r for r in written_html if r == target] + written_html: + p = state.artifact_path / rel + if p.is_file(): + return p.read_text(encoding="utf-8") + return None + + +async def _gen_verify_frontend(state: GenState) -> str | None: + if state.is_fullstack: + system = prompts.build_frontend_system_prompt(state.artifact_path) + else: + system = prompts.build_subagent_system_prompt( + state.artifact_type, state.artifact_path, primary=state.primary + ) + verdict = None # guards the terminal message when no attempt ever verified + # The loop's failure reason (round budget, no tool calls) is kept separately: + # it cannot ride on VerifyResult — a variable argument breaks the contract + # lock's AST walk (test_no_unresolvable_rule_literals). + last_loop_error: str | None = None + extra = "" + for attempt in range(GEN_VERIFY_MAX_RETRIES + 1): + if attempt > 0: + # With append, a retry would extend the truncated remains of the + # previous attempt. Delete deterministically: the "first chunk is + # always mode=\"w\"" rule is in the prompt, but cannot be relied on. + # Only for attempt > 0: before the first attempt the file may well be + # a working previous version of the artifact. + entry = ( + state.artifact_path / "static" / "index.html" + if state.is_fullstack + else state.artifact_path / (state.primary or HTML_APP_DEFAULT_PRIMARY) + ) + entry.unlink(missing_ok=True) + if state.is_fullstack: + kickoff = prompts.build_frontend_kickoff(_spec_context(state), state.api_spec or "{}") + extra + else: + kickoff = prompts.build_user_kickoff(_spec_context(state)) + extra + result = await engine._run_loop( + session=state.session, system=system, kickoff=kickoff, + artifact_path=state.artifact_path, + node_label="generate_frontend", attempt=attempt, trace=state.trace_log, + ) + if isinstance(result, str): + extra = f"\n\n## Previous attempt failed\n{result}\nFix it and try again." + state.record("generate_frontend", "error", result) + last_loop_error = result + continue + for f in result["files_written"]: + if f not in state.files_written: + state.files_written.append(f) + state.record("generate_frontend", "done", result.get("summary", "")) + + html = _read_frontend_html(state, result["files_written"]) + if html is None: + # One channel: this message rides on VerifyResult like every other + # check — otherwise the contract lock cannot see it and the terminal + # error loses its cause (verdict used to stay None). + verdict = VerifyResult(errors=[ + "No HTML entry file was written. Write static/index.html " + "(or the html-app page)." + ]) + else: + verdict = verifiers.verify_frontend(html, is_fullstack=state.is_fullstack) + state.trace_log.verifier( + node="verify_frontend", ok=verdict.ok, + errors=list(verdict.errors), warnings=list(verdict.warnings), + ) + if verdict.ok: + # The model may have named the file differently — bring metadata in + # line with the fact, or the renderer opens the wrong file. Only after + # successful verification: if both attempts fail, the attempt-1 cleanup + # deletes the file and a primary written earlier would dangle. + if not state.is_fullstack: + actual = next( + (f for f in result["files_written"] if f.endswith(".html")), None + ) + if actual and actual != (state.primary or HTML_APP_DEFAULT_PRIMARY): + from anton.core.tools.tool_handlers import _artifact_store + + store = _artifact_store(state.session) + if store is not None: + store.update(state.slug, primary=actual) + state.primary = actual + state.record("verify_frontend", "ok", "; ".join(verdict.warnings)) + return None + state.record("verify_frontend", "fail", "; ".join(verdict.errors)) + extra = ( + "\n\n## Verification failed — fix these\n" + + "\n".join(f"- {e}" for e in verdict.errors) + + ("\nWarnings:\n" + "\n".join(f"- {w}" for w in verdict.warnings) if verdict.warnings else "") + ) + detail = ( + "; ".join(verdict.errors) + if verdict is not None + else (last_loop_error or "generation did not produce a verifiable frontend") + ) + state.error = "Frontend verification failed after retry: " + detail + return state.error + + +# --------------------------------------------------------------------------- +# run_app / verify_fullstack + run() assembly +# --------------------------------------------------------------------------- + +import asyncio +import json +import urllib.error +import urllib.request + +from .state import RUNAPP_MAX_RETRIES + + +async def _launch_backend(**kwargs): + """Indirection so tests can monkeypatch the real launcher.""" + from anton.core.artifacts.backend_launcher import launch_artifact_backend + + return await launch_artifact_backend(**kwargs) + + +def _safe_get_paths(api_spec: str | None) -> list[str]: + """GET paths from openapi.json with no `{param}` and no required params.""" + if not api_spec: + return [] + try: + spec = json.loads(api_spec) + except json.JSONDecodeError: + return [] + out: list[str] = [] + for path, ops in (spec.get("paths") or {}).items(): + if "{" in path: + continue + get = (ops or {}).get("get") + if not isinstance(get, dict): + continue + params = get.get("parameters") or [] + if any(p.get("required") for p in params if isinstance(p, dict)): + continue + out.append(path) + return out + + +async def _probe_app(state: GenState, port: int) -> str | None: + """Health check + safe GET routes. Returns an error string or None.""" + base = f"http://127.0.0.1:{port}" + + def _get(url: str) -> tuple[int, str]: + try: + with urllib.request.urlopen(url, timeout=5) as resp: + return resp.status, resp.read(4096).decode(errors="replace") + except urllib.error.HTTPError as e: + return e.code, "" + except Exception as e: # noqa: BLE001 + return 0, str(e) + + status, _ = await asyncio.to_thread(_get, base + "/api/health") + if status != 200: + return f"health check GET /api/health returned {status} (expected 200)." + for path in _safe_get_paths(state.api_spec): + status, _ = await asyncio.to_thread(_get, base + path) + if status == 0 or status >= 500: + return f"endpoint GET {path} failed with status {status}." + return None + + +async def _tail_log(state: GenState, limit: int = 2000) -> str: + log = state.artifact_path / "backend.log" + if not log.is_file(): + return "" + text = log.read_text(encoding="utf-8", errors="replace") + return text[-limit:] + + +async def _run_and_verify_app(state: GenState) -> str | None: + # `_tracked_backends` is initialised on ChatSession (session.py), but use the + # same defensive getattr-or-create as handle_launch_backend for parity with + # non-standard session objects. + tracked = getattr(state.session, "_tracked_backends", None) + if not isinstance(tracked, dict): + tracked = {} + state.session._tracked_backends = tracked + problem = "" + for attempt in range(RUNAPP_MAX_RETRIES + 1): + launch = await _launch_backend( + slug=state.slug, + artifact_folder=state.artifact_path, + scratchpad_pool=state.session._scratchpads, + tracked_backends=tracked, + health_path="/api/health", + ) + if isinstance(launch, str): # launcher error (install / readiness) + state.record("run_app", "fail", launch) + problem = launch + else: + port = launch["port"] + from anton.core.tools.tool_handlers import _artifact_store + + store = _artifact_store(state.session) + if store is not None: + store.update(state.slug, port=port) + state.record("run_app", "done", f"port {port}") + probe_err = await _probe_app(state, port) + if probe_err is None: + state.record("verify_fullstack", "ok", f"port {port}") + return None + state.record("verify_fullstack", "fail", probe_err) + problem = probe_err + + if attempt >= RUNAPP_MAX_RETRIES: + break + # One backend-loop retry with the failure + log tail passed as kickoff + # context (NOT persisted into data_notes, so later steps stay clean). + tail = await _tail_log(state) + regen_err = await _gen_verify_backend( + state, + extra_context=( + f"## Previous launch failure\n{problem}\n\n" + f"## backend.log tail\n{tail}" + ), + ) + if regen_err is not None: + state.error = regen_err + return regen_err + + state.error = ( + f"Application failed to launch: {problem}\n\n" + f"--- backend.log tail ---\n{await _tail_log(state)}" + ) + return state.error + + +def _success(state: GenState) -> dict: + return { + "files_written": state.files_written, + # Generation input, not output: it physically sits in the artifact folder + # but is not an artifact for the user (see the design spec, 3.6). + "internal_files": state.internal_files, + "summary": "; ".join(f"{s.node}:{s.outcome}" for s in state.trace), + "trace": [{"node": s.node, "outcome": s.outcome, "detail": s.detail} for s in state.trace], + } + + +async def run(state: GenState) -> dict | str: + # is_data_enough loop + err = await _data_phase(state) + if err is not None: + return err + # make_tech_spec + err = await _write_tech_spec(state) + if err is not None: + return err + # is_fullstack (deterministic) + if not state.is_fullstack: + err = await _gen_verify_frontend(state) + if err is not None: + return err + return _success(state) + # fullstack: make_api_spec → parallel backend/frontend gen+verify + err = await _make_api_spec(state) + if err is not None: + return err + back_err, front_err = await asyncio.gather( + _gen_verify_backend(state), _gen_verify_frontend(state) + ) + if back_err is not None: + return back_err + if front_err is not None: + return front_err + # run_app → verify_fullstack (with one backend-loop retry) + err = await _run_and_verify_app(state) + if err is not None: + return err + return _success(state) diff --git a/anton/core/tools/generate_artifact/prompts.py b/anton/core/tools/generate_artifact/prompts.py new file mode 100644 index 00000000..da63b3f3 --- /dev/null +++ b/anton/core/tools/generate_artifact/prompts.py @@ -0,0 +1,745 @@ +"""System + kickoff prompts for the inner generation LLM. + +All prompt text lives here as dedicated constants — we do NOT re-use the +main-agent's ``BACKEND_GENERATION_PROMPT`` or ``VISUALIZATIONS_HTML_OUTPUT_FORMAT_PROMPT`` +verbatim because those are written for the *outer* agent's workflow (they include +artifact registration, scratchpad-cell discipline, ``launch_backend``, etc. +that are irrelevant and confusing to the sub-agent). + +Instead we extract only the technical rules relevant to the sub-agent's job: +write the files, nothing else. +""" + +from __future__ import annotations + +from pathlib import Path + + +# --------------------------------------------------------------------------- +# Canonical FSM graph — embedded in decision/generation prompts so every LLM +# call understands the whole pipeline and where its step sits. English only. +# --------------------------------------------------------------------------- + +FSM_DIGRAPH = """\ +digraph artifact_generation { + rankdir=TB; + is_data_enough [shape=diamond, label="Is there enough data to solve the task?"]; + define_required_data [shape=box, label="Determine the required data"]; + is_possible_to_fetch [shape=diamond, label="Is it possible to fetch the data?"]; + fetch_data_sample [shape=box, label="Fetch a data sample"]; + not_enough_data [shape=ellipse, label="Error: not enough data"]; + make_tech_spec [shape=box, label="Write a detailed technical specification (spec.md)"]; + is_fullstack [shape=diamond, label="Is a backend required? (derived from artifact type)"]; + make_api_spec [shape=box, label="Design the REST API specification (openapi.json)"]; + generate_backend [shape=box, label="Generate backend in a subagent"]; + verify_backend [shape=box, label="Verify and unit-test the backend"]; + generate_frontend [shape=box, label="Generate frontend in a subagent"]; + verify_frontend [shape=box, label="Verify the frontend"]; + run_app [shape=box, label="Launch the application"]; + verify_fullstack [shape=box, label="Verify the application is running"]; + + // entry node = is_data_enough + is_data_enough -> make_tech_spec [label="yes"]; + is_data_enough -> define_required_data [label="no"]; + define_required_data -> is_possible_to_fetch; + is_possible_to_fetch -> fetch_data_sample [label="yes"]; + is_possible_to_fetch -> not_enough_data [label="no"]; + fetch_data_sample -> is_data_enough; + make_tech_spec -> is_fullstack; + is_fullstack -> make_api_spec [label="yes"]; + is_fullstack -> generate_frontend [label="no"]; + make_api_spec -> generate_backend; + make_api_spec -> generate_frontend; + generate_backend -> verify_backend; + generate_frontend -> verify_frontend; + verify_backend -> run_app; + verify_frontend -> run_app; + run_app -> verify_fullstack; +} +""" + + +# --------------------------------------------------------------------------- +# Role / tool contract (shared across all artifact types) +# --------------------------------------------------------------------------- + +# Common half: fits both the nodes that write files and the fetch node, which +# does not. Everything about write_file lives in _ROLE_WRITE below. +_ROLE_COMMON = """\ +You are a focused, single-purpose worker inside an artifact-generation pipeline. +You do exactly the job your task section describes, then call `finish`. + +ALWAYS-AVAILABLE TOOLS: +- `scratchpad(action, name, ...)` — drive a persistent Python scratchpad + (`exec`, `view`, `dump`, `install`, `reset`, `remove`). Use it to reach the + real data described in the brief's `## Data` section. +- `finish(summary)` — terminate with a one-line summary. + +SCRATCHPAD DISCIPLINE (the same rules the main agent works under): +- The scratchpad starts with a clean namespace — nothing is pre-imported. Put + every import the cell needs at the top of THAT cell. Re-importing is free and + makes the cell work even if an earlier one failed. +- Each cell has a hard timeout of 120 seconds. On timeout the process is killed + and ALL state is lost — variables, imports, loaded data. Keep cells small; + split anything heavier across cells. +- Always `print(...)` what you want to see: the tool captures stdout, and a bare + expression at the end of a cell returns nothing. +- Connected data-source credentials arrive as environment variables named + `DS____` — read them from `os.environ`. NEVER read the + `data_vault` files directly. +- If a cell fails the same way twice, change strategy instead of re-running it: + different library, different query shape, a smaller batch. Repeating an + identical failing cell only burns the round budget. + +USING DATA: +- The brief's `## Data` section names the scratchpads and cells the main agent + already used, and what was done in them. `## Data gathered so far`, when + present, already contains those cells — read it before running anything. +- Use `scratchpad(action="exec", name="", code=...)` to pull or rebuild the + data you need (re-query, aggregate, reshape). Provide + `one_line_description` and `estimated_execution_time_seconds` on every `exec`. + Reuse the scratchpad name the brief gives you — a new name is an isolated + environment with none of the existing variables, imports or connection code.\ +""" + +# Write half: only for nodes that actually produce files. NOT mixed into the +# fetch node — there the role is immediately followed by "Do NOT write any +# artifact files", and the full _ROLE would contradict that instruction. +_ROLE_WRITE = """\ +YOUR OUTPUT IS FILES: produce them by calling `write_file`, then call `finish`. + +HARD RULES: +- Build every file with `write_file`. A large file MUST be written in several + chunks — one `mode="w"` call followed by `mode="a"` calls. A single call + carrying a whole file is cut off by the output limit and rejected. +- All `path` values are RELATIVE to the artifact folder — never write outside it. +- Call `finish(summary="")` exactly once when all files are written. + +FILE TOOLS: +- `write_file(path, content, mode="w"|"a")` — write a UTF-8 text file at + `/`. `"w"` creates or overwrites, `"a"` appends (creating the + file when absent). Default is `"w"`. +- `read_file(path)` — read a file you already wrote (for iterative refinement). + +DATA INTO FILES: +- For an html-app, the real data goes INTO the output file — but as its own + chunk: print the serialised data in a scratchpad cell, then append it with + `write_file(path, content, mode="a")` as a single `` + Initialise with `echarts.init(dom, 'dark')` and customise background to #0d1117. + NEVER use Plotly, matplotlib, or other chart libraries unless explicitly asked. +- Line smoothing: `smooth: false` on ALL line series by default. + Use `smooth: true` ONLY for cumulative / monotonic series (running totals, growth curves). + Line widths: 2.5 for primary, 1.5 for comparisons, 1 for reference lines. +- Chart readability: + - `axisLabel: { rotate: -45 }` on crowded axes. + - `grid: { containLabel: true }` so labels never clip. + - `legend: { type: 'scroll', bottom: 0 }` for many series. + - Pie/donut: `label: { show: true, position: 'outside' }` + `labelLayout: { hideOverlap: true }`. + - Rich `tooltip` with `formatter` functions for precise hover values. + - `dataZoom` on time series so users can zoom. +- Multi-tab dashboards: NEVER call `echarts.init()` on a hidden container. + Use lazy init — initialise charts only on first tab visibility. + Pattern: `const _rendered = new Set(['overview']); function showPage(name) { if (!_rendered.has(name)) { _rendered.add(name); initChartsFor(name); } }` +- Layout composition: + - Hero KPI cards at the top (large numbers, colour-coded ±, delta arrows). + - Main narrative chart immediately below KPIs. + - Supporting charts below, each with a subtitle explaining what it reveals. + - Use ECharts `markLine` for thresholds, `markPoint` for outliers, + `markArea` for highlighted regions. +- Responsive: + - `` + - Multi-card grid: `grid-template-columns: repeat(auto-fit, minmax(360px, 1fr))` + - Chart containers: `width: 100%; height: min(420px, 60vh)` + - Register `window.addEventListener('resize', () => chart.resize())` on every ECharts instance. + - Tables wrapped in `
` — never fixed widths. +- SECURITY: NEVER embed API keys, tokens, passwords, or connection strings in HTML/JS. + Credentials were already used server-side; serialise only the resulting data. + +HARD OUTPUT CONTRACT (a static verifier checks each of these; a violation +fails the step and costs a regeneration): +- Emit a complete HTML document with an explicit ``...``. +- Include ``. +- NEVER put an absolute URL in a `fetch()` call. Use relative paths only. +- NEVER put an absolute URL in a resource reference either — no `` for fonts or CSS, no ``. The ONLY + allowed absolute URL in the whole document is a ` + +
+ +""" + + +def test_good_fullstack_frontend_passes(): + r = verify_frontend(GOOD, is_fullstack=True) + assert r.ok, r.errors + + +def test_missing_viewport_is_error(): + html = GOOD.replace('', "") + r = verify_frontend(html, is_fullstack=True) + assert not r.ok + assert any("viewport" in e for e in r.errors) + + +def test_missing_api_base_is_error_for_fullstack(): + html = GOOD.replace('', "") + r = verify_frontend(html, is_fullstack=True) + assert not r.ok + assert any("api-base" in e for e in r.errors) + + +def test_absolute_fetch_url_is_error(): + html = GOOD.replace("fetch(api('/api/items'))", "fetch('http://localhost:8000/api/items')") + r = verify_frontend(html, is_fullstack=True) + assert not r.ok + assert any("absolute" in e.lower() for e in r.errors) + + +def test_bare_path_backend_call_is_error_for_fullstack(): + html = GOOD.replace("fetch(api('/api/items'))", "fetch('/items')") + r = verify_frontend(html, is_fullstack=True) + assert not r.ok + assert any("/api/" in e for e in r.errors) + + +def test_missing_body_is_error(): + r = verify_frontend("
no body
", is_fullstack=False) + assert not r.ok + assert any("body" in e.lower() for e in r.errors) + + +def test_forbidden_globals_are_errors(): + html = GOOD.replace("", "window.__antonCommentsLayer = 1;") + r = verify_frontend(html, is_fullstack=True) + assert not r.ok + assert any("__antonCommentsLayer" in e for e in r.errors) + + +def test_missing_ids_is_only_a_warning(): + html = GOOD.replace('
', "
") + r = verify_frontend(html, is_fullstack=True) + assert r.ok # warning, not error + assert r.warnings + + +def test_html_app_does_not_require_api_base(): + html = GOOD.replace('', "").replace("fetch(api('/api/items'));", "") + r = verify_frontend(html, is_fullstack=False) + assert r.ok, r.errors diff --git a/tests/test_builtin_skills.py b/tests/test_builtin_skills.py index d7250a07..16aa978e 100644 --- a/tests/test_builtin_skills.py +++ b/tests/test_builtin_skills.py @@ -230,3 +230,176 @@ async def test_summary_quoting_marker_does_not_suppress_resend(self, store): session = self._session(store, history) result = await handle_recall_skill(session, {"label": "build-html-dashboard"}) assert "## Procedure" in result + + +class TestArtifactWorkflowPointsAtTheTool: + """The always-on section must name the normal path. + + Before this change it described only manual assembly, even though + `generate_artifact` is registered alongside the other artifact tools whenever a + workspace is bound to the session (`session.py:940-947`), and its own + `ToolDef.prompt` demands using it INSTEAD of writing files by hand. The agent + received two mutually exclusive instructions. + """ + + def test_workflow_names_the_generator_for_supported_types(self): + from anton.core.llm.prompts import ARTIFACTS_PROMPT + + assert "generate_artifact" in ARTIFACTS_PROMPT + for t in ("html-app", "fullstack-stateless-app", "fullstack-stateful-app"): + assert t in ARTIFACTS_PROMPT + + def test_workflow_keeps_the_manual_path_for_the_other_types(self): + """The generator does not support document/dataset/image — their path stays manual.""" + from anton.core.llm.prompts import ARTIFACTS_PROMPT + + low = ARTIFACTS_PROMPT.lower() + assert "document" in low and "dataset" in low + assert "yourself" in low or "by hand" in low + + def test_workflow_still_requires_registration_first(self): + """create_artifact is still the first step — the generator takes a ready slug.""" + from anton.core.llm.prompts import ARTIFACTS_PROMPT + + assert "create_artifact" in ARTIFACTS_PROMPT + assert "BEFORE" in ARTIFACTS_PROMPT + + +class TestRecallIsConditionalNow: + """`recall_skill` stays, but stops being an unconditional instruction.""" + + def _texts(self): + from anton.core.llm.prompts import ( + BACKEND_GENERATION_PROMPT, + VISUALIZATIONS_HTML_OUTPUT_FORMAT_PROMPT, + VISUALIZATIONS_MARKDOWN_OUTPUT_FORMAT_PROMPT, + ) + + return { + "backend": BACKEND_GENERATION_PROMPT, + "viz_html": VISUALIZATIONS_HTML_OUTPUT_FORMAT_PROMPT, + "viz_markdown": VISUALIZATIONS_MARKDOWN_OUTPUT_FORMAT_PROMPT, + } + + def test_no_unconditional_mandatory_recall(self): + """"MANDATORY: call recall_skill …" is the very instruction that + conflicted with generate_artifact's ToolDef.prompt.""" + for name, text in self._texts().items(): + assert "MANDATORY: call `recall_skill" not in text, name + assert "MANDATORY: BEFORE writing" not in text, name + + def test_the_generator_is_named_as_the_normal_path(self): + for name, text in self._texts().items(): + assert "generate_artifact" in text, name + + def test_recall_is_tied_to_the_manual_path(self): + """The skill is not discarded — manual edits and unsupported types need it.""" + for name, text in self._texts().items(): + assert "recall_skill" in text, name + low = text.lower() + assert "by hand" in low or "yourself" in low, name + + def test_hints_still_point_at_shipped_skills(self, store): + """The rewording must not leave a pointer to a non-existent skill.""" + import re + + for text in self._texts().values(): + for label in re.findall(r'recall_skill\(\\?"([a-z0-9-]+)\\?"\)', text): + assert store.load(label) is not None, label + + +class TestToolDescriptionsAfterTheSwitch: + def test_create_artifact_no_longer_demands_recall_before_any_write(self): + from anton.core.tools.tool_defs import CREATE_ARTIFACT_TOOL + + d = CREATE_ARTIFACT_TOOL.description + assert "SKILL PREREQUISITE" not in d + assert "BEFORE writing any files" not in d + assert "generate_artifact" in d + + def test_create_artifact_keeps_a_conditional_recall_hint(self): + """The manual path does use the skill, and + test_tool_descriptions_reference_shipped_skills depends on this mention.""" + from anton.core.tools.tool_defs import CREATE_ARTIFACT_TOOL + + d = CREATE_ARTIFACT_TOOL.description + assert "recall_skill" in d + assert "by hand" in d.lower() or "yourself" in d.lower() + + def test_launch_backend_says_the_generator_launches_on_its_own(self): + from anton.core.tools.tool_defs import LAUNCH_BACKEND_TOOL + + d = LAUNCH_BACKEND_TOOL.description + assert "generate_artifact" in d + assert "If you haven't recalled it this conversation" not in d + + +class TestSkillDescriptionsSteerAway: + """The description is all the thalamus sees when deciding to preload a skill. + + `_inject_recalled_skills` puts the FULL body of up to three skills into the + context per turn — i.e. ~270 lines of "do it by hand" land in front of the + agent before it picks a path. So the description must open with the + applicability condition, not with "MANDATORY reading before ANY…". + """ + + LABELS = ("build-html-dashboard", "build-fullstack-backend") + + def test_no_mandatory_trigger_words(self, store): + for label in self.LABELS: + d = store.load(label).description + assert "MANDATORY" not in d, label + assert "before writing" not in d.lower(), label + assert "when in doubt" not in d.lower(), label + + def test_description_opens_with_the_applicability_condition(self, store): + """The condition must come FIRST, not somewhere inside the first sentence. + + A weak assert ("by hand" anywhere in the first sentence) also passes for a + topic-led wording: "For building a dashboard, chart or browser + visualization BY HAND …" — there the first ~56 characters are pure bait for + matching, while the catalog's framing text reads "When the user's request + matches one of them, call recall_skill(label)". So we pin the prefix. + """ + for label in self.LABELS: + d = store.load(label).description + assert d.startswith("ONLY "), f"{label}: {d[:50]!r}" + assert "by hand" in d[:55].lower(), f"{label}: {d[:55]!r}" + + def test_normal_path_disclaimer_precedes_the_contents(self, store): + """"NOT needed on the normal path" must come ABOVE the contents listing. + + Otherwise the description advertises the contract first and hides the + disclaimer in the tail. + """ + markers = { + "build-html-dashboard": "ECharts", + "build-fullstack-backend": "FastAPI", + } + for label, content_marker in markers.items(): + d = store.load(label).description + assert d.index("NOT needed") < d.index(content_marker), label + + def test_description_points_at_the_generator(self, store): + for label in self.LABELS: + assert "generate_artifact" in store.load(label).description, label + + def test_description_stays_findable_for_the_fallback(self, store): + """Steering away is not hiding: the contents must stay recognisable.""" + html = store.load("build-html-dashboard").description + assert "dashboard" in html.lower() and "chart" in html.lower() + backend = store.load("build-fullstack-backend").description + assert "backend" in backend.lower() and "fastapi" in backend.lower() + + def test_descriptions_still_fit_the_budget(self, store): + for label in self.LABELS: + assert len(store.load(label).description) <= 1024, label + + def test_skill_bodies_are_untouched(self, store): + """Design decision 1: bodies are not rewritten, the manual path still needs them.""" + html = store.load("build-html-dashboard").declarative_md + assert "REROUND DISCIPLINE" in html + assert "cdn.jsdelivr.net/npm/echarts" in html + backend = store.load("build-fullstack-backend").declarative_md + assert "handler = Mangum(app, lifespan=\"off\")" in backend + assert "DS____" in backend diff --git a/tests/test_thalamus.py b/tests/test_thalamus.py index 0c03025d..44ffe4f1 100644 --- a/tests/test_thalamus.py +++ b/tests/test_thalamus.py @@ -182,6 +182,29 @@ def _mock_skill_store(): return store +def _multi_skill_store(n: int): + """A store with n distinct skills: skill-0 … skill-(n-1). + + `_mock_skill_store` knows only one skill, and the injection cap is only + visible with more than three labels. + """ + skills = { + f"skill-{i}": SimpleNamespace( + label=f"skill-{i}", + name=f"Skill {i}", + description=f"desc {i}", + declarative_md=f"BODY {i}", + ) + for i in range(n) + } + store = MagicMock() + store.list_summaries.return_value = [ + {"label": k, "description": v.description} for k, v in skills.items() + ] + store.load.side_effect = lambda label: skills.get(label) + return store + + class TestSessionThalamus: async def test_thalamus_off_by_default(self): llm = make_mock_llm() @@ -320,6 +343,41 @@ async def test_preload_skips_skill_already_in_history(self): session._inject_recalled_skills(["csv-summary"]) assert len(session.history) == 2 # second preload is a no-op + async def test_preload_caps_at_three_skills(self): + # The cap bounds how much "do it by hand" reaches the context in a + # single turn: five named labels must yield three injected bodies, or one + # turn could inject the whole skill catalog. + llm = make_mock_llm() + session = ChatSession(ChatSessionConfig(llm_client=llm, router_enabled=True)) + session._skill_store = _multi_skill_store(5) + + session._inject_recalled_skills([f"skill-{i}" for i in range(5)]) + + # One injection appends EXACTLY two history entries (assistant with + # tool_use + user with tool_result) regardless of how many skills — so the + # blocks are what must be counted. + assert len(session.history) == 2 + tool_uses = [ + b + for msg in session.history + if isinstance(msg.get("content"), list) + for b in msg["content"] + if isinstance(b, dict) and b.get("type") == "tool_use" + ] + assert len(tool_uses) == 3 + present = [i for i in range(5) if f"BODY {i}" in str(session.history)] + assert present == [0, 1, 2] + + async def test_preload_dedupes_repeated_labels(self): + # The same label named three times must not inject the body three times. + llm = make_mock_llm() + session = ChatSession(ChatSessionConfig(llm_client=llm, router_enabled=True)) + session._skill_store = _multi_skill_store(1) + + session._inject_recalled_skills(["skill-0", "skill-0", "skill-0"]) + + assert str(session.history).count("BODY 0") == 1 + async def test_gate_usage_counted_on_delegate_streaming(self): # The gate hits every text turn; on delegate its usage must reach the # consumer as its own StreamComplete (like a planning round), or token diff --git a/tests/test_tool_handler_generate_artifact.py b/tests/test_tool_handler_generate_artifact.py new file mode 100644 index 00000000..9e84c7e2 --- /dev/null +++ b/tests/test_tool_handler_generate_artifact.py @@ -0,0 +1,111 @@ +"""handle_generate_artifact: FSM failures must come back wrapped with an +instruction to report to the user (never DIY the artifact); input-validation +errors stay unwrapped so the agent fixes its call instead.""" +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import anton.core.tools.generate_artifact as gen_pkg +from anton.core.artifacts import ArtifactStore +from anton.core.tools.tool_handlers import handle_generate_artifact + + +def _session(tmp_path: Path): + return SimpleNamespace( + _workspace=SimpleNamespace(artifacts_dir=tmp_path / "artifacts") + ) + + +def _make_artifact(tmp_path: Path) -> str: + store = ArtifactStore(tmp_path / "artifacts") + return store.create( + name="Clock", description="d", type="fullstack-stateless-app" + ).slug + + +async def test_fsm_failure_is_wrapped_with_report_instruction(tmp_path: Path, monkeypatch): + slug = _make_artifact(tmp_path) + + async def fake_generate(**kw): + return "Backend verification failed after retry: boom" + + monkeypatch.setattr(gen_pkg, "generate", fake_generate) + out = await handle_generate_artifact( + _session(tmp_path), {"slug": slug, "context": "brief"} + ) + assert "artifact generation failed" in out + assert "Backend verification failed after retry: boom" in out + assert "do NOT build or repair the artifact yourself" in out + assert "Report this failure to the user" in out + + +async def test_generator_crash_is_wrapped(tmp_path: Path, monkeypatch): + slug = _make_artifact(tmp_path) + + async def fake_generate(**kw): + raise RuntimeError("kaput") + + monkeypatch.setattr(gen_pkg, "generate", fake_generate) + out = await handle_generate_artifact( + _session(tmp_path), {"slug": slug, "context": "brief"} + ) + assert "artifact generation failed" in out + assert "kaput" in out + + +async def test_input_validation_errors_are_not_wrapped(tmp_path: Path): + out = await handle_generate_artifact( + _session(tmp_path), {"slug": "nope", "context": "b"} + ) + assert out.startswith("Error: no artifact found") + assert "generation failed" not in out + + +async def test_success_returns_json_unchanged(tmp_path: Path, monkeypatch): + slug = _make_artifact(tmp_path) + + async def fake_generate(**kw): + return {"files_written": ["backend.py"], "summary": "ok", "trace": []} + + monkeypatch.setattr(gen_pkg, "generate", fake_generate) + out = await handle_generate_artifact( + _session(tmp_path), {"slug": slug, "context": "brief"} + ) + assert '"files_written"' in out + assert "generation failed" not in out + + +async def test_handler_forwards_primary_to_generate(monkeypatch, tmp_path): + """The primary from metadata must reach the generator.""" + import anton.core.tools.tool_handlers as th + + captured = {} + + async def fake_generate(**kw): + captured.update(kw) + return {"files_written": ["report.html"], "summary": "s", "trace": []} + + monkeypatch.setattr( + "anton.core.tools.generate_artifact.generate", fake_generate, raising=False + ) + + class _Artifact: + type = "html-app" + slug = "a" + primary = "report.html" + + class _Store: + def open(self, slug): + return _Artifact() + + def folder_for(self, slug): + return tmp_path + + monkeypatch.setattr(th, "_artifact_store", lambda session: _Store()) + + out = await th.handle_generate_artifact( + object(), {"slug": "a", "context": "## User request\nx"} + ) + assert "report.html" in out + assert captured["primary"] == "report.html"