bench: SF10 perf validation, Tier-B token harness, and the finalized token-efficiency doc scope

[devopam]

Summary

• SF10 performance validation (docs/performance-benchmark.md) — the published perf writeup, generated from a real TPC-H SF10 run.
• Tier-B agent-loop token harness (benchmarks/tokens/tier_b/) — built out with checkpoint/resume, a real-Claude-Code (claude -p) on/off comparison mode, and real tool-trace + cost instrumentation. Fixed two --allowedTools scoping bugs found while using it (missing Bash, an incomplete curated tool list) that were causing spurious permission-stall "failures." Internal diagnostic tooling — not part of the published claim (see below).
• translate_nl_to_sql token-usage fix — the tool's internal LLM call was a total blind spot in cost accounting; it now reports tokens_in/tokens_out from the actual provider response and wires them into the existing (previously dead) audit_nl2sql columns.
• mcpg --demo fix — reviews.customer_id was an unplanted, unindexed FK; now indexed as the seed script's own comment already claimed.
• Finalized the published token-efficiency claim to Tier A only. docs/token-efficiency.md no longer promises a future agent-loop/session-level savings result; roadmap rows 19.4–19.6 and the original benchmark-suite plan updated to match. The agent-loop objective is descoped from the published claim; its harness stays in-tree as internal tooling only.
• Merged main (this branch was 9 commits behind) — resolved 4 conflicts, all between independently-written fixes for the same underlying issues (Windows event-loop policy on Python 3.14; Windows-only test path bugs); kept the more robust side in each case, verified no functional loss.

Roadmap linkage

Advances roadmap row: 19.1
Advances roadmap row: 19.2
Advances roadmap row: 19.3
Advances roadmap row: 19.4
Advances roadmap row: 19.5
Advances roadmap row: 19.6

Checklist

• Tests added/updated first (TDD); suite passes locally (2832 passed, 2 skipped — unit; 47 passed — contract)
• ruff, ruff format, and mypy src/mcpg pass
• CHANGELOG.md updated under [Unreleased]
• Roadmap rows cited above; 19.4 and 19.6 marked ✅ shipped, 19.5 marked descoped
• No hand-edits to src/mcpg/_vendor/

🤖 Generated with Claude Code

https://claude.ai/code/session_015CgKWtcUQYcfuQv3mpyGQ9

Summary by Sourcery

Add SF10 performance benchmark artifacts, enhance Tier-B token study harness and diagnostics, and finalize token-efficiency documentation scope to deterministic per-call accounting only.

New Features:

• Introduce SF10 performance benchmark run JSON and HTML dashboard artifacts and document the constrained-environment validation run.
• Extend Tier-B token benchmark tasks with analytical NL→SQL questions and real-harness diagnostic task sets.
• Add real-harness comparison and forced-translate_nl_to_sql experiment scripts for Tier-B to study tool choice and token/cost behavior.

Bug Fixes:

• Ensure translate_nl_to_sql surfaces its internal provider token usage on TranslationResult and into nl2sql audit records, fixing a blind spot in cost accounting.
• Index reviews.customer_id in the demo dataset to remove an unintended unindexed foreign key while preserving the intended orders.customer_id planted finding.
• Fix subprocess bin-allowlist tests to use truly absolute paths on Windows and POSIX.
• Allow performance benchmark runners to configure query timeouts consistently via a new --timeout flag and MCPG_STATEMENT_TIMEOUT_MS, preventing spurious timeouts on heavy SF10 queries.

Enhancements:

• Add checkpoint/resume support and e2e-path failure resilience to the performance benchmark runner so long runs are safely persisted and interruptions only lose minimal work.
• Add checkpoint/resume and configuration-matching safeguards to the Tier-B synthetic harness so paid trials are not re-run and incompatible aggregates are avoided.
• Track tools with hidden internal LLM cost in the Tier-B agent loop and fold their token usage into total_tokens while preserving a separate breakdown.
• Extend translate_nl_to_sql’s provider implementations to parse and expose vendor-specific usage blocks for Anthropic, OpenAI-compatible, and Gemini responses, with robust defaults when usage is missing.
• Update contract tests and tool-output schema expectations to include translate_nl_to_sql token fields.
• Improve benchmark dashboard generation to read/write files with explicit UTF-8 encoding for portability.

Documentation:

