test(agent): remove per-capability test boilerplate

Adding a capability meant editing five duplicated _FakeService classes
and an exact 30-tuple registry assertion. similarity_search shipped
broken this way and was repaired five days later.

The stubs now live in tests/fakes.py, where an unknown capability
resolves through __getattr__, so a new capability needs no test edit.
Registry-derived parity tests replace the ordered tuple and turn a
missing tool contract, missing parity route, or mismatched
response_model into one named failure instead of a cascade.

Because __getattr__ also hides a typo'd handler from every stub-backed
test, one check builds the registry against the real service.

tests/ goes on sys.path via pythonpath instead of pytest's implicit
conftest insertion, which breaks under --import-mode=importlib, if
tests/__init__.py is added, or once a nested conftest.py shadows the
name.

Also repairs tests/agent/test_session_store_light_metadata.py, which
was silently uncollectable: it imported tests.agent.test_hosted_runtime,
and tests/agent is not an importable package.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

pyproject.toml

@@ -92,6 +92,14 @@ packages = {find = {where = ["src"], namespaces = false}}
 [tool.setuptools.package-data]
 "*" = ["*.json"]
 
+[tool.pytest.ini_options]
+# Put `tests/` on sys.path so shared test doubles live in an ordinary module
+# (`tests/fakes.py`) that every test package can import. Relying on pytest's
+# implicit rootdir insertion instead would break under
+# `--import-mode=importlib`, if `tests/__init__.py` were added, or as soon as a
+# nested `conftest.py` shadowed the name.
+pythonpath = ["tests"]
+
 [tool.ruff]
 line-length = 119
 

tests/agent/test_capability_parity.py

@@ -0,0 +1,177 @@
+"""Parity checks between the capability registry and its downstream surfaces.
+
+`src/TerraFin/agent/runtime/capability.py` is the single source of truth for the
+agent capability surface. Three things must stay in step with it, and each one
+otherwise fails only at runtime (or silently, in generated docs):
+
+* every capability needs a `HOSTED_TOOL_CONTRACTS` entry, because the tool
+  adapter looks one up for each registered capability when listing tools
+* every capability that declares an `http_route_path` needs that route to exist,
+  since HTTP-only agents depend on the parity route
+* declared metadata (`summary`, `response_model_name`) must be populated and the
+  response model must actually exist somewhere in the package
+
+These tests are deliberately derived from the registry rather than hardcoded, so
+adding a capability requires no edit here.
+"""
+
+import ast
+import pathlib
+
+import pytest
+from fakes import BaseFakeService, fake_chart_opener
+
+from TerraFin.agent.contracts.tool_contracts import HOSTED_TOOL_CONTRACTS
+from TerraFin.agent.runtime import build_default_capability_registry
+
+
+# Hosted-only tools live outside the capability registry: the three guru
+# consults and `current_view_context` need a live session, so they are wired
+# straight into the tool adapter instead.
+HOSTED_ONLY_TOOLS = frozenset(
+    {
+        "consult_warren_buffett",
+        "consult_howard_marks",
+        "consult_stanley_druckenmiller",
+        "current_view_context",
+    }
+)
+
+# `open_chart` mutates hosted session state and has no stateless HTTP surface.
+CAPABILITIES_WITHOUT_ROUTE = frozenset({"open_chart"})
+
+# `open_chart` is handed the injected `chart_opener` callable rather than a
+# `TerraFinAgentService` method, so its handler name never matches the
+# capability name.
+CAPABILITIES_WITH_INJECTED_HANDLER = frozenset({"open_chart"})
+
+_SRC_ROOT = pathlib.Path(__file__).resolve().parents[2] / "src" / "TerraFin"
+
+
+@pytest.fixture(scope="module")
+def capabilities():
+    registry = build_default_capability_registry(BaseFakeService(), chart_opener=fake_chart_opener)
+    return registry.list()
+
+
+@pytest.fixture(scope="module")
+def live_route_paths() -> set[str | None]:
+    """Paths served by the assembled app.
+
+    Module-scoped because `create_app()` is not side-effect free: it loads the
+    repo `.env` into `os.environ`, resets chart/calendar module state, and
+    builds process-wide singletons. Build it once here rather than per test.
+    """
+
+    from TerraFin.interface.server import create_app
+
+    return {getattr(route, "path", None) for route in create_app().routes}
+
+
+def _declared_class_names() -> set[str]:
+    """Collect every class name defined under src/TerraFin without importing it."""
+
+    names: set[str] = set()
+    for path in _SRC_ROOT.rglob("*.py"):
+        try:
+            tree = ast.parse(path.read_text(encoding="utf-8"))
+        except (SyntaxError, UnicodeDecodeError):  # pragma: no cover - defensive
+            continue
+        names.update(node.name for node in ast.walk(tree) if isinstance(node, ast.ClassDef))
+    return names
+
+
+def test_every_capability_has_a_tool_contract(capabilities) -> None:
+    missing = sorted(c.name for c in capabilities if c.name not in HOSTED_TOOL_CONTRACTS)
+    assert not missing, (
+        "capabilities registered without a HOSTED_TOOL_CONTRACTS entry "
+        f"(tool listing would raise KeyError): {missing}"
+    )
+
+
+def test_every_tool_contract_is_registered_or_hosted_only(capabilities) -> None:
+    registered = {c.name for c in capabilities}
+    orphans = sorted(set(HOSTED_TOOL_CONTRACTS) - registered - HOSTED_ONLY_TOOLS)
+    assert not orphans, f"tool contracts with no registered capability and no hosted-only exemption: {orphans}"
+
+
+def test_every_capability_declares_generator_metadata(capabilities) -> None:
+    missing_summary = sorted(c.name for c in capabilities if not c.summary)
+    assert not missing_summary, f"capabilities missing `summary`: {missing_summary}"
+
+    missing_model = sorted(c.name for c in capabilities if not c.response_model_name)
+    assert not missing_model, f"capabilities missing `response_model_name`: {missing_model}"
+
+    missing_route_path = sorted(
+        c.name for c in capabilities if not c.http_route_path and c.name not in CAPABILITIES_WITHOUT_ROUTE
+    )
+    assert not missing_route_path, f"capabilities missing `http_route_path`: {missing_route_path}"
+
+
+def test_declared_response_models_exist(capabilities) -> None:
+    """Weak existence check: the name matches *some* class under src/TerraFin.
+
+    Response models are spread across agent contracts, data contracts, private
+    provider models, and page route modules, so this only catches an outright
+    typo. `test_response_model_names_match_tool_contracts` is the strict check.
+    """
+
+    declared = _declared_class_names()
+    unresolved = sorted(
+        f"{c.name} -> {c.response_model_name}"
+        for c in capabilities
+        if c.response_model_name and c.response_model_name not in declared
+    )
+    assert not unresolved, f"`response_model_name` values with no matching class under src/TerraFin: {unresolved}"
+
+
+def test_response_model_names_match_tool_contracts(capabilities) -> None:
+    """The registry and the tool contract declare the same response model.
+
+    `HOSTED_TOOL_CONTRACTS[name]["response_model"]` is shipped to the model as
+    tool metadata, so a disagreement between the two declarations sends the LLM
+    a response shape that does not match the route's.
+    """
+
+    mismatches = sorted(
+        f"{c.name}: registry={c.response_model_name!r} contract={HOSTED_TOOL_CONTRACTS[c.name].get('response_model')!r}"
+        for c in capabilities
+        if c.name in HOSTED_TOOL_CONTRACTS
+        and HOSTED_TOOL_CONTRACTS[c.name].get("response_model") != c.response_model_name
+    )
+    assert not mismatches, f"registry / tool-contract response_model disagreement: {mismatches}"
+
+
+def test_every_capability_handler_binds_on_the_real_service() -> None:
+    """Guard the one thing the shared stub cannot catch.
+
+    `BaseFakeService.__getattr__` resolves any attribute, so a capability whose
+    handler names a method the real `TerraFinAgentService` does not implement
+    (a typo, or a handler added before its service method) passes every
+    stub-backed test. Building the registry against the real service is the
+    check that fails loudly, and it is cheap: no network, no env mutation.
+    """
+
+    from TerraFin.agent.service import TerraFinAgentService
+
+    registry = build_default_capability_registry(TerraFinAgentService(), chart_opener=fake_chart_opener)
+
+    mislabelled = sorted(
+        f"{c.name} -> {getattr(c.handler, '__name__', repr(c.handler))}"
+        for c in registry.list()
+        if c.name not in CAPABILITIES_WITH_INJECTED_HANDLER
+        and getattr(c.handler, "__name__", None) not in (c.name, None)
+    )
+    assert not mislabelled, (
+        "capability handlers bound to a differently-named service method "
+        f"(likely a copy-paste error): {mislabelled}"
+    )
+
+
+def test_declared_http_routes_exist(capabilities, live_route_paths) -> None:
+    missing = sorted(
+        f"{c.name} -> {c.http_route_path}"
+        for c in capabilities
+        if c.http_route_path and c.http_route_path not in live_route_paths
+    )
+    assert not missing, f"capabilities whose declared http_route_path has no live route: {missing}"