• Document the SF10 performance validation run, its methodology deviations, and findings in docs/performance-benchmark.md.
• Narrow docs/token-efficiency.md to Tier-A per-call compactness only, remove future Tier-B agent-loop promises, and clarify scope and caveats.
• Clarify the original benchmark-suite plan’s status and mark the agent-loop objective as descoped from the published claim.
• Update the feature shortlist roadmap to mark token accounting Tier A and the combined writeup as shipped and Tier-B agent-loop study as descoped.
• Refine Tier-B task documentation to distinguish planted-flaw audit tasks from analytical NL→SQL tasks and explain grading and correctness margins.

Tests:

• Add unit tests to verify translate_nl_to_sql propagates provider token usage across all result branches and that providers parse usage from real HTTP response shapes with safe defaults.
• Add Tier-B aggregation and hidden-token tests to ensure total_tokens includes internal tool LLM usage and that hidden tokens default correctly when absent.
• Add tests for Tier-B agent hidden-token extraction from tool structuredContent and ensure only approved tools contribute hidden cost.
• Adjust demo dataset tests to account for the new reviews.customer_id index.
• Strengthen config tests around subprocess bin allowlists on multi-platform absolute paths.
• Update dashboard generation tests indirectly via new SF10 artifacts and token-report handling.

[gemini-code-assist-2]

Code Review

This pull request introduces several enhancements to the benchmark suite and token accounting, most notably capturing and reporting the internal LLM token usage for the translate_nl_to_sql tool across Anthropic, OpenAI, and Gemini providers. It also adds checkpointing and resumability to the performance and Tier-B token runners to prevent data loss during long runs, introduces a --timeout option for the performance runner, and indexes reviews.customer_id in the demo seed script. Documentation and feature plans have been updated accordingly. Feedback on the changes points out a potential UnicodeEncodeError on Windows when writing the performance checkpoint file without an explicit UTF-8 encoding, and a code suggestion has been provided to resolve this.

[gemini-code-assist-2]

Writing the JSON checkpoint file without specifying an explicit encoding (e.g., encoding="utf-8") defaults to the platform-dependent locale encoding. On Windows environments, this often defaults to CP1252, which can cause a UnicodeEncodeError if any non-ASCII characters are present in the metadata (such as the PostgreSQL version string or platform information). Specifying encoding="utf-8" ensures consistent and robust behavior across all platforms, matching the fixes applied to other benchmark scripts in this PR.

Suggested change

	args.output.parent.mkdir(parents=True, exist_ok=True)
	args.output.write_text(json.dumps(run.to_dict(), indent=2) + "\n")
	args.output.parent.mkdir(parents=True, exist_ok=True)
	args.output.write_text(json.dumps(run.to_dict(), indent=2) + "\n", encoding="utf-8")

[sourcery-ai]

Hey - I've found 5 issues, and left some high level feedback:

• Consider keeping the generated benchmark artifacts (perf-sf10.html/json, real-harness comparison JSONs, forced-nl2sql-comparison.json) out of the tracked tree or moving them under a dedicated artifacts/ or samples/ area so they don’t grow the repo and get confused with source-of-truth inputs.
• The checkpoint/resume guards for Tier-B and the real-harness comparison only key off model/trials/config flags; it would be safer to include database_url (and for real_harness, worktree_dir) in _RESUME_KEYS so runs against different databases or code trees can’t be silently merged into one aggregate.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- Consider keeping the generated benchmark artifacts (perf-sf10.html/json, real-harness comparison JSONs, forced-nl2sql-comparison.json) out of the tracked tree or moving them under a dedicated `artifacts/` or `samples/` area so they don’t grow the repo and get confused with source-of-truth inputs.
- The checkpoint/resume guards for Tier-B and the real-harness comparison only key off model/trials/config flags; it would be safer to include `database_url` (and for real_harness, `worktree_dir`) in `_RESUME_KEYS` so runs against different databases or code trees can’t be silently merged into one aggregate.

## Individual Comments

### Comment 1
<location path="benchmarks/tokens/tier_b/tasks.py" line_range="123-132" />
<code_context>
+    ]
+
+
+def audit_tasks() -> list[Task]:
+    """The three planted-flaw DBA tasks from ``default_tasks()``, isolated for the real-harness diagnostic.
+
+    The synthetic Tier-B harness measured these three costing MCPg's
+    advisor arm 2.66x more tokens than bare SQL — the opposite of what the
+    tool descriptions promise (one advisor call vs. many exploratory
+    queries) — and that number was never explained, only recorded. This
+    subset lets ``real_harness_comparison`` re-run just these three through
+    real Claude Code with tool-trace capture, to see *why*: whether the
+    advisor tool gets called and then the agent explores anyway, whether
+    its output isn't trusted, or something else.
+    """
+    return default_tasks()[:3]
+
+
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Selecting audit tasks by list slice is brittle and will silently change if default_tasks() ordering ever changes.

audit_tasks() assumes the first three entries in default_tasks() are always the planted-flaw audit tasks. Any future reordering or prepending in default_tasks() will change this subset without obvious breakage. Please select these tasks by explicit identifiers (e.g., missing_index, pii_columns, naming_violation) instead of positional slicing to keep the behavior stable as default_tasks() evolves.

Suggested implementation:

```python
    ]


 def audit_tasks() -> list[Task]:
     """The three planted-flaw DBA tasks from ``default_tasks()``, isolated for the real-harness diagnostic.

     The synthetic Tier-B harness measured these three costing MCPg's
     advisor arm 2.66x more tokens than bare SQL — the opposite of what the
     tool descriptions promise (one advisor call vs. many exploratory
     queries) — and that number was never explained, only recorded. This
     subset lets ``real_harness_comparison`` re-run just these three through
     real Claude Code with tool-trace capture, to see *why*: whether the
     advisor tool gets called and then the agent explores anyway, whether
     its output isn't trusted, or something else.

     Tasks are selected by explicit identifiers (``missing_index``,
     ``pii_columns``, ``naming_violation``) rather than positional slicing,
     so changes to ``default_tasks()`` ordering do not silently alter this
     subset.
     """
     audit_ids = ("missing_index", "pii_columns", "naming_violation")

     # Build an index of tasks by identifier to avoid relying on list order.
     tasks_by_id: dict[str, Task] = {task.id: task for task in default_tasks()}

     try:
         return [tasks_by_id[audit_id] for audit_id in audit_ids]
     except KeyError as exc:
         # Fail loudly if any expected planted-flaw task is missing, rather
         # than silently changing the audit subset.
         missing = ", ".join(audit_id for audit_id in audit_ids if audit_id not in tasks_by_id)
         raise RuntimeError(
             f"Expected planted-flaw audit tasks missing from default_tasks(): {missing}"
         ) from exc


```

I assumed the task identifier attribute is named `id`. If your `Task` model uses a different field (e.g. `slug`, `name`, or `key`), update `task.id` and the `audit_ids` values to match the actual identifiers for the planted-flaw tasks (`missing_index`, `pii_columns`, `naming_violation`). Also ensure `Task` is hashable and that `default_tasks()` returns a list containing these three tasks with those identifiers.
</issue_to_address>

### Comment 2
<location path="benchmarks/tokens/tier_b/agent.py" line_range="70-74" />
<code_context>
+_TOOLS_WITH_HIDDEN_LLM_COST = {"translate_nl_to_sql"}
+
+
+def _hidden_tokens(tool_name: str, result: Any) -> tuple[int, int]:
+    """The tokens ``tool_name``'s own internal LLM call spent, or (0, 0)."""
+    if tool_name not in _TOOLS_WITH_HIDDEN_LLM_COST:
+        return 0, 0
+    structured = getattr(result, "structuredContent", None) or {}
+    return int(structured.get("tokens_in") or 0), int(structured.get("tokens_out") or 0)
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Assuming structuredContent is always a dict can raise at runtime if the SDK payload shape changes.

`structured = getattr(result, "structuredContent", None) or {}` assumes `structuredContent` is a mapping. If a future SDK/tool returns a list or other non-mapping type, `structured.get(...)` will raise `AttributeError` and break trial accounting. Please guard with `isinstance(structured, dict)` (or `Mapping`) before using `.get`, and return `(0, 0)` for unsupported types.
</issue_to_address>