tests/agent/test_hosted_runtime.py

@@ -2,6 +2,8 @@ import time
 from pathlib import Path
 
 import pytest
+from fakes import BaseFakeService as _FakeService
+from fakes import fake_chart_opener as _fake_chart_opener
 
 from TerraFin.agent.definitions import (
     DEFAULT_HOSTED_AGENT_NAME,
@@ -21,200 +23,6 @@ from TerraFin.agent.session_store import SQLiteHostedSessionStore
 from TerraFin.agent.transcript_store import HostedTranscriptStore
 
 
-def _processing() -> dict[str, object]:
-    return {
-        "requestedDepth": "auto",
-        "resolvedDepth": "full",
-        "loadedStart": "2024-01-01",
-        "loadedEnd": "2024-12-31",
-        "isComplete": True,
-        "hasOlder": False,
-        "sourceVersion": "test-source",
-        "view": "daily",
-    }
-
-
-class _FakeService:
-    def resolve(self, query: str) -> dict[str, object]:
-        return {"type": "stock", "name": query.upper(), "path": f"/stock/{query.upper()}", "processing": _processing()}
-
-    def market_data(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"ticker": name, "seriesType": "candlestick", "count": 1, "data": [], "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def indicators(
-        self,
-        name: str,
-        indicators: str,
-        *,
-        depth: str = "auto",
-        view: str = "daily",
-    ) -> dict[str, object]:
-        return {
-            "ticker": name,
-            "indicators": {"rsi": {"name": "rsi", "offset": 0, "values": {"value": 55.0}}},
-            "unknown": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view, "indicatorQuery": indicators},
-        }
-
-    def patterns(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"ticker": name, "signals": [], "total": 0, "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def market_snapshot(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {
-            "ticker": name,
-            "price_action": {"current": 100.0},
-            "indicators": {"rsi": 55.0},
-            "market_breadth": [],
-            "watchlist": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view},
-        }
-
-    def lppl_analysis(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"name": name, "confidence": 0.2, "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def company_info(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "shortName": f"{ticker} Corp", "processing": _processing()}
-
-    def earnings(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "earnings": [], "processing": _processing()}
-
-    def financials(self, ticker: str, *, statement: str = "income", period: str = "annual") -> dict[str, object]:
-        return {"ticker": ticker, "statement": statement, "period": period, "columns": [], "rows": [], "processing": _processing()}
-
-    def portfolio(self, guru: str) -> dict[str, object]:
-        return {"guru": guru, "info": {}, "holdings": [], "count": 0, "processing": _processing()}
-
-    def economic(self, indicators: str) -> dict[str, object]:
-        return {"indicators": {indicators: {"latest_value": 3.0}}, "processing": _processing()}
-
-    def macro_focus(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {
-            "name": name,
-            "info": {"name": name, "type": "index", "description": "Macro", "currentValue": 1.0, "change": 0.0, "changePercent": 0.0},
-            "seriesType": "line",
-            "count": 1,
-            "data": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view},
-        }
-
-    def calendar_events(
-        self,
-        *,
-        year: int,
-        month: int,
-        categories: str | None = None,
-        limit: int | None = None,
-    ) -> dict[str, object]:
-        return {"events": [], "count": 0, "month": month, "year": year, "categories": categories, "limit": limit, "processing": _processing()}
-
-    def fundamental_screen(self, ticker: str) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "moat": {"score": "wide"},
-            "earnings_quality": {},
-            "balance_sheet": {},
-            "capital_allocation": {},
-            "pricing_power": {},
-            "warnings": [],
-            "processing": _processing(),
-        }
-
-    def risk_profile(self, name: str, *, depth: str = "auto") -> dict[str, object]:
-        return {
-            "ticker": name,
-            "tail_risk": {},
-            "convexity": {},
-            "volatility": {"requestedDepth": depth},
-            "drawdown": {},
-            "warnings": [],
-            "processing": _processing(),
-        }
-
-    def valuation(self, ticker: str) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "dcf": {"status": "ready", "intrinsic_value": 120.0},
-            "reverse_dcf": {"status": "ready", "implied_growth_pct": 8.0},
-            "relative": {"trailing_pe": 22.0},
-            "graham_number": 100.0,
-            "margin_of_safety_pct": 12.0,
-            "current_price": 107.0,
-            "processing": _processing(),
-        }
-
-    def sec_filings(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "cik": 1, "forms": [], "filings": [], "processing": _processing()}
-
-    def sec_filing_document(
-        self, ticker: str, accession: str, primaryDocument: str, *, form: str = "10-Q"
-    ) -> dict[str, object]:
-        return {"ticker": ticker, "accession": accession, "primaryDocument": primaryDocument, "toc": [], "charCount": 0, "indexUrl": "", "documentUrl": "", "processing": _processing()}
-
-    def sec_filing_section(
-        self, ticker: str, accession: str, primaryDocument: str, sectionSlug: str, *, form: str = "10-Q"
-    ) -> dict[str, object]:
-        return {"ticker": ticker, "accession": accession, "sectionSlug": sectionSlug, "sectionTitle": "stub", "markdown": "", "charCount": 0, "documentUrl": "", "processing": _processing()}
-
-    def fcf_history(self, ticker: str, years: int = 10) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "years": years,
-            "rows": [],
-            "candidates": {"threeYearAvg": None, "latestAnnual": None, "ttm": None},
-            "autoSelectedSource": "annual",
-            "processing": _processing(),
-        }
-
-    def similarity_search(
-        self,
-        ticker: str,
-        universe: str = "sp500+nasdaq100+kospi200",
-        period: str = "1y",
-        top_n: int = 20,
-    ) -> dict[str, object]:
-        return {"ticker": ticker, "period": period, "pool": {}, "results": [], "count": 0, "processing": _processing()}
-
-    def fear_greed(self) -> dict[str, object]:
-        return {"score": 50, "rating": "Neutral", "processing": _processing()}
-
-    def sp500_dcf(self) -> dict[str, object]:
-        return {"status": "ready", "currentIntrinsicValue": 5000.0, "processing": _processing()}
-
-    def beta_estimate(self, ticker: str) -> dict[str, object]:
-        return {"symbol": ticker, "beta": 1.0, "adjustedBeta": 1.0, "rSquared": 0.5, "processing": _processing()}
-
-    def top_companies(self) -> dict[str, object]:
-        return {"companies": [], "count": 0, "processing": _processing()}
-
-    def market_regime(self) -> dict[str, object]:
-        return {"summary": "stub", "confidence": "low", "signals": [], "processing": _processing()}
-
-    def trailing_forward_pe(self) -> dict[str, object]:
-        return {"date": "2026-04-01", "latestValue": 0.0, "history": [], "processing": _processing()}
-
-    def market_breadth(self) -> dict[str, object]:
-        return {"metrics": [], "processing": _processing()}
-
-    def watchlist(self) -> dict[str, object]:
-        return {"items": [], "count": 0, "processing": _processing()}
-
-
-def _fake_chart_opener(
-    data_or_names,
-    *,
-    session_id: str | None = None,
-    **kwargs,
-) -> dict[str, object]:
-    _ = kwargs
-    return {
-        "ok": True,
-        "sessionId": session_id or "agent:chart",
-        "chartUrl": f"http://127.0.0.1:8001/chart?sessionId={session_id or 'agent:chart'}",
-        "processing": _processing(),
-        "inputEcho": data_or_names,
-    }
-
-
 def _runtime(
     agent_registry: TerraFinAgentDefinitionRegistry | None = None,
     *,

tests/agent/test_loop.py

@@ -1,12 +1,14 @@
 import json
 
 import pytest
+from fakes import BaseFakeService as _FakeService
+from fakes import fake_chart_opener as _fake_chart_opener
 
+from TerraFin.agent.conversation import is_internal_only_message
 from TerraFin.agent.definitions import (
     DEFAULT_HOSTED_AGENT_NAME,
     build_default_agent_definition_registry,
 )
-from TerraFin.agent.conversation import is_internal_only_message
 from TerraFin.agent.guru import (
     GuruResearchMemo,
     GuruRoutePlan,
@@ -29,193 +31,12 @@ from TerraFin.agent.session_store import SQLiteHostedSessionStore
 from TerraFin.agent.transcript_store import HostedTranscriptStore
 
 
-def _processing() -> dict[str, object]:
-    return {
-        "requestedDepth": "auto",
-        "resolvedDepth": "full",
-        "loadedStart": "2024-01-01",
-        "loadedEnd": "2024-12-31",
-        "isComplete": True,
-        "hasOlder": False,
-        "sourceVersion": "test-source",
-        "view": "daily",
-    }
-
-
 def _public_roles(messages):
     return [message.role for message in messages if not is_internal_only_message(message)]
 
 
-class _FakeService:
-    def resolve(self, query: str) -> dict[str, object]:
-        return {"type": "stock", "name": query.upper(), "path": f"/stock/{query.upper()}", "processing": _processing()}
-
-    def market_data(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"ticker": name, "seriesType": "candlestick", "count": 1, "data": [], "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def patterns(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"ticker": name, "signals": [], "total": 0, "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def fcf_history(self, ticker: str, years: int = 10) -> dict[str, object]:
-        return {
-            "ticker": ticker, "years": years, "rows": [],
-            "candidates": {"threeYearAvg": None, "latestAnnual": None, "ttm": None},
-            "autoSelectedSource": "annual", "processing": _processing(),
-        }
-
-    def similarity_search(self, ticker: str, universe: str = "sp500+nasdaq100+kospi200", period: str = "1y", top_n: int = 20) -> dict[str, object]:
-        return {"ticker": ticker, "period": period, "pool": {}, "results": [], "count": 0, "processing": _processing()}
-
-    def indicators(
-        self,
-        name: str,
-        indicators: str,
-        *,
-        depth: str = "auto",
-        view: str = "daily",
-    ) -> dict[str, object]:
-        return {
-            "ticker": name,
-            "indicators": {"rsi": {"name": "rsi", "offset": 0, "values": {"value": 55.0}}},
-            "unknown": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view, "indicatorQuery": indicators},
-        }
-
-    def market_snapshot(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {
-            "ticker": name,
-            "price_action": {"current": 100.0},
-            "indicators": {"rsi": 55.0},
-            "market_breadth": [],
-            "watchlist": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view},
-        }
-
-    def lppl_analysis(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"name": name, "confidence": 0.2, "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def company_info(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "shortName": f"{ticker} Corp", "processing": _processing()}
-
-    def earnings(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "earnings": [], "processing": _processing()}
-
-    def financials(self, ticker: str, *, statement: str = "income", period: str = "annual") -> dict[str, object]:
-        return {"ticker": ticker, "statement": statement, "period": period, "columns": [], "rows": [], "processing": _processing()}
-
-    def portfolio(self, guru: str) -> dict[str, object]:
-        return {"guru": guru, "info": {}, "holdings": [], "count": 0, "processing": _processing()}
-
-    def economic(self, indicators: str) -> dict[str, object]:
-        return {"indicators": {indicators: {"latest_value": 3.0}}, "processing": _processing()}
 
-    def macro_focus(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {
-            "name": name,
-            "info": {"name": name, "type": "index", "description": "Macro", "currentValue": 1.0, "change": 0.0, "changePercent": 0.0},
-            "seriesType": "line",
-            "count": 1,
-            "data": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view},
-        }
-
-    def calendar_events(
-        self,
-        *,
-        year: int,
-        month: int,
-        categories: str | None = None,
-        limit: int | None = None,
-    ) -> dict[str, object]:
-        return {"events": [], "count": 0, "month": month, "year": year, "categories": categories, "limit": limit, "processing": _processing()}
-
-    def fundamental_screen(self, ticker: str) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "moat": {"score": "wide"},
-            "earnings_quality": {},
-            "balance_sheet": {},
-            "capital_allocation": {},
-            "pricing_power": {},
-            "warnings": [],
-            "processing": _processing(),
-        }
-
-    def risk_profile(self, name: str, *, depth: str = "auto") -> dict[str, object]:
-        return {
-            "ticker": name,
-            "tail_risk": {},
-            "convexity": {},
-            "volatility": {"requestedDepth": depth},
-            "drawdown": {},
-            "warnings": [],
-            "processing": _processing(),
-        }
-
-    def valuation(self, ticker: str) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "dcf": {"status": "ready", "intrinsic_value": 120.0},
-            "reverse_dcf": {"status": "ready", "implied_growth_pct": 8.0},
-            "relative": {"trailing_pe": 22.0},
-            "graham_number": 100.0,
-            "margin_of_safety_pct": 12.0,
-            "current_price": 107.0,
-            "processing": _processing(),
-        }
-
-    def sec_filings(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "cik": 1, "forms": [], "filings": [], "processing": _processing()}
-
-    def sec_filing_document(
-        self, ticker: str, accession: str, primaryDocument: str, *, form: str = "10-Q"
-    ) -> dict[str, object]:
-        return {"ticker": ticker, "accession": accession, "primaryDocument": primaryDocument, "toc": [], "charCount": 0, "indexUrl": "", "documentUrl": "", "processing": _processing()}
-
-    def sec_filing_section(
-        self, ticker: str, accession: str, primaryDocument: str, sectionSlug: str, *, form: str = "10-Q"
-    ) -> dict[str, object]:
-        return {"ticker": ticker, "accession": accession, "sectionSlug": sectionSlug, "sectionTitle": "stub", "markdown": "", "charCount": 0, "documentUrl": "", "processing": _processing()}
 
-    def fear_greed(self) -> dict[str, object]:
-        return {"score": 50, "rating": "Neutral", "processing": _processing()}
-
-    def sp500_dcf(self) -> dict[str, object]:
-        return {"status": "ready", "currentIntrinsicValue": 5000.0, "processing": _processing()}
-
-    def beta_estimate(self, ticker: str) -> dict[str, object]:
-        return {"symbol": ticker, "beta": 1.0, "adjustedBeta": 1.0, "rSquared": 0.5, "processing": _processing()}
-
-    def top_companies(self) -> dict[str, object]:
-        return {"companies": [], "count": 0, "processing": _processing()}
-
-    def market_regime(self) -> dict[str, object]:
-        return {"summary": "stub", "confidence": "low", "signals": [], "processing": _processing()}
-
-    def trailing_forward_pe(self) -> dict[str, object]:
-        return {"date": "2026-04-01", "latestValue": 0.0, "history": [], "processing": _processing()}
-
-    def market_breadth(self) -> dict[str, object]:
-        return {"metrics": [], "processing": _processing()}
-
-    def watchlist(self) -> dict[str, object]:
-        return {"items": [], "count": 0, "processing": _processing()}
-
-
-def _fake_chart_opener(
-    data_or_names,
-    *,
-    session_id: str | None = None,
-    **kwargs,
-) -> dict[str, object]:
-    _ = kwargs
-    return {
-        "ok": True,
-        "sessionId": session_id or "agent:chart",
-        "chartUrl": f"http://127.0.0.1:8001/chart?sessionId={session_id or 'agent:chart'}",
-        "processing": _processing(),
-        "inputEcho": data_or_names,
-    }
 
 
 def _loop(model_client, *, max_steps: int = 8, service: _FakeService | None = None) -> TerraFinHostedAgentLoop:

tests/agent/test_runtime.py

@@ -1,203 +1,22 @@
 import pytest
+from fakes import BaseFakeService as _FakeService
+from fakes import fake_chart_opener as _fake_chart_opener
 
 import TerraFin.agent.runtime as agent_runtime
 
 
-def _processing() -> dict[str, object]:
-    return {
-        "requestedDepth": "auto",
-        "resolvedDepth": "full",
-        "loadedStart": "2024-01-01",
-        "loadedEnd": "2024-12-31",
-        "isComplete": True,
-        "hasOlder": False,
-        "sourceVersion": "test-source",
-        "view": "daily",
-    }
-
-
-class _FakeService:
-    def resolve(self, query: str) -> dict[str, object]:
-        return {"type": "stock", "name": query.upper(), "path": f"/stock/{query.upper()}", "processing": _processing()}
-
-    def market_data(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"ticker": name, "seriesType": "candlestick", "count": 1, "data": [], "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def patterns(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"ticker": name, "signals": [], "total": 0, "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def fcf_history(self, ticker: str, years: int = 10) -> dict[str, object]:
-        return {
-            "ticker": ticker, "years": years, "rows": [],
-            "candidates": {"threeYearAvg": None, "latestAnnual": None, "ttm": None},
-            "autoSelectedSource": "annual", "processing": _processing(),
-        }
-
-    def similarity_search(self, ticker: str, universe: str = "sp500+nasdaq100+kospi200", period: str = "1y", top_n: int = 20) -> dict[str, object]:
-        return {"ticker": ticker, "period": period, "pool": {}, "results": [], "count": 0, "processing": _processing()}
-
-    def indicators(
-        self,
-        name: str,
-        indicators: str,
-        *,
-        depth: str = "auto",
-        view: str = "daily",
-    ) -> dict[str, object]:
-        return {
-            "ticker": name,
-            "indicators": {"rsi": {"name": "rsi", "offset": 0, "values": {"value": 55.0}}},
-            "unknown": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view, "indicatorQuery": indicators},
-        }
-
-    def market_snapshot(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {
-            "ticker": name,
-            "price_action": {"current": 100.0},
-            "indicators": {"rsi": 55.0},
-            "market_breadth": [],
-            "watchlist": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view},
-        }
-
-    def lppl_analysis(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"name": name, "confidence": 0.2, "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def company_info(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "shortName": f"{ticker} Corp", "processing": _processing()}
-
-    def earnings(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "earnings": [], "processing": _processing()}
-
-    def financials(self, ticker: str, *, statement: str = "income", period: str = "annual") -> dict[str, object]:
-        return {"ticker": ticker, "statement": statement, "period": period, "columns": [], "rows": [], "processing": _processing()}
-
-    def portfolio(self, guru: str) -> dict[str, object]:
-        return {"guru": guru, "info": {}, "holdings": [], "count": 0, "processing": _processing()}
-
-    def economic(self, indicators: str) -> dict[str, object]:
-        return {"indicators": {indicators: {"latest_value": 3.0}}, "processing": _processing()}
-
-    def macro_focus(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {
-            "name": name,
-            "info": {"name": name, "type": "index", "description": "Macro", "currentValue": 1.0, "change": 0.0, "changePercent": 0.0},
-            "seriesType": "line",
-            "count": 1,
-            "data": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view},
-        }
-
-    def calendar_events(
-        self,
-        *,
-        year: int,
-        month: int,
-        categories: str | None = None,
-        limit: int | None = None,
-    ) -> dict[str, object]:
-        return {"events": [], "count": 0, "month": month, "year": year, "categories": categories, "limit": limit, "processing": _processing()}
-
-    def fundamental_screen(self, ticker: str) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "moat": {"score": "wide"},
-            "earnings_quality": {},
-            "balance_sheet": {},
-            "capital_allocation": {},
-            "pricing_power": {},
-            "warnings": [],
-            "processing": _processing(),
-        }
-
-    def risk_profile(self, name: str, *, depth: str = "auto") -> dict[str, object]:
-        return {
-            "ticker": name,
-            "tail_risk": {},
-            "convexity": {},
-            "volatility": {"requestedDepth": depth},
-            "drawdown": {},
-            "warnings": [],
-            "processing": _processing(),
-        }
-
-    def valuation(self, ticker: str) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "dcf": {"status": "ready", "intrinsic_value": 120.0},
-            "reverse_dcf": {"status": "ready", "implied_growth_pct": 8.0},
-            "relative": {"trailing_pe": 22.0},
-            "graham_number": 100.0,
-            "margin_of_safety_pct": 12.0,
-            "current_price": 107.0,
-            "processing": _processing(),
-        }
-
-    def sec_filings(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "cik": 1, "forms": [], "filings": [], "processing": _processing()}
-
-    def sec_filing_document(
-        self, ticker: str, accession: str, primaryDocument: str, *, form: str = "10-Q"
-    ) -> dict[str, object]:
-        return {"ticker": ticker, "accession": accession, "primaryDocument": primaryDocument, "toc": [], "charCount": 0, "indexUrl": "", "documentUrl": "", "processing": _processing()}
-
-    def sec_filing_section(
-        self, ticker: str, accession: str, primaryDocument: str, sectionSlug: str, *, form: str = "10-Q"
-    ) -> dict[str, object]:
-        return {"ticker": ticker, "accession": accession, "sectionSlug": sectionSlug, "sectionTitle": "stub", "markdown": "", "charCount": 0, "documentUrl": "", "processing": _processing()}
-
-    def fear_greed(self) -> dict[str, object]:
-        return {"score": 50, "rating": "Neutral", "processing": _processing()}
-
-    def sp500_dcf(self) -> dict[str, object]:
-        return {"status": "ready", "currentIntrinsicValue": 5000.0, "processing": _processing()}
-
-    def beta_estimate(self, ticker: str) -> dict[str, object]:
-        return {"symbol": ticker, "beta": 1.0, "adjustedBeta": 1.0, "rSquared": 0.5, "processing": _processing()}
-
-    def top_companies(self) -> dict[str, object]:
-        return {"companies": [], "count": 0, "processing": _processing()}
-
-    def market_regime(self) -> dict[str, object]:
-        return {"summary": "stub", "confidence": "low", "signals": [], "processing": _processing()}
-
-    def trailing_forward_pe(self) -> dict[str, object]:
-        return {"date": "2026-04-01", "latestValue": 0.0, "history": [], "processing": _processing()}
-
-    def market_breadth(self) -> dict[str, object]:
-        return {"metrics": [], "processing": _processing()}
-
-    def watchlist(self) -> dict[str, object]:
-        return {"items": [], "count": 0, "processing": _processing()}
-
-
 class _ExplodingService(_FakeService):
     def market_snapshot(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
         _ = name, depth, view
         raise RuntimeError("snapshot failed")
 
 
-def _fake_chart_opener(
-    data_or_names,
-    *,
-    session_id: str | None = None,
-    **kwargs,
-) -> dict[str, object]:
-    _ = kwargs
-    return {
-        "ok": True,
-        "sessionId": session_id or "agent:chart",
-        "chartUrl": f"http://127.0.0.1:8001/chart?sessionId={session_id or 'agent:chart'}",
-        "processing": _processing(),
-        "inputEcho": data_or_names,
-    }
-
-
-def test_default_capability_registry_contains_kernel_capabilities() -> None:
-    registry = agent_runtime.build_default_capability_registry(_FakeService(), chart_opener=_fake_chart_opener)
-
-    assert registry.names() == (
+# Capabilities that must always be present. Membership is asserted rather than
+# an exact ordered tuple so that adding a capability does not require editing
+# this list; `test_default_capability_registry_ordering_convention` covers the
+# grouping rule that ordering is actually meant to express.
+KERNEL_CAPABILITIES = frozenset(
+    {
         "resolve",
         "market_data",
         "indicators",
@@ -211,9 +30,6 @@ def test_default_capability_registry_contains_kernel_capabilities() -> None:
         "economic",
         "macro_focus",
         "calendar_events",
-        # Dashboard widget-parity capabilities, inserted before `open_chart` so
-        # registry ordering tracks grouping (research read-only first, then
-        # chart-opening, then SEC filings).
         "fear_greed",
         "sp500_dcf",
         "beta_estimate",
@@ -231,7 +47,34 @@ def test_default_capability_registry_contains_kernel_capabilities() -> None:
         "sec_filings",
         "sec_filing_document",
         "sec_filing_section",
-    )
+    }
+)
+
+
+def _registry_names() -> tuple[str, ...]:
+    registry = agent_runtime.build_default_capability_registry(_FakeService(), chart_opener=_fake_chart_opener)
+    return registry.names()
+
+
+def test_default_capability_registry_contains_kernel_capabilities() -> None:
+    names = _registry_names()
+
+    missing = KERNEL_CAPABILITIES - set(names)
+    assert not missing, f"kernel capabilities missing from the registry: {sorted(missing)}"
+    assert len(names) == len(set(names)), "registry contains duplicate capability names"
+
+
+def test_default_capability_registry_ordering_convention() -> None:
+    """Registry order groups research read-only, then chart-opening, then SEC filings."""
+
+    names = _registry_names()
+    sec_positions = [index for index, name in enumerate(names) if name.startswith("sec_")]
+
+    assert sec_positions, "expected SEC filing capabilities in the registry"
+    assert sec_positions == list(
+        range(sec_positions[0], sec_positions[-1] + 1)
+    ), f"SEC filing capabilities must stay contiguous, got positions {sec_positions}"
+    assert names.index("open_chart") < sec_positions[0], "`open_chart` must precede the SEC filing group"
 
 
 def test_context_call_records_focus_and_capability_history() -> None:

tests/agent/test_session_store_light_metadata.py

@@ -1,5 +1,11 @@
 from datetime import UTC, datetime
 
+# Reuse the shared stub service + chart opener from tests/conftest.py:
+# build_default_capability_registry needs every service method, but
+# list_light_metadata itself never invokes any capability.
+from fakes import BaseFakeService as _FakeService
+from fakes import fake_chart_opener as _fake_chart_opener
+
 from TerraFin.agent.contracts.conversation_state import RUNTIME_MODEL_METADATA_KEY
 from TerraFin.agent.runtime import build_default_capability_registry
 from TerraFin.agent.runtime.context import create_agent_context
@@ -11,11 +17,6 @@ from TerraFin.agent.session_store import (
     TerraFinHostedSessionRecord,
 )
 
-# Reuse the full-featured fake service + chart opener the hosted-runtime tests
-# already maintain: build_default_capability_registry needs every service
-# method, but list_light_metadata itself never invokes any capability.
-from tests.agent.test_hosted_runtime import _FakeService, _fake_chart_opener
-
 
 def _ts(hour: int) -> datetime:
     return datetime(2026, 4, 16, hour, 0, tzinfo=UTC)

tests/agent/test_tools.py

@@ -1,4 +1,6 @@
 import pytest
+from fakes import BaseFakeService, processing
+from fakes import fake_chart_opener as _fake_chart_opener
 
 from TerraFin.agent.definitions import (
     DEFAULT_HOSTED_AGENT_NAME,
@@ -10,206 +12,9 @@ from TerraFin.agent.runtime import build_default_capability_registry
 from TerraFin.agent.tools import TerraFinHostedToolAdapter
 
 
-def _processing() -> dict[str, object]:
-    return {
-        "requestedDepth": "auto",
-        "resolvedDepth": "full",
-        "loadedStart": "2024-01-01",
-        "loadedEnd": "2024-12-31",
-        "isComplete": True,
-        "hasOlder": False,
-        "sourceVersion": "test-source",
-        "view": "daily",
-    }
-
-
-class _FakeService:
-    def resolve(self, query: str) -> dict[str, object]:
-        return {"type": "stock", "name": query.upper(), "path": f"/stock/{query.upper()}", "processing": _processing()}
-
-    def market_data(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"ticker": name, "seriesType": "candlestick", "count": 1, "data": [], "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def indicators(
-        self,
-        name: str,
-        indicators: str,
-        *,
-        depth: str = "auto",
-        view: str = "daily",
-    ) -> dict[str, object]:
-        return {
-            "ticker": name,
-            "indicators": {"rsi": {"name": "rsi", "offset": 0, "values": {"value": 55.0}}},
-            "unknown": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view, "indicatorQuery": indicators},
-        }
-
-    def patterns(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"ticker": name, "signals": [], "total": 0, "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def market_snapshot(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {
-            "ticker": name,
-            "price_action": {"current": 100.0},
-            "indicators": {"rsi": 55.0},
-            "market_breadth": [],
-            "watchlist": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view},
-        }
-
-    def lppl_analysis(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {"name": name, "confidence": 0.2, "processing": {**_processing(), "requestedDepth": depth, "view": view}}
-
-    def company_info(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "shortName": f"{ticker} Corp", "processing": _processing()}
-
-    def earnings(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "earnings": [], "processing": _processing()}
-
-    def financials(self, ticker: str, *, statement: str = "income", period: str = "annual") -> dict[str, object]:
-        return {"ticker": ticker, "statement": statement, "period": period, "columns": [], "rows": [], "processing": _processing()}
-
-    def portfolio(self, guru: str) -> dict[str, object]:
-        return {"guru": guru, "info": {}, "holdings": [], "count": 0, "processing": _processing()}
-
-    def economic(self, indicators: str) -> dict[str, object]:
-        return {"indicators": {indicators: {"latest_value": 3.0}}, "processing": _processing()}
-
-    def macro_focus(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
-        return {
-            "name": name,
-            "info": {"name": name, "type": "index", "description": "Macro", "currentValue": 1.0, "change": 0.0, "changePercent": 0.0},
-            "seriesType": "line",
-            "count": 1,
-            "data": [],
-            "processing": {**_processing(), "requestedDepth": depth, "view": view},
-        }
-
-    def calendar_events(
-        self,
-        *,
-        year: int,
-        month: int,
-        categories: str | None = None,
-        limit: int | None = None,
-    ) -> dict[str, object]:
-        return {"events": [], "count": 0, "month": month, "year": year, "categories": categories, "limit": limit, "processing": _processing()}
-
-    def fundamental_screen(self, ticker: str) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "moat": {"score": "wide"},
-            "earnings_quality": {},
-            "balance_sheet": {},
-            "capital_allocation": {},
-            "pricing_power": {},
-            "warnings": [],
-            "processing": _processing(),
-        }
-
-    def risk_profile(self, name: str, *, depth: str = "auto") -> dict[str, object]:
-        return {
-            "ticker": name,
-            "tail_risk": {},
-            "convexity": {},
-            "volatility": {"requestedDepth": depth},
-            "drawdown": {},
-            "warnings": [],
-            "processing": _processing(),
-        }
-
-    def valuation(self, ticker: str) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "dcf": {"status": "ready", "intrinsic_value": 120.0},
-            "reverse_dcf": {"status": "ready", "implied_growth_pct": 8.0},
-            "relative": {"trailing_pe": 22.0},
-            "graham_number": 100.0,
-            "margin_of_safety_pct": 12.0,
-            "current_price": 107.0,
-            "processing": _processing(),
-        }
-
+class _FakeService(BaseFakeService):
     def sec_filings(self, ticker: str) -> dict[str, object]:
-        return {"ticker": ticker, "cik": 1, "forms": ["10-K"], "filings": [], "processing": _processing()}
-
-    def sec_filing_document(
-        self, ticker: str, accession: str, primaryDocument: str, *, form: str = "10-Q"
-    ) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "accession": accession,
-            "primaryDocument": primaryDocument,
-            "toc": [],
-            "charCount": 0,
-            "indexUrl": "",
-            "documentUrl": "",
-            "processing": _processing(),
-        }
-
-    def sec_filing_section(
-        self,
-        ticker: str,
-        accession: str,
-        primaryDocument: str,
-        sectionSlug: str,
-        *,
-        form: str = "10-Q",
-    ) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "accession": accession,
-            "sectionSlug": sectionSlug,
-            "sectionTitle": "stub",
-            "markdown": "",
-            "charCount": 0,
-            "documentUrl": "",
-            "processing": _processing(),
-        }
-
-    def fcf_history(self, ticker: str, years: int = 10) -> dict[str, object]:
-        return {
-            "ticker": ticker,
-            "years": years,
-            "rows": [],
-            "candidates": {"threeYearAvg": None, "latestAnnual": None, "ttm": None},
-            "autoSelectedSource": "annual",
-            "processing": _processing(),
-        }
-
-    def similarity_search(
-        self,
-        ticker: str,
-        universe: str = "sp500+nasdaq100+kospi200",
-        period: str = "1y",
-        top_n: int = 20,
-    ) -> dict[str, object]:
-        return {"ticker": ticker, "period": period, "pool": {}, "results": [], "count": 0, "processing": _processing()}
-
-    def fear_greed(self) -> dict[str, object]:
-        return {"score": 50, "rating": "Neutral", "processing": _processing()}
-
-    def sp500_dcf(self) -> dict[str, object]:
-        return {"status": "ready", "currentIntrinsicValue": 5000.0, "processing": _processing()}
-
-    def beta_estimate(self, ticker: str) -> dict[str, object]:
-        return {"symbol": ticker, "beta": 1.0, "adjustedBeta": 1.0, "rSquared": 0.5, "processing": _processing()}
-
-    def top_companies(self) -> dict[str, object]:
-        return {"companies": [], "count": 0, "processing": _processing()}
-
-    def market_regime(self) -> dict[str, object]:
-        return {"summary": "stub", "confidence": "low", "signals": [], "processing": _processing()}
-
-    def trailing_forward_pe(self) -> dict[str, object]:
-        return {"date": "2026-04-01", "latestValue": 0.0, "history": [], "processing": _processing()}
-
-    def market_breadth(self) -> dict[str, object]:
-        return {"metrics": [], "processing": _processing()}
-
-    def watchlist(self) -> dict[str, object]:
-        return {"items": [], "count": 0, "processing": _processing()}
+        return {"ticker": ticker, "cik": 1, "forms": ["10-K"], "filings": [], "processing": processing()}
 
 
 class _RetryingFakeService(_FakeService):