### Comment 3
<location path="benchmarks/perf/runner.py" line_range="215-224" />
<code_context>
+def _checkpoint(
</code_context>
<issue_to_address>
**nitpick (bug_risk):** Checkpoint writing omits an explicit encoding, unlike other new write_text calls that standardize on UTF-8.

In `_checkpoint()`, `args.output.write_text(...)` still uses the platform default encoding, unlike the other updated paths that explicitly set `encoding="utf-8"`. Please pass `encoding="utf-8"` here as well to keep checkpoint files consistent and avoid cross-platform issues (e.g., non-UTF-8 defaults on Windows).
</issue_to_address>

### Comment 4
<location path="benchmarks/tokens/tier_b/experiments/real_harness_comparison.py" line_range="178" />
<code_context>
+def _invoke_claude(
</code_context>
<issue_to_address>
**suggestion (bug_risk):** The claude subprocess invocation does not enforce any overall timeout, so a stuck CLI could hang the experiment.

`_invoke_claude()` starts `claude -p` and calls `proc.communicate()` without a timeout. If the CLI or its network call hangs, the experiment can block indefinitely and never checkpoint the current trial. Please consider wrapping `communicate()` in `asyncio.wait_for` with a reasonable upper bound and, on timeout, recording a `TrialResult` error so one bad invocation cannot stall the entire run.
</issue_to_address>

### Comment 5
<location path="benchmarks/tokens/tier_b/runner.py" line_range="131" />
<code_context>
+    return trials
+
+
+async def _run(args: argparse.Namespace) -> TierBReport:
+    nl2sql_key = os.environ.get("MCPG_NL2SQL_API_KEY") or os.environ.get("ANTHROPIC_API_KEY", "")
+    if not nl2sql_key:
</code_context>
<issue_to_address>
**issue (complexity):** Consider encapsulating the resume/checkpoint behavior into a dedicated helper that owns configuration, state, and persistence so `_run` can focus solely on orchestrating task execution.

The resume/checkpoint logic is useful, but it now cross‑cuts `_run` and introduces several implicit contracts. You can keep all functionality while isolating this concern and making the main loop easier to follow.

### 1. Extract a small helper to own resume/checkpoint state

Right now `_load_resumable`, `_checkpoint`, `_RESUME_KEYS`, and the `done` set are spread across the module. Pull them into a tiny helper that owns:

- loading existing trials
- deciding whether a `(task_id, arm, trial)` is done
- checkpointing after each new trial
- writing the final report

This keeps `_run` focused on task execution.

```python
@dataclass(frozen=True)
class RunConfig:
    model: str
    trials_per_arm: int
    max_turns: int
    mcpg_tools: list[str]

    @classmethod
    def from_args(cls, args: argparse.Namespace, resolved_mcpg_tools: set[str] | None = None) -> "RunConfig":
        tools = sorted(resolved_mcpg_tools or args.mcpg_tools)
        return cls(
            model=args.model,
            trials_per_arm=args.trials,
            max_turns=args.max_turns,
            mcpg_tools=tools,
        )

class ResumableRun:
    def __init__(self, args: argparse.Namespace, resolved_mcpg_tools: set[str]):
        self.args = args
        self.config = RunConfig.from_args(args, resolved_mcpg_tools)
        self.trials: list[TrialResult] = self._load_resumable()
        self._done = {(t.task_id, t.arm, t.trial) for t in self.trials}

    def is_done(self, task_id: str, arm: str, trial: int) -> bool:
        return (task_id, arm, trial) in self._done

    def record(self, result: TrialResult, *, complete: bool = False) -> TierBReport:
        self.trials.append(result)
        self._done.add((result.task_id, result.arm, result.trial))
        return self._checkpoint(complete=complete)

    def finalize(self) -> TierBReport:
        return self._checkpoint(complete=True)

    # existing _load_resumable / _checkpoint logic moved in here, using self.config
    def _load_resumable(self) -> list[TrialResult]:
        ...
    def _checkpoint(self, *, complete: bool) -> TierBReport:
        ...
```

Then `_run` becomes much clearer:

```python
async def _run(args: argparse.Namespace) -> TierBReport:
    settings = load_settings({"MCPG_DATABASE_URL": args.database_url})
    model = AnthropicClient(args.model)
    tasks = default_tasks()

    async with AsyncExitStack() as stack:
        server = create_server(settings)
        session = await stack.enter_async_context(
            create_connected_server_and_client_session(server, read_timeout_seconds=_READ_TIMEOUT)
        )
        listed = await session.list_tools()
        server_names = {t.name for t in listed.tools}
        mcpg_tools = {n for n in args.mcpg_tools if n in server_names}

        missing = set(args.mcpg_tools) - server_names
        if missing:
            print(f"warning: requested mcpg tools not on the server, skipped: {sorted(missing)}")

        runner = ResumableRun(args, resolved_mcpg_tools=mcpg_tools)
        arms = ((ARM_BASELINE, _BASELINE_TOOLS & server_names), (ARM_MCPG, mcpg_tools))

        for task in tasks:
            for trial in range(args.trials):
                for arm, allowed in arms:
                    if runner.is_done(task.id, arm, trial):
                        print(f"  {task.id:16} {arm:9} #{trial}: skipped (already recorded)")
                        continue

                    result = await run_trial(
                        task,
                        session=session,
                        model=model,
                        allowed_tools=allowed,
                        arm=arm,
                        trial=trial,
                        max_turns=args.max_turns,
                    )
                    flag = "PASS" if result.passed else ("ERR " if result.error else "FAIL")
                    print(
                        f"  {task.id:16} {arm:9} #{trial}: "
                        f"tok={result.total_tokens:6} tools={result.tool_calls:2} turns={result.turns:2} {flag}"
                    )
                    runner.record(result, complete=False)

        return runner.finalize()
```

This keeps the nested loops “for task, for trial, for arm, run_trial” visually intact, with resume/checkpoint concerns pushed behind `ResumableRun`.

### 2. Make the metadata/config contract explicit

You currently duplicate the config fields in `_checkpoint` and `_load_resumable` and rely on `_RESUME_KEYS`. The `RunConfig` above can be used to:

- serialize/deserialize the run configuration to/from metadata
- compare the on‑disk config with the current invocation

For example inside `ResumableRun._load_resumable`:

```python
def _load_resumable(self) -> list[TrialResult]:
    if not self.args.output.exists():
        return []
    try:
        data = json.loads(self.args.output.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return []

    meta = data.get("metadata", {})
    if meta.get("complete"):
        print(f"note: {self.args.output} already holds a complete run; overwriting with a fresh one.")
        return []

    disk_config = RunConfig(
        model=meta.get("model"),
        trials_per_arm=meta.get("trials_per_arm"),
        max_turns=meta.get("max_turns"),
        mcpg_tools=meta.get("mcpg_tools") or [],
    )
    if disk_config != self.config:
        print(f"note: {self.args.output} exists but its config differs from this run; starting fresh, not resuming.")
        return []

    trials = [TrialResult(**t) for t in data.get("trials", [])]
    if trials:
        print(f"resuming from {self.args.output}: {len(trials)} trial(s) already recorded, will not be re-run.")
    return trials
```

And `_checkpoint` can build metadata from `self.config`, reducing repetition:

```python
def _checkpoint(self, *, complete: bool) -> TierBReport:
    metadata: dict[str, Any] = {
        "timestamp": self.args.timestamp,
        "git_sha": self.args.git_sha,
        "mcpg_version": __version__,
        "model": self.config.model,
        "trials_per_arm": self.config.trials_per_arm,
        "max_turns": self.config.max_turns,
        "mcpg_tools": self.config.mcpg_tools,
        "host": {"python": platform.python_version(), "os": platform.platform()},
        "complete": complete,
    }
    report = TierBReport(metadata=metadata, trials=self.trials, aggregate=aggregate(self.trials))
    self.args.output.parent.mkdir(parents=True, exist_ok=True)
    self.args.output.write_text(json.dumps(report.to_dict(), indent=2) + "\n", encoding="utf-8")
    return report
```

This makes the matching criteria explicit and central, avoiding implicit contracts between multiple helpers.

### 3. Optional: decouple checkpoint frequency from loop structure

If you later want to reduce I/O without changing semantics, you can let `ResumableRun` control checkpoint cadence (e.g. every N trials, or per task) without touching the loop:

```python
class ResumableRun:
    def __init__(..., checkpoint_interval: int = 1):
        ...
        self._checkpoint_interval = checkpoint_interval

    def record(self, result: TrialResult, *, complete: bool = False) -> TierBReport | None:
        self.trials.append(result)
        self._done.add((result.task_id, result.arm, result.trial))
        if complete or len(self.trials) % self._checkpoint_interval == 0:
            return self._checkpoint(complete=complete)
        return None
```

You can keep the default `checkpoint_interval=1` to preserve the “after every trial” behavior while making it easier to tune later.

These changes should retain all existing behavior (resume safety, per‑trial checkpointing, config checks) but significantly reduce the cognitive load in `_run` and make the resume logic easier to maintain.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}