@@ -238,20 +43,6 @@ class _MacroFocusEquityMisuseService(_FakeService):
         raise LookupError(f"Unknown macro instrument: '{name}'")
 
 
-def _fake_chart_opener(
-    data_or_names,
-    *,
-    session_id: str | None = None,
-    **kwargs,
-) -> dict[str, object]:
-    _ = kwargs
-    return {
-        "ok": True,
-        "sessionId": session_id or "agent:chart",
-        "chartUrl": f"http://127.0.0.1:8001/chart?sessionId={session_id or 'agent:chart'}",
-        "processing": _processing(),
-        "inputEcho": data_or_names,
-    }
 
 
 def _adapter(

tests/fakes.py

@@ -0,0 +1,228 @@
+"""Shared stub doubles for the TerraFin test suite.
+
+Imported as `from fakes import ...`; `tests/` is on `sys.path` via the
+`pythonpath` setting in `pyproject.toml`.
+
+`BaseFakeService` is the single stub service used by every test that builds the
+default capability registry. `build_default_capability_registry` binds
+`service.<method>` eagerly for each registered capability, so a fake that is
+missing a method fails at registry-build time with `AttributeError`.
+
+To keep that from turning every new capability into a mechanical edit across
+several test modules, unknown attributes resolve through `__getattr__` to a
+generic stub. Methods are still spelled out explicitly below when a test
+asserts on their payload shape; anything added later works without changes
+here. Override a single method in a subclass when a module needs a different
+payload.
+"""
+
+from typing import Any
+
+
+def processing() -> dict[str, object]:
+    """Return the standard `processing` metadata block used by stub payloads."""
+
+    return {
+        "requestedDepth": "auto",
+        "resolvedDepth": "full",
+        "loadedStart": "2024-01-01",
+        "loadedEnd": "2024-12-31",
+        "isComplete": True,
+        "hasOlder": False,
+        "sourceVersion": "test-source",
+        "view": "daily",
+    }
+
+
+class BaseFakeService:
+    """Stub `TerraFinAgentService` covering the default capability registry."""
+
+    def __getattr__(self, name: str) -> Any:
+        """Resolve capabilities with no explicit stub to a generic payload.
+
+        Private and dunder lookups must still raise so that `copy`, `pickle`,
+        and pytest introspection keep working.
+        """
+
+        if name.startswith("_"):
+            raise AttributeError(name)
+
+        def _auto_stub(*args: Any, **kwargs: Any) -> dict[str, object]:
+            return {
+                "capability": name,
+                "args": list(args),
+                "kwargs": kwargs,
+                "processing": processing(),
+            }
+
+        return _auto_stub
+
+    def resolve(self, query: str) -> dict[str, object]:
+        return {"type": "stock", "name": query.upper(), "path": f"/stock/{query.upper()}", "processing": processing()}
+
+    def market_data(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
+        return {"ticker": name, "seriesType": "candlestick", "count": 1, "data": [], "processing": {**processing(), "requestedDepth": depth, "view": view}}
+
+    def patterns(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
+        return {"ticker": name, "signals": [], "total": 0, "processing": {**processing(), "requestedDepth": depth, "view": view}}
+
+    def fcf_history(self, ticker: str, years: int = 10) -> dict[str, object]:
+        return {
+            "ticker": ticker, "years": years, "rows": [],
+            "candidates": {"threeYearAvg": None, "latestAnnual": None, "ttm": None},
+            "autoSelectedSource": "annual", "processing": processing(),
+        }
+
+    def similarity_search(self, ticker: str, universe: str = "sp500+nasdaq100+kospi200", period: str = "1y", top_n: int = 20) -> dict[str, object]:
+        return {"ticker": ticker, "period": period, "pool": {}, "results": [], "count": 0, "processing": processing()}
+
+    def indicators(
+        self,
+        name: str,
+        indicators: str,
+        *,
+        depth: str = "auto",
+        view: str = "daily",
+    ) -> dict[str, object]:
+        return {
+            "ticker": name,
+            "indicators": {"rsi": {"name": "rsi", "offset": 0, "values": {"value": 55.0}}},
+            "unknown": [],
+            "processing": {**processing(), "requestedDepth": depth, "view": view, "indicatorQuery": indicators},
+        }
+
+    def market_snapshot(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
+        return {
+            "ticker": name,
+            "price_action": {"current": 100.0},
+            "indicators": {"rsi": 55.0},
+            "market_breadth": [],
+            "watchlist": [],
+            "processing": {**processing(), "requestedDepth": depth, "view": view},
+        }
+
+    def lppl_analysis(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
+        return {"name": name, "confidence": 0.2, "processing": {**processing(), "requestedDepth": depth, "view": view}}
+
+    def company_info(self, ticker: str) -> dict[str, object]:
+        return {"ticker": ticker, "shortName": f"{ticker} Corp", "processing": processing()}
+
+    def earnings(self, ticker: str) -> dict[str, object]:
+        return {"ticker": ticker, "earnings": [], "processing": processing()}
+
+    def financials(self, ticker: str, *, statement: str = "income", period: str = "annual") -> dict[str, object]:
+        return {"ticker": ticker, "statement": statement, "period": period, "columns": [], "rows": [], "processing": processing()}
+
+    def portfolio(self, guru: str) -> dict[str, object]:
+        return {"guru": guru, "info": {}, "holdings": [], "count": 0, "processing": processing()}
+
+    def economic(self, indicators: str) -> dict[str, object]:
+        return {"indicators": {indicators: {"latest_value": 3.0}}, "processing": processing()}
+
+    def macro_focus(self, name: str, *, depth: str = "auto", view: str = "daily") -> dict[str, object]:
+        return {
+            "name": name,
+            "info": {"name": name, "type": "index", "description": "Macro", "currentValue": 1.0, "change": 0.0, "changePercent": 0.0},
+            "seriesType": "line",
+            "count": 1,
+            "data": [],
+            "processing": {**processing(), "requestedDepth": depth, "view": view},
+        }
+
+    def calendar_events(
+        self,
+        *,
+        year: int,
+        month: int,
+        categories: str | None = None,
+        limit: int | None = None,
+    ) -> dict[str, object]:
+        return {"events": [], "count": 0, "month": month, "year": year, "categories": categories, "limit": limit, "processing": processing()}
+
+    def fundamental_screen(self, ticker: str) -> dict[str, object]:
+        return {
+            "ticker": ticker,
+            "moat": {"score": "wide"},
+            "earnings_quality": {},
+            "balance_sheet": {},
+            "capital_allocation": {},
+            "pricing_power": {},
+            "warnings": [],
+            "processing": processing(),
+        }
+
+    def risk_profile(self, name: str, *, depth: str = "auto") -> dict[str, object]:
+        return {
+            "ticker": name,
+            "tail_risk": {},
+            "convexity": {},
+            "volatility": {"requestedDepth": depth},
+            "drawdown": {},
+            "warnings": [],
+            "processing": processing(),
+        }
+
+    def valuation(self, ticker: str) -> dict[str, object]:
+        return {
+            "ticker": ticker,
+            "dcf": {"status": "ready", "intrinsic_value": 120.0},
+            "reverse_dcf": {"status": "ready", "implied_growth_pct": 8.0},
+            "relative": {"trailing_pe": 22.0},
+            "graham_number": 100.0,
+            "margin_of_safety_pct": 12.0,
+            "current_price": 107.0,
+            "processing": processing(),
+        }
+
+    def sec_filings(self, ticker: str) -> dict[str, object]:
+        return {"ticker": ticker, "cik": 1, "forms": [], "filings": [], "processing": processing()}
+
+    def sec_filing_document(
+        self, ticker: str, accession: str, primaryDocument: str, *, form: str = "10-Q"
+    ) -> dict[str, object]:
+        return {"ticker": ticker, "accession": accession, "primaryDocument": primaryDocument, "toc": [], "charCount": 0, "indexUrl": "", "documentUrl": "", "processing": processing()}
+
+    def sec_filing_section(
+        self, ticker: str, accession: str, primaryDocument: str, sectionSlug: str, *, form: str = "10-Q"
+    ) -> dict[str, object]:
+        return {"ticker": ticker, "accession": accession, "sectionSlug": sectionSlug, "sectionTitle": "stub", "markdown": "", "charCount": 0, "documentUrl": "", "processing": processing()}
+
+    def fear_greed(self) -> dict[str, object]:
+        return {"score": 50, "rating": "Neutral", "processing": processing()}
+
+    def sp500_dcf(self) -> dict[str, object]:
+        return {"status": "ready", "currentIntrinsicValue": 5000.0, "processing": processing()}
+
+    def beta_estimate(self, ticker: str) -> dict[str, object]:
+        return {"symbol": ticker, "beta": 1.0, "adjustedBeta": 1.0, "rSquared": 0.5, "processing": processing()}
+
+    def top_companies(self) -> dict[str, object]:
+        return {"companies": [], "count": 0, "processing": processing()}
+
+    def market_regime(self) -> dict[str, object]:
+        return {"summary": "stub", "confidence": "low", "signals": [], "processing": processing()}
+
+    def trailing_forward_pe(self) -> dict[str, object]:
+        return {"date": "2026-04-01", "latestValue": 0.0, "history": [], "processing": processing()}
+
+    def market_breadth(self) -> dict[str, object]:
+        return {"metrics": [], "processing": processing()}
+
+    def watchlist(self) -> dict[str, object]:
+        return {"items": [], "count": 0, "processing": processing()}
+
+
+def fake_chart_opener(
+    data_or_names,
+    *,
+    session_id: str | None = None,
+    **kwargs,
+) -> dict[str, object]:
+    _ = kwargs
+    return {
+        "ok": True,
+        "sessionId": session_id or "agent:chart",
+        "chartUrl": f"http://127.0.0.1:8001/chart?sessionId={session_id or 'agent:chart'}",
+        "processing": processing(),
+        "inputEcho": data_or_names,
+    }