Forem: Takayuki Kawazoe

Claude Code Skills vs deterministic verify commands — same checks, very different ergonomics

Takayuki Kawazoe — Sun, 03 May 2026 08:05:21 +0000

A peer showed me their verification pipeline last week. We were both running coding agents over real tickets, both wanted "did the agent actually finish the work" to be a hard gate, and we both had a thing for it. Theirs was a Claude Code Skill — a Markdown file with a few prompt fragments and tool permissions, invoked by the agent itself when it thought verification was warranted. Mine was a deterministic shell command pinned into a workflow step that runs every time, no agent judgment involved. Same intent. Very different operational shape.

The conversation that followed — "why did you pick the one you picked, and where does it bite you" — is what this post is. I'm building an AI dev harness called Codens; what's relevant here is that its verification layer is on the command side of this divide, and my read on Skill-based verification is partly observational from peers running them and partly from the Skills I do have installed for developer-side work in the same repo. I'll be honest about which is which.

What "Skill-based verification" looks like

A Skill in Claude Code is a Markdown-defined behavior contract. It lives in .claude/skills/<name>/SKILL.md (project-scoped) or in the user's home directory, has a YAML frontmatter with a name and a description, and a body that's free-form prompt text plus optional references to scripts, sub-agents, or tool permissions. The agent reads the available Skills' descriptions on every turn and decides — in the same way it decides whether to call a tool — whether the current situation warrants invoking one.

The shape of a verification Skill is something like:

name: verify-task
description: |
  Run after implementing a Notion ticket. Checks that lint, types,
  and unit tests pass for the touched code paths, and reports back
  in plain English with a recommended next action.

# Verify Task

When invoked, do the following:
1. Read the diff and infer which packages were touched.
2. For each touched package, run the appropriate lint/type/test commands.
3. If everything passes, return a one-line summary.
4. If something fails, summarize the smallest reproducer, link to the
   failing line, and suggest one concrete fix.

The strengths are what you'd expect from giving an LLM a verification rubric instead of a script:

Contextual judgment. The Skill decides what to verify based on the diff. A pure-frontend change skips the backend test suite.
Natural-language failure explanations. When something breaks, the agent doesn't dump 1500 bytes of stderr — it summarizes. "The test test_normalize_uv_venv fails because the function now strips a trailing slash the test expects to remain."
Output that varies in shape with the run. A clean pass returns a one-liner; a messy failure returns a structured triage.
The agent decides if and when to call it. No engine logic to wire.

That last property is also where Skill-based verification gets a little scary. It depends on the agent reading the room correctly — exactly the property that makes me reluctant to put production gating on it.

What `verify_commands` looks like

The mechanism on the other side is what Codens uses internally. The unit of verification is a shell command (or a chain of them) declared on the ticket itself, and the workflow engine runs it as a step. No model in the loop, no judgment about whether to run, no judgment about how to interpret the result. Pass means exit 0. Fail means non-zero. Output is captured verbatim and the tail surfaces in the failure message.

The runner is small. Here's the actual thing, from purple-codens/backend/src/infrastructure/workflow/steps/run_tests.py:

test_command = config.get("test_command", "npm run test")
# Interpolate context variables (e.g., {verify_commands} from Notion)
try:
    test_command = test_command.format(**context.variables)
except (KeyError, ValueError):
    pass
# Build final command: prepend project-level defaults before task-specific commands.
# Each chunk runs inside its own subshell `(...)` so a `cd dir && ...` step
# in the project default does not leak its pwd into the task-level verify and
# cause the next `cd dir` to fail. The agent service runs the final string
# under /bin/sh (dash on Debian/Ubuntu), where a failed `cd` returns exit 2 —
# which RunTestsStep would otherwise surface as the misleading
# "Tests failed (exit code: 2)".
default_verify = context.get_var("default_verify_commands")
task_verify = context.get_var("verify_commands")
if default_verify or task_verify:
    parts = [f"({p})" for p in [default_verify, task_verify] if p]
    test_command = " && ".join(parts)

Three things worth pointing out:

Two layers chained. default_verify_commands is the project-level "always run this first" — typically format && lint && typecheck. verify_commands is the task-level — usually a targeted test that exercises the specific change. They concatenate with &&, so defaults must pass before task-specific runs.
Subshell wrapping. Each chunk is wrapped in (...) so a cd dir && ... in one part doesn't leak its working directory into the next. The kind of thing you only fix after staring at "Tests failed (exit code: 2)" in Slack and realizing dash exits 2 when cd fails.
The preset wires it. From notion_compatible.json:

{
  "id": "verify",
  "type": "run_tests",
  "config": {
    "test_command": "{verify_commands}",
    "timeout_seconds": 300
  },
  "on_success": "notify_success",
  "on_failure": "fix_verify"
}

When verify fails, the surfaced error includes the tail of the captured output — not just the exit code — so the next step has something concrete to feed back to the agent:

output = result.get("output", "") or ""
_ERROR_TAIL_BYTES = 1500
tail = output[-_ERROR_TAIL_BYTES:] if len(output) > _ERROR_TAIL_BYTES else output
exit_code = result.get("exit_code", "unknown")
if tail.strip():
    msg = f"Tests failed (exit code: {exit_code}). Last output:\n{tail}"
else:
    msg = f"Tests failed (exit code: {exit_code}); agent reported no output"
return StepResult.error(
    msg,
    {
        "test_output": output,
        "previous_step_output": output,
    },
)

previous_step_output is the variable the next claude_code step (fix_verify) interpolates into its prompt: "Verification failed. Fix the issues. Output: <last 1500 bytes>." The agent gets stderr to react to, but the decision about whether verification passed or failed is made by the engine reading the exit code — not by the agent reading the output.

Strengths, mirroring the Skill side:

It always runs. No "did the agent reach for the Skill" question. Step exists, step executes, exit code resolves.
Bool pass/fail. Engine routes on on_success or on_failure. No interpretation layer.
Structured failure output. A 1500-byte tail of real stderr, not an LLM's summary of it. Sometimes the LLM is wrong about what failed; the raw tail is not.
Drop into CI. The same command line runs in the workflow, in GitHub Actions, and on a contributor's laptop. Portable in a way a Skill is not.

Where each one is the better fit

Both mechanisms can do "verification." The interesting question is when each one is ergonomically the right tool, not just when it's technically possible.

Skill is the better fit when:

The check requires contextual judgment about what to look at. "Did the PR include sensible test coverage for the changed lines" is a Skill question — what counts as sensible varies by file, by convention, by whether the change is a refactor or new behavior. A command can't make that judgment without becoming an LLM under the hood.
The output is a report, not a gate. "Summarize the risk of this change" produces prose that varies with the run. Forcing it into a deterministic command's stdout feels backwards.
The check is opportunistic — useful when relevant, skippable when not. The agent deciding "this is a docs-only change, verify-task doesn't apply" is the right behavior. A command would have to bake that branch in.
The check is for the human reviewer. PR descriptions, change-impact estimates, "what else might this affect" — content for the conversation log, not a gate that blocks the PR from existing.

Command is the better fit when:

The check is a gate. The pipeline must not advance if it fails. There must be no path where the agent decided not to call it.
The check has a natural deterministic implementation. Linters, type checkers, test runners, schema diffs, contract tests against an OpenAPI spec — they exist already and have spent years getting their exit codes right. Wrapping them in a Skill adds an interpretation layer that only makes them less reliable.
You want one source of truth across local, CI, and the agent loop. The same npm run test:e2e runs in all three. If the Skill version wins, you now have two truth sources, and they will eventually disagree.
The failure output is load-bearing. When the agent gets routed back into a fix loop with "Verification failed. Output: <tail>," it needs the actual stderr, not a paraphrase. Paraphrases drop information the next turn needs.

A meta-property: commands compose with other commands; Skills compose less well with non-LLM tooling. If verify also gates Slack, the merge step, the deploy — that's engine-level routing, and the engine wants bool.

The traps each side hits

Skill side, failure modes I keep seeing in peers' setups:

The agent forgets to invoke. The Skill description matched ten previous turns; this turn the agent is mid tool-use chain and never reaches for it. No "did you remember to verify" hook — just the agent's running judgment, which is a function of the prompt window at that exact turn.
Skill bloat eats context. Each installed Skill's description sits in the context window every turn. A repo with twenty Skills pays twenty descriptions of attention budget for the chance one matches. Verification needs to be always-available, which means its description has to fit in the "always relevant" slot.
Hard to verify which version is running. Skills get edited; the agent reads the file at invocation time. Did the last fix actually take effect? With a command, the answer is git log -- path/to/script. With a Skill, it's "let me read the file and hope no one edited it in the last hour."
Self-reports are not gates. A Skill returning "verification passed" is an LLM telling you the LLM thinks it passed. Fine for a PR comment or triage note. Not fine for "should this merge to main." The pass-signal is structurally weaker than an exit code, and the only way to upgrade it is to put a deterministic check inside the Skill — at which point the Skill is just a shell wrapper around a command.

Command side, scars I have from each:

Feedback latency. Commands take minutes. A full verify chain (format → lint → typecheck → test) is slower. A Skill that says "looks fine" in three seconds feels great by comparison — until it's wrong.
Output-size overflow. Captured stdout can be megabytes. We tail to 1500 bytes for the error message; truncation can land mid-error. If the traceback is at the top of the output and the bottom is just pytest's summary, the tail you surface is the wrong tail.
Sloppy exit codes hide "why." Plenty of CLI tools exit non-zero for both "your code failed" and "I couldn't even start." cd nonexistent && pytest exits 2 from the failed cd, which the runner reports as "Tests failed (exit code: 2)" — same shape as a real test exit 2. The subshell-wrapping fix landed precisely because this confusion was costing afternoons.
No graceful "doesn't apply." A command that runs pytest backend/tests/ on a frontend-only PR runs the whole suite anyway, or fails because backend fixtures aren't set up on this branch. There's no built-in "skip if irrelevant" — either the command short-circuits itself, or the engine adds routing logic. Both are work the Skill gets for free from the agent's judgment.

My current division of labor (provisional)

In Codens, the production workflow is commands all the way down. Verify is a run_tests step. Conflict checks are a check_conflicts step. CI gating is a wait_ci_checks step. The agent (claude_code step) is the thing that makes the changes; everything that judges whether a change is acceptable is deterministic. The reasoning is the gating property — these checks must run, must pass before the next step, and must produce structured output that the next agent step can react to.

What I do have on the Skill side is developer-facing: the Skills in purple-codens/.claude/skills/ are test-generator, pr-creator, and bug-analyzer. These run when I'm working in Claude Code locally on the Codens repo itself — when I ask "add tests for this," the test-generator Skill picks up. They are not invoked by the production workflow that runs on customer code. So the honest answer to "does Codens use Skills" is "yes, for the human-in-Claude-Code experience inside the repo; no, for the autonomous workflow that ships to customers." Different mechanisms for different operating modes, even though the underlying tasks (generate tests, write PR descriptions, analyze bugs) are similar.

A first-pass mapping of what I've settled on for which mechanism does what:

Task	Mechanism	Why
Lint, type check, unit test	command	Gate. Already a tool with a good exit code. Composes with CI.
Run task-specific verify	command	Gate. Output is load-bearing for `fix_verify`.
Generate PR description	Skill	Report, not a gate. Wants prose. Varies with run.
Analyze bug from stack trace	Skill	Contextual. Output is for the human reading triage.
Estimate change impact across files	Skill	Judgment-heavy. Doesn't gate anything.
Conflict detection	command	Gate. Engine routes on result.
Test generation	Skill	One-shot, in-conversation. Not a workflow step.
Wait for CI checks	command	External signal, deterministic poll.

The pattern that fell out: gating belongs to commands, judgment-as-a-report belongs to Skills. When I find myself wanting "the gate decision should depend on the agent's judgment," that's usually a sign that the gate is wrong, not that I should put a Skill behind it. Either tighten the command so it fails on the right things, or stop trying to gate that property.

The unresolved question

Both mechanisms have escape hatches that, if I leaned on them, would blur the distinction.

On the Skill side, you could imagine a hook — "force the agent to invoke this Skill before exiting" — that turned the Skill into a mandatory step. That removes the "agent forgets to invoke" trap, but in exchange you've recreated the command. The Skill's contextual judgment now happens in a fixed slot, the engine routes on its output, and the only difference from a command is that the Skill body is Markdown instead of a shell script. It's not clear that buys you anything except a worse failure-output story.

On the command side, you could add "skip this command if X" routing — if diff_files match frontend/**, skip backend/run_tests — and recover some of the Skill's irrelevance handling. That works for a few cases, but the routing logic accumulates: skip if frontend-only, skip if docs-only, skip if dependency-only, skip if revert. Each new condition is a string of YAML or a few lines of Python. At some point you've hand-coded the agent's judgment into a rule engine, and you'd have been better off just letting the model decide.

So maybe the real shape is that "Skill vs command" is the two ends of a slider labeled how much judgment do I delegate to the model. All the way to the command end: zero delegation, the engine decides everything. All the way to the Skill end: full delegation, the model decides whether and how to verify. The interesting question isn't "which one" but "how much delegation does this specific check tolerate." Gates tolerate very little. Reports tolerate a lot. Most things sit somewhere on the slider, and the choice between Skill and command is really a choice about where on the slider to draw the line for that one check.

The version of this I haven't resolved: what's a check that genuinely benefits from being on the slider's middle, not at either end? "Verify-with-judgment" — the engine gates on a deterministic signal, but the interpretation of the signal for the next agent's prompt is generated by a Skill. That'd give you the gate's certainty and the Skill's prose. I haven't built it yet. I suspect when I do, the seam between the two halves — "the deterministic part decided pass/fail, the prose part explains why" — is where the next class of bugs lives.

Closing

If you've shipped both Skill-based and command-based verification in the same agent system and have a heuristic for when to use which, I'd want to hear your rule. My current rule — if it gates, it's a command; if it reports, it's a Skill — feels right but I've only been running it for a few months, and I haven't seen it under the load that breaks it. The peer who showed me their Skill-based pipeline was running it for a smaller, less gated workflow than mine, and we both came away unconvinced that either of us had the universally right answer.

The other thing I'm curious about is the middle of the slider: have you found a way to combine the two — deterministic gate plus agent-generated explanation — that doesn't decay into either "the explanation is unreliable so we ignore it" or "the gate is too rigid so we override it"? That's where I think the next interesting design lives, and I don't yet have the shape of it.

I shipped a regex to catch AI agents committing only half the work. Then I ripped it back out.

Takayuki Kawazoe — Sat, 02 May 2026 11:12:05 +0000

The agent committed the test file, pushed it, and reported success. The implementation file the test was supposed to exercise was not in the diff. Exit code 0. PR open. CI green. From every downstream system, the run looked clean. The only reason I caught it was that someone glanced at the PR diff before merging and said "wait, where's the actual change."

That moment, six months ago, is when I decided I needed something stronger than "the agent said it was done." This post is the story of what I built, why I kept patching it for half a year, and why I finally deleted the whole thing in a single commit: 911005fd, 742 lines deleted, 39 added, net -703, plus around 280 unit tests removed. The thing I deleted was a regex-based "diff shape guard" that tried to detect partial implementations by comparing the agent's diff against the file paths mentioned in the ticket. The thing I kept was an existing behavior-based check — verify_commands — that turns out to subsume everything the regex was trying to do, and a lot more.

I'm building an AI dev harness called Codens — happy to talk about it but it's not the point of this post. The point is the lesson: shape validation of AI output is a trap that compounds every time you try to fix it.

Background: what "partial implementation" looks like

When you hand a ticket to a coding agent, the failure modes that aren't "the agent crashed" are the ones that bite you. Exit 0, commits made, work incomplete in a specific way:

Test file written, implementation file missing.
Implementation written, migration missing.
Function signature updated, call sites in three other modules still pass the old shape.
Route handler added, OpenAPI spec the SDK regenerates from unchanged.

The agent is locally consistent in all of these — from inside its context window it did the natural next step. From outside, the PR is half a feature.

Our workflow launches Claude Code on a remote worker, hands it a Notion ticket, lets it create a branch, implement, commit, push. Then a verify step runs the ticket's verify_commands — a per-ticket shell snippet that says "task is done iff these commands pass." If verify fails, a fix_verify claude_code step gets the verify output piped into its prompt and tries again. The preset matters here because the whole article hinges on the relationship between develop and verify:

// backend/src/infrastructure/workflow/presets/notion_compatible.json (excerpt)
{
  "id": "develop",
  "type": "claude_code",
  "config": { "model": "opus", "max_turns": 30, "timeout_seconds": 1800 },
  "on_success": "verify",
  "on_failure": "notify_failure"
},
{
  "id": "verify",
  "type": "run_tests",
  "config": {
    "test_command": "{verify_commands}",
    "timeout_seconds": 300
  },
  "on_success": "notify_success",
  "on_failure": "fix_verify"
},
{
  "id": "fix_verify",
  "type": "claude_code",
  "config": {
    "model": "opus",
    "prompt_override": "Verification failed. Fix the issues.\nOutput:\n{previous_step_output}",
    "max_retries": 2
  },
  "on_success": "verify",
  "on_failure": "notify_failure"
}

verify_commands is the behavior-based check: did the thing the ticket described actually happen, judged by running real tests against real code. Slow, expensive (seconds to minutes), semantically strong.

What I added on top of it was a shape-based check: did the diff at least touch the files the ticket mentioned. Cheap, fast, runs immediately after develop. The intuition was "if the ticket says modify src/notion_sync.py and the agent's diff doesn't touch it, fail fast — don't even bother running verify."

The intuition was wrong in a way that took me half a year to fully accept.

What the shape guard actually was

The mechanism had three pieces, all in backend/src/utils/expected_files.py. The first was a regex that pulled file-shaped substrings out of the ticket body:

# backend/src/utils/expected_files.py (DELETED)
# Matches relative file paths with known source-code extensions.
# Accepts optional leading directory segments (lowercase, hyphens, underscores,
# digits) followed by a filename that may begin with uppercase or digits.
_FILE_PATH_RE = re.compile(
    r"(?:[a-z][a-z0-9_-]*/)*[a-zA-Z0-9][a-zA-Z0-9_.-]*\.(?:py|tsx?|md|sh|ya?ml)"
)

The extension allowlist is restrictive on purpose — .docx, .png, .csv would have generated noise without ever being something the agent should be modifying.

The second piece was the cover-check, which used a deliberately loose suffix match so that a ticket saying "modify notion_sync.py" would still match a diff that included src/infrastructure/notion_sync.py:

# (DELETED) _diff_covers_expected
def _diff_covers_expected(
    diff_files: List[str],
    expected_files: List[str],
) -> Tuple[bool, List[str]]:
    """Return (all_covered, missing_files)."""
    if not expected_files:
        return True, []
    missing = [
        ef
        for ef in expected_files
        if not any(
            df == ef
            or df.endswith("/" + ef)
            or ef.endswith("/" + df)
            for df in diff_files
        )
    ]
    return len(missing) == 0, missing

The third piece was the call site inside claude_code.py, sitting right after develop finished and right before the no-op detection:

# backend/src/infrastructure/workflow/steps/claude_code.py (DELETED block)
if exit_code == 0 and repos_committed_set and step_type != "resolve_conflicts":
    _guard_body = (
        context.get_var("notion_body", "")
        or context.get_var("task_goal", "")
        or ""
    )
    _guard_expected = _extract_expected_files(_guard_body)
    if _guard_expected:
        _guard_diff_raw = result.get("diff_files")
        if _guard_diff_raw is not None:
            _guard_diff = [
                f.strip() for f in _guard_diff_raw.split(",") if f.strip()
            ]
        else:
            # worker did not supply diff_files — fall back to the agent's
            # marker block, then to scanning stdout for path-shaped substrings.
            _guard_diff = (
                _parse_modified_files_marker(output)
                or _extract_expected_files(output)
            )
        _guard_ok, _guard_missing = _diff_covers_expected(
            _guard_diff, _guard_expected
        )
        if not _guard_ok:
            return StepResult.error(
                "Expected files missing from diff: "
                + ", ".join(_guard_missing)
            )

The first time it caught a real bug — a ticket that mentioned src/notion_sync.py::_normalize_uv_venv where the agent had only created the test file — it felt like the right investment. A cheap regex was catching a class of error that would otherwise have wasted minutes of verify_commands time and produced a misleadingly-green run. That single early win paid for the rest of the half-year.

It also paid for what I now think were four bad bets, each of which I patched my way through before finally giving up.

Reason 1 we ripped it out: half the tickets had no file paths to extract

The first crack was statistical. After a month in production, roughly half of all tickets did not mention a single concrete file path. They were tickets like "The reaction bar feels sluggish on mobile, tighten it" or "When the agent fails three times in a row, send a Slack alert." Normal tickets — they describe the outcome, not the implementation surface area. The guard's response, by deliberate design, was to skip:

# (DELETED docstring)
"""Returns an empty list when no paths are found, which causes the shape guard
to skip (false-positive avoidance for tickets that don't mention specific files)."""

That skip was the only safe response — firing on expected_files=[] would have failed every outcome-style ticket. But the consequence is that the guard only protected runs where someone had bothered to write file paths into the description. The other half ran with no protection at all. I had built a safety net with holes the size of normal human ticket-writing, and the math on that was bad before I even hit the next three problems.

Reason 2: the regex caught file paths in "References" and "Notes" sections too

The second crack appeared when teams started writing more structured tickets — the kind with sections for "Implementation," "References," "Test Plan," "Notes." The regex pulled file paths out of all of them. So a ticket that said "implement X in foo.py, and as background see how bar.py does it" would set expected_files = ["foo.py", "bar.py"]. The agent would correctly modify foo.py, leave the reference file alone, commit, and the guard would fail the run because bar.py wasn't in the diff.

The patch was a section-aware preprocessor. Strip out the headings I didn't want the regex to scan, then run the regex on what's left:

# (DELETED) _EXCLUDED_SECTION_TITLES
_EXCLUDED_SECTION_TITLES = (
    "verification checklist",
    "verify",
    "verification",
    "test plan",
    "tests",
    "テスト",
    "検証",
    "checklist",
    "references",
    "参考",
    "notes",
    "備考",
)

Plus a markdown heading walker that propagated exclusion to subsections:

# (DELETED) _strip_excluded_sections
_HEADING_RE = re.compile(r"^(#{1,6})\s+(.+)$", re.MULTILINE)

def _strip_excluded_sections(text: str) -> str:
    if not text:
        return text
    headings = list(_HEADING_RE.finditer(text))
    if not headings:
        return text

    excluded: List[bool] = [False] * len(headings)
    excluded_depth: int | None = None
    for i, match in enumerate(headings):
        depth = len(match.group(1))
        title = match.group(2).strip().lower()
        if excluded_depth is not None and depth <= excluded_depth:
            excluded_depth = None
        if excluded_depth is not None:
            excluded[i] = True
        elif title in _EXCLUDED_SECTION_TITLES:
            excluded[i] = True
            excluded_depth = depth
    # ... rebuild kept text from non-excluded chunks ...

This worked in the narrow sense. The next week, a ticket came in with a "## Background" section. Then "## 関連 PR" (Related PR). Then "## 過去の議論" (Prior Discussion). Then English variants the original list missed: "## Prior Art," "## Context," "## Linked Issues."

Every new convention any engineer adopted for organizing tickets now required me to add a string to _EXCLUDED_SECTION_TITLES, in two languages — or to go educate the engineer about which section names my regex understood. The ticket-writing humans did not exist to make my regex work. They were communicating intent to an agent that had no trouble understanding what "Background" meant. Only the regex did. Every new exclusion added complexity while reducing scope. Each patch was net-negative on actual protection.

Reason 3: the agent's stdout fallback broke when the stdout got truncated

The shape guard had two ways to know what the agent had touched. The first was result["diff_files"], a comma-separated list of paths from the worker, derived from git diff --name-only after the agent finished. The second was a fallback that scanned the agent's stdout for path-shaped substrings, used when the worker hadn't yet been upgraded to return diff_files:

# (DELETED fallback)
_guard_diff = (
    _parse_modified_files_marker(output)
    or _extract_expected_files(output)
)

The fallback worked until the stdout got long enough to hit our truncator. Claude Code can produce thousands of lines of tool-use output. We truncate captured output to a fixed byte budget. When that truncator landed mid-path — src/utils/expe... instead of src/utils/expected_files.py — the regex stopped matching, the fallback returned an incomplete list, and the cover-check failed against a perfectly good diff.

The patch was to ask the agent to emit a deterministic marker block at the end of its output:

[PURPLE-MODIFIED-FILES]
src/utils/expected_files.py
src/infrastructure/workflow/steps/claude_code.py
[/PURPLE-MODIFIED-FILES]

Parsed by _parse_modified_files_marker, which survived the cleanup because it's still useful for diagnostics:

# backend/src/utils/expected_files.py (kept)
_MODIFIED_FILES_MARKER_RE = re.compile(
    r"\[PURPLE-MODIFIED-FILES\](.*?)\[/PURPLE-MODIFIED-FILES\]",
    re.DOTALL | re.IGNORECASE,
)

The marker survives truncation because it's emitted near the end of the run. It also handles paths the regex couldn't — Next.js dynamic routes like app/[locale]/page.tsx were tripping _FILE_PATH_RE because of the bracket segments.

But the agent doesn't emit the marker reliably. When max_turns is exhausted mid-tool-use, when the run hits a timeout, when stop_reason ends as tool_use rather than a final assistant message — the summary block at the end never gets written. So now I had a fallback for the fallback: the original regex scan applied to the truncated stdout I'd just established was unreliable. Three tiers of input source, each with its own failure modes, and the guard's behavior was a function of which tier happened to be active for any given run. I was spending more time reasoning about the guard's failure modes than the guard was saving me from.

Reason 4: the lockfile + no-op marker collision broke the design assumption

This is the one that finally killed it.

Some of our verify_commands include npm install to make sure dependencies are present before running tests. There's also a marker convention for "this task is already done":

# backend/src/infrastructure/workflow/steps/claude_code.py (kept)
NOOP_VERIFIED_MARKER = "[PURPLE-NO-OP-VERIFIED-OK]"

If the agent emits that marker, the workflow treats the run as a verified no-op — no commits expected, no diff expected, success.

The collision: when the agent runs npm install as part of checking whether the task is already done, npm regenerates package-lock.json. The agent's session honestly concludes "nothing to do here" and emits [PURPLE-NO-OP-VERIFIED-OK]. But the worker's commit hook commits the regenerated lockfile because it's the one mechanical change that happened during the session.

Post-run state: repos_committed_set = {"frontend"} because of the lockfile, [PURPLE-NO-OP-VERIFIED-OK] is in the output, ticket says "fix the ReactionBar timing." The shape guard fires, looks for ReactionBar.tsx in the diff, finds only package-lock.json, stops the run with Expected files missing from diff: ReactionBar.tsx.

Nothing has gone wrong. The agent correctly determined the work was already done. The lockfile is an unrelated side effect of the verification step. The guard is producing a false positive because its assumption — "if there are commits, the diff should match the ticket" — doesn't survive the case where commits exist for reasons orthogonal to the ticket.

The patch was to short-circuit the guard when the no-op marker is present:

# backend/src/infrastructure/workflow/steps/claude_code.py (current)
if exit_code == 0 and repos_committed_set and step_type != "resolve_conflicts":
    # When [PURPLE-NO-OP-VERIFIED-OK] is present but diff_files is unavailable
    # (Option B above cannot clear repos_committed_set), re-route to the no-op
    # path so the verified-no-op handling takes effect correctly.
    if self._has_already_done_marker(output):
        logger.info(
            "ClaudeCodeStep: [PURPLE-NO-OP-VERIFIED-OK] marker present for "
            "run=%s step=%s — re-routing to no-op path",
            context.run_id,
            context.step_id,
        )
        repos_committed_set = set()
        result["repos_committed"] = ""

Adding that branch is the moment I lost faith in the design. The shape guard, whose job was to be an external observer of the agent's output, was now branching on the agent's self-report. The architectural premise had been "we don't trust the agent to tell us if the work is done — we look at the diff." Trusting the marker for the no-op case was the exact kind of trust I'd set out to avoid by writing the guard in the first place. I was paying a maintenance tax on a thing that no longer had its original justification.

The behavior-based alternative was already in the workflow

Once I stepped back and asked what verify_commands actually catches, the answer was: everything the shape guard was trying to catch, plus more.

Wrote the test, didn't write the implementation: pytest fails on import error or assertion.
Wrote the implementation, didn't write the migration: the app fails to start with a schema mismatch.
Updated the function signature, didn't update the callers: mypy catches it. Or the test does.
Wrote the route, didn't update the OpenAPI spec: the SDK regen step in verify_commands fails.

The run_tests.py step is small in the current code. The interesting bits are the variable interpolation, the merging of project-level defaults with task-specific verify, and the subshell wrapping that prevents cd leakage between chunks:

# backend/src/infrastructure/workflow/steps/run_tests.py (current)
test_command = config.get("test_command", "npm run test")
try:
    test_command = test_command.format(**context.variables)
except (KeyError, ValueError):
    pass
# Build final command: prepend project-level defaults before task-specific commands.
# Each chunk runs inside its own subshell `(...)` so a `cd dir && ...` step
# in the project default does not leak its pwd into the task-level verify and
# cause the next `cd dir` to fail. The agent service runs the final string
# under /bin/sh (dash on Debian/Ubuntu), where a failed `cd` returns exit 2 —
# which RunTestsStep would otherwise surface as the misleading
# "Tests failed (exit code: 2)".
default_verify = context.get_var("default_verify_commands")
task_verify = context.get_var("verify_commands")
if default_verify or task_verify:
    parts = [f"({p})" for p in [default_verify, task_verify] if p]
    test_command = " && ".join(parts)

Two details worth flagging: the subshell wrapping and the && chaining. The subshell exists because dash (the default /bin/sh on Debian/Ubuntu) propagates cd failures as exit 2, which used to surface as a confusing "Tests failed (exit code: 2)" error that had nothing to do with the actual test outcome. Wrapping each chunk in ( ... ) keeps cd scoped to its own chunk. The && chaining means project defaults run first and task verify only runs if defaults pass — the right shape for "format → lint → typecheck → test."

When verify fails, the surfaced error includes the tail of the captured output, not just the bare exit code:

output = result.get("output", "") or ""
_ERROR_TAIL_BYTES = 1500
tail = output[-_ERROR_TAIL_BYTES:] if len(output) > _ERROR_TAIL_BYTES else output
exit_code = result.get("exit_code", "unknown")
if tail.strip():
    msg = f"Tests failed (exit code: {exit_code}). Last output:\n{tail}"
else:
    msg = f"Tests failed (exit code: {exit_code}); agent reported no output"
return StepResult.error(
    msg,
    {
        "test_output": output,
        "previous_step_output": output,
    },
)

previous_step_output is the variable fix_verify's prompt_override interpolates. When verify fails, the next claude_code step receives the actual stderr tail in its prompt: "Verification failed. Fix the issues. Output: ." Agent reads what failed, edits, workflow loops back to verify, runs again. Two retries before giving up.

The shape guard was checking "did the diff include this filename." Verify is checking "does the codebase, after the agent's commits, satisfy the executable contract." The latter is strictly stronger. Anything shape caught, verify catches and hands the agent something concrete to fix in the next loop. Anything shape missed — outcome-style tickets, agent edits to unrelated files, lockfile drift — verify handles correctly because it doesn't care what the diff looks like, only whether the code passes.

The single case where verify is weaker is "the ticket has no verify_commands written," and the right response to that is to make verify_commands a required ticket field, not to maintain a 130-line regex module as a fallback safety net.

What was removed, what was kept

The deletion commit is 911005fd, footprint across five files: 39 lines added, 742 deleted. Net 703 lines gone. Around 280 unit tests deleted — regex variants for _extract_expected_files, heading-scope tests for _strip_excluded_sections, suffix-match patterns for _diff_covers_expected, two integration tests in test_claude_code_step.py that exercised the guard's fail and pass paths.

What survived in expected_files.py is just the marker parser, with a docstring rewrite to reflect that it's no longer a fallback for shape detection:

# backend/src/utils/expected_files.py (current, post-cleanup)
"""Modified-files marker parsing utilities (used for no-op diagnostics)."""

NOOP_VERIFIED_MARKER and _has_already_done_marker also stayed. They let the agent self-report a no-op in a way that's safely scoped: the worst case of trusting the marker incorrectly is that we treat a real-work run as a no-op, catchable by other guards downstream and by verify itself if there's something to verify. Self-reports are fine when the trust scope is fail-safe — when the failure mode of trusting wrongly is benign. They are not fine when the trust scope is "this guard runs partial-implementation detection," because then the failure mode of trusting wrongly is letting partial implementations through.

The residue I'd put in front of anyone considering an agent self-report marker: what's the worst thing that happens if the agent emits the marker dishonestly or by mistake? If the answer is "the workflow takes a safer path than it would have," ship it. If the answer is "the workflow skips a check that was supposed to catch real bugs," don't.

The general pattern: shape vs behavior validation of AI output

Every shape check is a proxy for some behavior check you actually care about. "The diff contains the filename" is a proxy for "the agent did the work the ticket described." "The output ends with [DONE]" is a proxy for "the agent reached a clean stopping condition." "The PR title matches feat:|fix:|chore:" is a proxy for "the change is conventionally classified."

Proxies are cheap and look like good engineering when you write them — fast, deterministic, no LLM in the loop, easy to test. But the moment you start patching them against false positives, you are in a race against the unbounded variety of human and agent expression, and you will lose. Each patch shrinks the proxy's scope while expanding its surface area. Eventually you're maintaining hundreds of lines that catch a smaller set of bugs than the underlying behavior check would catch on its own.

The lesson: if a behavior check exists and runs in your pipeline already, do not put a shape proxy in front of it. The shape proxy gives you false confidence ("the cheap check passed, so the run is probably fine") and false alarms ("the cheap check failed, so the run is probably bad") in roughly equal measure, while the behavior check has already done a strictly more accurate job. The proxy's only honest selling point is latency — fail fast before the expensive thing runs. For AI agent output that selling point almost never pays off, because the proxies are too leaky to safely fail-fast on.

If you find yourself naming a function _strip_excluded_sections on day three of building a check, take that as the signal that the check itself is in trouble.

A side note worth keeping

The [PURPLE-MODIFIED-FILES] marker is still useful for diagnostics even though we no longer use it as input to a guard. When a run does something surprising, having the agent's own list of files-it-thinks-it-touched in the captured output saves time during post-mortems. The pattern I now look for: code that's bad as a control input can be fine as an observability input, and the right move is to keep the data and delete the control logic.

Closing

If you've shipped a similar diff-shape guard for AI agent output and later stripped it back out, I'd love to hear what tipped you over. My suspicion is that the false-positive curve looks roughly the same shape for everyone — early easy wins, then a plateau where every patch shrinks the cohort of tickets the guard protects, until the cohort is small enough that the maintenance cost dominates. The version of this question I keep asking is: is there a shape check that doesn't have this maintenance arc, or is shape validation of unstructured agent output just structurally bad and I should stop reaching for it?

The marker that did survive — [PURPLE-NO-OP-VERIFIED-OK] — survived because its trust scope is narrow and fail-safe. If you've shipped agent self-report markers with a wider scope that didn't decay into the same tar pit, I'd want to hear how.

3 MCP server failure modes that bit us in production, and how we ship around them

Takayuki Kawazoe — Sat, 02 May 2026 02:54:28 +0000

MCP feels easy until it isn't. The first time you wire up a stdio server and call a tool from a Claude Agent SDK loop, the whole thing fits on a slide. Then you put it in front of customer codebases, customer GitHub credentials, customer build containers, and the sharp edges show up in places the spec is silent on. Tools start shadowing each other. The agent confidently uses a built-in Read when you wanted it to go through your sandboxed file server. Environment variables you set on the parent process reappear inside the MCP child and become tokens-in-prompts.

I'm building a SaaS that uses MCP heavily across a few different services (Codens, an AI dev harness with several specialized agents — happy to talk about it but it isn't the point of this post). Across those services we have GitHub MCPs for repo reads, an in-process Playwright MCP for browser exploration, and per-workspace local-file MCPs that let an agent navigate a cloned repo without escaping the sandbox. Three failure modes have shown up enough to rewrite the way we build this stuff.

Below is each one with the code we actually ship, not a hypothetical fix.

Failure 1: built-in tools shadow your MCP tools, silently

The first time it bit us was in the test-case generation flow. The agent was supposed to be reading files only through a workspace-scoped MCP server — a thin wrapper that resolves paths, refuses to escape the workspace root, and truncates large files. Standard sandbox stuff.

@tool(
    "read_file",
    "Read file content from local workspace. Provide the relative file path.",
    {"file_path": str},
)
async def read_file_tool(args: dict[str, Any]) -> dict[str, Any]:
    file_path = args["file_path"]
    if file_path.startswith("/"):
        file_path = file_path[1:]
    full_path = workspace_path / file_path
    if not full_path.exists():
        return {"status": "error", "message": f"File not found: {file_path}"}
    try:
        full_path.resolve().relative_to(workspace_path.resolve())
    except ValueError:
        return {"status": "error", "message": "Access denied: path outside workspace"}
    # ... read + truncate ...

That relative_to check is the whole point of the wrapper. The agent should not be able to read /etc/passwd, or /app/... (where our backend code lives in the container), or anything else outside the cloned customer repo. The MCP server enforces the boundary; the agent only ever sees what the tool returns.

The bug: the Claude Agent SDK ships built-in tools — Read, Write, Edit, Glob, Grep, Bash — and unless you tell it otherwise, those built-ins are also available. The agent had two ways to read a file: the sandboxed mcp__local__read_file, and the unsandboxed built-in Read. We passed allowed_tools listing the MCP tools by full name and assumed that was an allowlist. It wasn't tight enough — built-ins kept being chosen.

The first sign was test cases that referenced files the agent couldn't possibly have seen if it was only reading from the customer's workspace. It was reading our own backend code from /app/ because the container Python path was sitting right there, and the built-in Read tool happily walked into it.

The fix is two-part, and both parts matter:

options = ClaudeAgentOptions(
    model=self._model,
    max_turns=40,
    system_prompt=system_prompt,
    cwd=str(self._workspace_path),
    mcp_servers={"local": self._mcp_server},
    allowed_tools=[
        "mcp__local__read_file",
        "mcp__local__list_directory",
        "mcp__local__get_file_structure",
        "mcp__local__get_multiple_files",
    ],
    disallowed_tools=[
        "Read",
        "Write",
        "Edit",
        "Glob",
        "Grep",
        "Bash",
        "MultiEdit",
        "NotebookEdit",
    ],
)

allowed_tools is necessary but not sufficient. The agent treats it as "these are allowed", not as "these are the only tools that exist." If a built-in is available and not explicitly in disallowed_tools, it stays in the toolbox. We also reinforce the boundary in the system prompt with text the model actually pays attention to:

=== CRITICAL FILE ACCESS RULES ===
1. You MUST ONLY use these MCP tools to access files:
   - mcp__local__read_file
   - mcp__local__list_directory
   - mcp__local__get_file_structure
   - mcp__local__get_multiple_files
2. NEVER use built-in tools like Read, Glob, Bash, or Grep
3. NEVER read files from /app/ - that is a different application
4. ALL file paths you read should be relative to the workspace

Note point 3 — explicit-path negation. We learned that one the hard way. The model has a tendency to "explore" by trying common system paths if it's confused about the workspace. Telling it NEVER read files from /app/ (the container path where our service actually runs) is dumb-looking but absolutely necessary to stop the model from leaking implementation details about the host into its analysis output.

Setting cwd=str(self._workspace_path) is the third leg. Without it, when a built-in did slip through during early debugging, it ran with the SDK's default working directory, which made the boundary violation worse. With cwd set, even an accidental tool use is at least scoped to the right tree.

If I were starting again, I'd treat the agent's tool list as a default-deny zone from line one. The mental model "built-ins exist; my MCP tools are added on top" is wrong for any production agent that needs sandboxing. Default-deny everything, then pass allowed_tools listing exactly the MCP tool names you want, then triple-belt-and-suspenders by enumerating the built-ins in disallowed_tools so a future SDK version that adds a new built-in (Read2, Search, etc.) doesn't quietly enable it for you. The cost of the extra few lines is nothing. The cost of an agent reading the wrong filesystem in production is everything.

Failure 2: tool name collisions across MCP servers, with no useful error

Once you have more than one MCP server in the same agent loop, you find out fast that "the same tool name in different servers" is a thing. Our regression test generator runs two MCPs simultaneously — a local server for code analysis and a playwright server for browser interaction:

mcp_servers: dict[str, Any] = {"playwright": playwright_server}
allowed_tools = [
    "mcp__playwright__browser_navigate",
    "mcp__playwright__browser_snapshot",
    "mcp__playwright__browser_get_links",
    "mcp__playwright__browser_click",
    "mcp__playwright__browser_type",
    "mcp__playwright__browser_wait",
    "mcp__playwright__browser_get_visited_pages",
    "mcp__playwright__browser_console_messages",
]
if self._local_mcp_server:
    mcp_servers["local"] = self._local_mcp_server
    allowed_tools += [
        "mcp__local__read_file",
        "mcp__local__list_directory",
        "mcp__local__get_file_structure",
    ]

options = ClaudeAgentOptions(
    model=self._model,
    max_turns=60,
    system_prompt=system_prompt,
    mcp_servers=mcp_servers,
    allowed_tools=allowed_tools,
)

Notice the prefixing convention: every tool name is mcp__<server>__<tool>. That's not cosmetic. It is the difference between "the agent knows which server to call" and "two read_file tools shadow each other and you find out at runtime which one it picks."

The collision case for us was real. Across our services, several different MCP servers each define a read_file tool — one for the workspace, one for GitHub repo reads, one inside the GitHub-tools server we use for fix generation:

# github-tools server: reads from a remote repo via the GitHub API
@tool(
    "read_file",
    "Read file content from GitHub repository",
    {"owner": str, "repo": str, "file_path": str, "branch": str},
)
async def read_file_tool(args: Dict[str, Any]) -> Dict[str, Any]:
    content = await github_client.get_file_content(...)

# local server: reads from the cloned workspace
@tool(
    "read_file",
    "Read file content from local workspace. Provide the relative file path.",
    {"file_path": str},
)
async def read_file_tool(args: dict[str, Any]) -> dict[str, Any]:
    full_path = workspace_path / file_path

Same name, different schemas, different intent. If both servers are mounted into the same agent loop and you list the tools as bare read_file, the SDK has to pick one, and which one it picks is not the kind of thing you want the LLM's prompt-time context to silently determine. Even worse, the agent will eventually try to call the other one and pass the wrong arg shape — {"owner": ..., "repo": ..., "file_path": ...} against the local server's {"file_path": ...} schema — and you get either a tool error the agent retries past, or, worse, an opaque "tool not found" that the agent reasons around by giving up on file access entirely.

The fix is to never refer to MCP tools by bare name. Always use mcp__<server>__<tool>. The system prompt gets paranoid too:

ONLY use mcp__local__ tools. Start with get_file_structure.

Use BOTH mcp__local__ (code analysis) and mcp__playwright__ (browser) tools.

The leading mcp__local__ prefix shows up in the system prompt because the model picks tools partly from the prompt's pattern matching. Saying "use the read_file tool" leaves it ambiguous; saying "use mcp_local_read_file" tells it both the server and the operation, and the model is much less likely to pick a built-in or the wrong server's tool.

There's a deeper point here about how you scope a MCP server. We deliberately scope by concern, not by protocol: there's a github server (operations against the GitHub API), a local server (operations against a cloned workspace), and a playwright server (operations against a live browser). Within each concern, tool names are short and don't repeat across concerns. We do not have one mega-server with 30 tools because at that point the system prompt has to disambiguate which tool to use within the same namespace, and that's a much harder prompt-engineering problem than just naming the servers right.

If I were starting again: pick a strict naming convention for your MCP servers (we use lowercase singular nouns: local, github, playwright, appium) and a strict prefix for every tool reference, both in allowed_tools and in the system prompt. Treat any unprefixed tool name in your code or prompts as a code smell. The 10 minutes you spend writing a small lint rule that flags read_file without mcp__<server>__ will save you hours of debugging "why did the agent call the wrong server" in production.

Failure 3: the parent process's environment bleeds into the MCP server

This one is the most interesting because the bug isn't in the MCP transport — it's in how MCP server subprocesses (and even in-process tools that hit external APIs) inherit the parent's process environment.

Concretely: our automatic-fix flow runs against customer repositories. Some customers configure their own AWS credentials so the fix-generation agent can run their tests in a sandbox. Those credentials get loaded by the use case and passed through to the SDK as extra env vars:

# generate_fix_use_case.py
extra_env: Dict[str, str] = {}
if project.credential_set_id:
    extra_env = await self._credential_use_case.get_env_vars_for_project(
        project_id=project.id,
    )
    if "AWS_DEFAULT_PROFILE" not in extra_env and default_profile:
        extra_env["AWS_PROFILE"] = default_profile

    if extra_env:
        logger.info(
            f"Injecting {len(extra_env)} extra env vars for project "
        )

So far so good. The agent needs AWS_* to talk to the customer's AWS, and we pass them in via extra_env. The bug is what happens when this collides with our own ANTHROPIC_API_KEY.

The naive way to merge extra env into the agent's environment — and the way we wrote it the first time — looks like this:

# WRONG. extra_env can overwrite ANTHROPIC_API_KEY.
env = {
    "ANTHROPIC_API_KEY": settings.ANTHROPIC_API_KEY.get_secret_value(),
    **(extra_env or {}),
}
options = ClaudeAgentOptions(model=self._model, env=env)

Spread order matters. With extra_env last, anything in extra_env overrides keys we set first. If a customer's credentials bundle happens to contain an ANTHROPIC_API_KEY (for whatever reason — they were experimenting, they have an internal Bedrock+Anthropic setup, they pasted the wrong env block into the credentials UI), that key now silently replaces ours in the agent's process. We'd be billing the agent's work to the wrong key, the customer would be billing the agent's work to their key, or — depending on what the value actually was — the SDK would fail authentication entirely and we'd see "401" errors that look like quota issues but aren't.

The fix is one line, but only because we now know it's a one-line fix:

# Create agent options — extra_env is merged first so ANTHROPIC_API_KEY cannot
# be overridden by user-supplied credentials
env = {
    **(extra_env or {}),
    "ANTHROPIC_API_KEY": settings.ANTHROPIC_API_KEY.get_secret_value(),
}
options = ClaudeAgentOptions(
    model=self._model,
    max_turns=10,
    env=env,
)

extra_env first, our key last. Now the customer's env can't shadow our key, no matter what they put in their credentials bundle. The comment on top of every one of these blocks (we have four of them across analyze_error, generate_fix, improve_fix_with_test_feedback, and analyze_and_fix_error_with_agent) is verbatim the same text, because if anyone ever "cleans up" the merge order in a future refactor, they need a giant flag in the diff that says no, this is load-bearing.

There's a wider principle here that's worth saying directly: anything you let into an MCP child process's environment is data you've handed to a place the LLM can read indirectly. If the MCP server prints diagnostics with env vars in them, those go to logs the model can see in some configurations. If the MCP server reads os.environ to pick up credentials and accidentally reflects them in an error message, those land in the tool result the model receives. Treat the env passed to an MCP server like you'd treat the request body of an internal API: minimum scope, validated keys, no secrets you wouldn't want the model to be aware of.

The other half of this fix is logging discipline. When extra_env shows up, we log only its size:

if extra_env:
    logger.info(
        f"Injecting {len(extra_env)} extra env vars for project "
    )

Not the keys. Not the values. A count. If a customer reports "the agent didn't see my AWS creds," the answer is in the count and a separate audit log of which credential set was selected, never in a debug dump of the env block itself. The same log line that says "we injected 4 vars" tells you the wiring is working without putting the secrets where any future debugger can grab them.

If I were starting again: I would build a build_env(*sources, base) helper from day one, with base always last in the spread and a hardcoded list of "system-owned" keys (the Anthropic key, our own internal service tokens) that simply cannot be overridden no matter what comes in *sources. The four spread-order copies in our codebase all do the same thing; they would do it once if I'd written a helper. They don't, because each one was added when its calling use case was added, and the merge-order discipline was learned at a time when the helper would have meant a refactor I didn't have time for. Every new place you spread env into an SDK options object is a place you can get the order wrong. Centralize it.

What these three failure modes have in common

The failure modes all live at the same seam: the boundary between the agent process and the things the agent process delegates to. With built-in shadowing, the agent has access to a tool you didn't want it to have because you didn't enumerate the disallowed set. With name collisions, the agent has access to two tools by the same name and resolves which one ambiguously. With env bleeding, the agent's children inherit context (credentials, paths, caches) that the parent didn't deliberately decide to pass through.

In each case the spec was technically silent. There's nothing in the MCP spec that says "you must shadow built-ins by listing them in disallowed_tools" because the SDK is the one that exposes built-ins, not MCP. There's nothing that says "always prefix tool names" because the MCP spec is per-server; the prefixing convention comes from how you wire multiple servers into one agent. Env inheritance is a Unix process detail that has nothing to do with MCP. But all three behave like MCP problems because they're invisible until the agent is running and you can't reason backwards from a "tool not found" or a "wrong API key" message to which seam was leaky.

The mental model that helped most was: the MCP server lifecycle is its own concern, not a free side effect of the agent process lifecycle. Just because your agent process was started with the right environment, the right tool list, and the right working directory doesn't mean the MCP children are running with that same shape. Every option you set on ClaudeAgentOptions — env, cwd, allowed_tools, disallowed_tools, mcp_servers, max_turns — has a "what does this look like from inside an MCP child" reading, and the production-correct answer is almost always more restrictive than the demo-correct answer.

Closing

If you've shipped a multi-MCP-server agent into production, I'd love to hear which of these bit you and which ones I haven't found yet. Especially: how do you handle MCP servers that have legitimate reasons to need real environment variables (a Stripe MCP that genuinely needs STRIPE_API_KEY, an internal-tools MCP that needs a service token), without giving up the "no parent env bleed" property? I have an answer that involves a small per-call env builder, but I'm half-convinced there's a cleaner pattern I haven't seen yet.

One SSE connection, N upstream agents — the stream relay we shipped

Takayuki Kawazoe — Fri, 01 May 2026 08:22:34 +0000

The pod that was tailing one of our long-running agents got rotated during a deploy. The client connection to the dashboard didn't disconnect — it sat there, EventSource happily believing the world was fine, with zero events coming through. The agent was still working. The events were still being emitted upstream. The middle box that was supposed to forward them to the browser was simply gone.

That moment is the reason this post exists. If you're going to put a relay between N independent event-emitting backends and one client-facing SSE connection, you have to answer a specific question: when the relay process dies, what guarantees that another one picks up where it left off, and that the client doesn't notice? A lot of "just proxy SSE" designs fall apart at exactly this seam.

I'm building a SaaS that touches this problem (Codens, an AI dev harness with a few specialized agents — happy to talk about it but it isn't the point of this post). The relay I'll walk through fans in live telemetry from N agent jobs running on independent nodes into a single per-run aggregated stream the dashboard subscribes to.

The shape of the problem

The setup, concretely:

N agent jobs run, each on its own host. Each exposes GET /jobs/{job_id}/stream and emits events for one workflow run only.
The dashboard wants a live feed for a specific run. It opens one SSE connection to our API, not N.
We have multiple instances of the relay process. Any can die at any time. Deploys rotate them constantly.
Reconnects are constant — laptops sleep, networks blip, tabs reload.

The naive shape that almost works: each browser opens an SSE connection through one of our API pods, and that pod opens an SSE connection upstream to the agent. Two-hop proxy, no state. Here's why it breaks:

Two browsers viewing the same run = two upstream connections. Hits the agent's per-job stream twice. Some agents fight back; some let you do it but dedupe is your problem.
Reconnects = a new upstream connection too. You either lose all events between disconnect and reconnect, or you require the upstream to support Last-Event-ID. Many homegrown SSE servers don't.
No durable buffer. A client offline for 10 seconds loses those events. No tape.
Per-pod independent tailers = N tailers per run with N pods. Wasted upstream connections, no single source of truth for "what events have we seen for run X."

The shape we ended up with: exactly one stream-relay pod owns the upstream connection for a given run at a time. That pod tails the agent's SSE, writes every event into a Redis Stream keyed by run, and the API layer reads from Redis Streams to serve clients. Many client SSE connections, one upstream connection, durable buffer in the middle.

┌──────────────┐      one connection per run     ┌─────────────────┐
│  Agent pod A │ ─────────────────────────────▶  │ stream-relay    │
└──────────────┘                                 │  pod (owner)    │
┌──────────────┐                                 │                 │
│  Agent pod B │ ─────────────────────────────▶  │ XADD events:{r} │
└──────────────┘                                 └────────┬────────┘
   ... N agent jobs ...                                   │
                                                          ▼
                                                 ┌─────────────────┐
                                                 │  Redis Streams  │
                                                 │  events:{run_id}│
                                                 │   (maxlen 10k)  │
                                                 └────────┬────────┘
                                                          │
                                  many client SSE conns   │
                              ◀───────────────────────────┘
                              (FastAPI sse_starlette pods)

The interesting design decisions are all about the middle box: who tails, how takeover works, how the API layer reads from the buffer, what counts as "alive."

Stateless relay, leased ownership

The relay process is stateless. It boots, looks at the database for active workflow runs with an active job, and tries to acquire a per-run ownership lock in Redis. If it gets the lock, it spawns an asyncio task that opens the upstream SSE and tails it. If another relay instance got the lock first, it doesn't spawn a tailer. Every 5 seconds, it scans again.

The main loop, abridged from stream_relay/main.py:

POLL_INTERVAL = 5.0  # seconds between active-run scans

while True:
    # 1. Refresh owned locks
    await ownership.refresh_all()

    # 2. Clean up finished tasks
    finished = [rid for rid, t in active_tasks.items() if t.done()]
    for rid in finished:
        try:
            active_tasks[rid].result()
        except asyncio.CancelledError:
            pass
        except Exception:
            logger.exception("stream-relay: tailer task error run_id={}", rid)
        del active_tasks[rid]

    # 3. Discover new running workflow_runs
    async with session_factory() as session:
        running_infos = await _find_running_runs(session)

    active_run_ids = {info.run_id for info in running_infos}

    # Cancel tasks for runs that are no longer active
    stale = [rid for rid in active_tasks if rid not in active_run_ids]
    for rid in stale:
        active_tasks[rid].cancel()

    # 4. Acquire + start new tasks
    for info in running_infos:
        if info.run_id in active_tasks:
            continue
        acquired = await ownership.acquire(info.run_id)
        if not acquired:
            continue
        task = asyncio.create_task(
            tail_stream(http, redis_client, info, ownership, streams_client),
            name=f"tail-{info.run_id}",
        )
        active_tasks[info.run_id] = task

    await asyncio.sleep(POLL_INTERVAL)

Three pieces are doing real work. (1) The lock refresh runs first so we don't lose ownership while we're busy. (2) The DB scan is the source of truth for "what runs should be tailed right now"; the relay never trusts its own in-memory state. (3) Acquisition is best-effort — if someone else got there first, we move on.

Stateless is deliberate. The relay doesn't keep a cursor file. There's no "what was I doing before I crashed" reconciliation. If a pod restarts, on its next 5-second tick it refreshes whatever locks it still holds (none, just booted), looks at active runs, and tries to acquire each. If the previous owner died, the lock has expired (60s TTL) and this pod takes over.

The instance ID is computed once at boot — ECS task ID, falling back to hostname, falling back to UUID:

_INSTANCE_ID = (
    settings.STREAM_RELAY_INSTANCE_ID
    or os.environ.get("ECS_TASK_ID")
    or socket.gethostname()
    or str(uuid.uuid4())
)

This matters because the ownership value stored in Redis is this instance ID. When checking "do I still own this lock," you compare the value, not just check existence.

The lock, in Redis, in nine lines

Every run_id gets one Redis key with a 60-second TTL. The owner refreshes it every 20 seconds. If the owner dies, the lock expires within 60s and another pod can grab it.

LOCK_TTL_SECS = 60
REFRESH_INTERVAL_SECS = 20
_KEY_PREFIX = "stream_owner"

async def acquire(self, run_id: str) -> bool:
    key = self._key(run_id)
    try:
        acquired = await self._redis.set(
            key,
            self._instance_id,
            nx=True,
            ex=LOCK_TTL_SECS,
        )
        if acquired:
            self._owned[run_id] = time.monotonic()
            return True
        # Check if we already own it (e.g. after a crash + restart)
        owner = await self._redis.get(key)
        if owner == self._instance_id:
            self._owned[run_id] = time.monotonic()
            return True
        return False
    except Exception:
        logger.exception("OwnershipManager.acquire failed run_id={}", run_id)
        return False

SET key value NX EX 60 is the entire trick. NX = only if not exists, EX = TTL in seconds. Atomic. The "already own it after a crash + restart" branch is the case where the hostname is stable across restarts and we want to recognize the lock as ours rather than wait out the TTL.

Refresh is defensive too. It checks the value before extending TTL, so if someone else has taken over we don't blindly re-extend their lock:

async def refresh(self, run_id: str) -> bool:
    key = self._key(run_id)
    try:
        owner = await self._redis.get(key)
        if owner != self._instance_id:
            logger.warning(
                "stream_owner: lock lost for run_id={} (owner={} expected={})",
                run_id, owner, self._instance_id,
            )
            self._owned.pop(run_id, None)
            return False
        await self._redis.expire(key, LOCK_TTL_SECS)
        self._owned[run_id] = time.monotonic()
        return True
    except Exception:
        logger.exception("OwnershipManager.refresh failed run_id={}", run_id)
        return True  # conservative: don't stop on Redis error

That return True on Redis error is deliberate. If Redis is briefly unreachable, we keep tailing rather than abandon a working stream. Redis comes back, next refresh confirms or denies. The downside: a brief window where two pods could both think they own a run during a Redis flap, but they're both writing to the same events:{run_id} stream — duplicates, not data loss. Duplicates the consumer can dedupe; data loss it can't recover.

The 60s TTL / 20s refresh ratio gives a 3x safety factor. The pod refreshes three times before TTL would expire; even one missed refresh is fine.

Pod rotation, in practice

The flow when a relay pod is rotated mid-tail:

Pod A is tailing run X. Its stream_owner:X Redis key has its instance ID and a 60s TTL, refreshed every 20s.
Pod A receives SIGTERM (deploy, scale-down, ECS task draining).
Inside relay_main, the shutdown handler cancels all active tasks, awaits them, closes Redis.
tail_stream's finally block calls ownership.release(run_id), which deletes the Redis key — but only if it still belongs to this instance.
Pod B, on its next 5-second poll, finds run X in the DB, tries to acquire, gets it, spawns a tailer.

async def release(self, run_id: str) -> None:
    key = self._key(run_id)
    try:
        owner = await self._redis.get(key)
        if owner == self._instance_id:
            await self._redis.delete(key)
    except Exception:
        logger.exception("OwnershipManager.release failed run_id={}", run_id)
    self._owned.pop(run_id, None)

The bad case — the one that brought me to this design — is a pod dying without running its shutdown handler. SIGKILL after a stuck SIGTERM. OOM. Network partition. The key still exists with its TTL ticking down; within 60 seconds it expires and Pod B's next poll picks it up.

Worst-case takeover: ~60s for TTL + up to 5s for the next poll = ~65s of no events relayed. The dashboard sees a gap. The events emitted during that gap aren't lost — they're still buffered in the upstream agent's own per-job queue, and the new tailer reads them on reconnect. (Whether that holds depends on your upstream; ours has a per-job in-memory replay buffer.)

We picked the boring 60/20 numbers because they're easy to reason about and a one-minute gap during deploys is tolerable. Shorter TTLs halve worst-case latency at the cost of more Redis traffic. The knobs are there.

The tailer: one upstream connection per run

The tailer is the asyncio task spawned for each owned run. It opens an httpx streaming GET to the agent's SSE endpoint and parses events:

async def tail_stream(
    http: httpx.AsyncClient,
    redis_client: aioredis.Redis,
    run_info: RunInfo,
    ownership: OwnershipManager,
    streams_client: RedisStreamsClient,
) -> None:
    run_id = run_info.run_id
    url = f"{run_info.endpoint_url.rstrip('/')}/jobs/{run_info.job_id}/stream"
    headers = _build_headers(run_info.api_key_hash)
    consecutive_errors = 0

    try:
        while True:
            # --- ownership check (before reconnect) ---
            if not await ownership.is_owned_by_us(run_id):
                return

            try:
                timeout = httpx.Timeout(connect=15.0, read=1800.0, write=15.0, pool=15.0)
                async with http.stream("GET", url, headers=headers, timeout=timeout) as response:
                    response.raise_for_status()
                    consecutive_errors = 0

                    check_deadline = time.monotonic() + CHECK_INTERVAL_SECS

                    async for event in _parse_sse_stream(response):
                        await streams_client.xadd(run_id, event)
                        # ... heartbeats + terminal-event handling ...

Two details that matter.

The httpx client is shared across all tailers in the same pod. We have a single httpx.AsyncClient and pass it into every tailer task. If you create one client per tailer, you blow your file descriptor budget at ~100 concurrent runs and lose connection pooling per host:

http_limits = httpx.Limits(
    max_keepalive_connections=200,
    max_connections=200,
    keepalive_expiry=None,
)
async with httpx.AsyncClient(
    timeout=None,
    verify=False,
    limits=http_limits,
) as http:
    # all tailers share `http`

The read timeout is 30 minutes. SSE streams can be quiet for a long time and that's normal. Set the read timeout to something sensible-sounding like 30 seconds and you get spurious ReadTimeout exceptions during legitimate slow phases of the agent's work. Learned this the hard way.

Reconnect logic is unsurprising — exponential backoff on connection/read errors, give up after 5 consecutive failures, release the lock:

except (httpx.ReadTimeout, httpx.RemoteProtocolError, httpx.ReadError) as exc:
    consecutive_errors += 1
except httpx.ConnectError as exc:
    consecutive_errors += 1
# ...
if consecutive_errors >= MAX_CONSECUTIVE_ERRORS:
    return

backoff = min(5 * 2 ** (consecutive_errors - 1), 60)
await asyncio.sleep(backoff)

5s, 10s, 20s, 40s, capped at 60s. After 5 failures we exit; the main loop's next poll re-acquires and starts fresh.

Two heartbeats: liveness vs forward progress

You'd think one heartbeat is enough. It isn't. The bug: an upstream agent stops making real progress (stuck on a tool call, the underlying LLM silently retrying internally) but its SSE stream still emits keepalive / ping events every 15 seconds because the agent's web framework wants the connection warm. If your stall detector reads a single "is the SSE alive" heartbeat, it sees green forever. The job is stuck and you don't know.

The fix is two heartbeats, read by different consumers:

LIVENESS_TTL_SECS = 300        # 5 minutes
RUN_PROGRESS_TTL_SECS = 3600   # 60 minutes

# SSE event types that indicate Claude is making forward progress.
_FORWARD_PROGRESS_TYPES = {"output", "result", "assistant", "tool_result"}

async for event in _parse_sse_stream(response):
    await streams_client.xadd(run_id, event)
    now_ts = str(time.time())

    # Liveness heartbeat: any SSE frame proves the connection is alive.
    await redis_client.setex(
        f"liveness_heartbeat:{run_id}",
        LIVENESS_TTL_SECS,
        now_ts,
    )

    # Forward-progress heartbeat: only real work advances this.
    if _is_forward_progress(event):
        await redis_client.setex(
            f"forward_progress_heartbeat:{run_id}",
            RUN_PROGRESS_TTL_SECS,
            now_ts,
        )

liveness_heartbeat:{run_id} advances on any SSE event including pings. It answers "is the upstream connection alive at all?"

forward_progress_heartbeat:{run_id} advances only on event types that mean the agent emitted something it worked for — output, result, assistant, tool_result. It answers "is the agent doing useful work, or idling?" If this is stale but liveness is fresh, the connection is fine but the agent is stalled.

A separate orphan-detection job reads the forward-progress key, not the liveness key. Mix them up and you get false positives during legitimate slow-thinking phases or false negatives during real stalls.

The buffer: Redis Streams with a hard cap

Every event the tailer parses is appended to a Redis Stream keyed by run:

class RedisStreamsClient:
    def __init__(self, redis_client: aioredis.Redis, prefix: str = "events") -> None:
        self._redis = redis_client
        self._prefix = prefix

    def stream_key(self, run_id: str) -> str:
        return f"{self._prefix}:{run_id}"

    async def xadd(
        self,
        run_id: str,
        data: dict[str, Any],
        maxlen: int = _STREAM_MAXLEN,  # 10_000
    ) -> Optional[bytes]:
        key = self.stream_key(run_id)
        try:
            entry_id = await self._redis.xadd(
                key,
                {"data": json.dumps(data)},
                maxlen=maxlen,
                approximate=True,
            )
            return entry_id
        except Exception:
            logger.exception("RedisStreamsClient.xadd failed run_id={}", run_id)
            return None

maxlen=10_000, approximate=True is the load-bearing flag. Without approximate=True, Redis trims exactly to 10k on every XADD — O(n) walk. With it, Redis trims at radix-tree-node boundaries: slightly fuzzy but vastly faster. Streams stabilize around 10k–11k entries, which is fine.

10k events per run is roughly half an hour of busy work for our agents. A client reconnecting within that window gets the entire history replayed from Redis. Beyond that, old events fall off; persist them separately if you need them long-term.

We use Streams (XADD / XREAD / XREADGROUP) and not just Pub/Sub precisely for this durability. Pub/Sub is fire-and-forget — no subscriber, event gone. Streams keep the bytes around so a client reconnecting after a blip replays from where it left off.

There's a layer above this in our codebase that bridges the buffered events into the per-client SSE response, including a history replay on connect so a late joiner sees prior events:

async def _subscribe_to_channel(channel: str, request: Request) -> AsyncGenerator[dict, None]:
    r = aioredis.from_url(settings.REDIS_URL, encoding="utf-8", decode_responses=True)
    pubsub = r.pubsub()
    try:
        # Subscribe BEFORE fetching history so we don't miss events published
        # between the LRANGE and the first get_message() call.
        await pubsub.subscribe(channel)

        history_key = f"sse:history:{channel}"
        history_events: list[str] = await r.lrange(history_key, 0, -1)

        last_replayed_seq: int | None = None
        for event_json in history_events:
            event_data = json.loads(event_json)
            seq = event_data.get("event_seq")
            if seq is not None:
                last_replayed_seq = seq
            yield {"event": "message", "data": event_json}

The order matters: subscribe before reading history. Otherwise events published between LRANGE and the first get_message() disappear. Live messages are then deduplicated against last_replayed_seq — every event carries a monotonic event_seq and the loop drops anything whose seq was already replayed. That seq is what gives the client a continuous-looking stream across pod takeovers, even if a reconnect briefly lands on a different API pod.

Backpressure on slow clients

The trap with SSE fan-out is that a single slow client (hotel wifi, throttled background tab, dev who left a curl pipe running for two days) piles up unsent events in the per-connection buffer.

The choices that protect the rest:

The relay never blocks on a client. It only writes to Redis. A slow client is the API pod's problem, not the relay's. The relay keeps tailing at full speed.
The Redis Stream is bounded. 10k entries cap. A reconnecting client gets at most 10k events; beyond that, the stream has aged out.
Per-connection sse-starlette buffers are small. Connections that fall too far behind get closed. We don't extend their grace period.

The single most important rule is the first: the upstream-tailing path is independent of any single client. Many pods serve SSE to many clients all reading from the same Redis Streams buffer. A slow consumer only slows itself.

When this is overkill

If you have one service emitting events, one client per run, no deploys, a short session bound, and reconnects are rare — a single-pod direct SSE proxy with no Redis in the middle is fine. Ship that.

The relay is justified by N agents × M clients × frequent deploys × long session lengths. Take any of the four out and the cost-benefit shifts. The specific moment where this design pays off is the rotated-pod-mid-stream case. If that never happens for you, most of the ownership-lock complexity is theatre. Keep the Redis Streams buffer for reconnect-replay; drop the leased ownership.

What I'd build differently

A real Last-Event-ID header path. Reconnect today uses a history-list replay keyed off the channel. That's fine for our common case but it's not the SSE-spec way. Cleaner: client sends Last-Event-ID on reconnect, the API seeks the Redis Stream to that ID. We will get there.

Fewer Redis round-trips per event. Every event is one XADD plus a SETEX for liveness, sometimes two more SETEXes for forward progress. Three to four Redis calls per event. A pipelined batch would cut latency. We haven't needed it yet, but it'll show up the first time someone points a debugger at the inner loop.

If you've shipped a multi-instance SSE relay, I'd love to hear which corner you cut. Especially: did you go with a leased single-tailer-per-run like this, or did you let every pod tail every run and dedupe at the consumer layer? I'm half-convinced the second design is simpler and I keep talking myself out of it.

Why we built our own credit ledger instead of using Stripe metered billing

Takayuki Kawazoe — Fri, 01 May 2026 07:08:30 +0000

A few months in, an agent I was running reserved about 1,000 credits' worth of tokens for a long-running task, then crashed mid-flight. The credits were sitting in a "reserved" bucket on the org. The user's available balance had gone down. The work hadn't actually happened.

If your billing layer is Stripe metered, you have an awkward conversation with yourself at this point. stripe.SubscriptionItem.create_usage_record(...) doesn't have a quantity=-1000 "actually never mind" path that respects your prior idempotency keys. Negative quantities exist, but they aren't the same primitive as a release; they show up as a separate usage record and they don't cancel the original reservation, because there was no reservation — Stripe metered doesn't have one. The model is: emit a usage event, it accrues, you bill at period end.

That moment is when I started writing my own credit ledger.

I'm building a SaaS that touches this problem (Codens, an AI dev harness with a few specialized agents that share an org-wide credit pool — happy to talk about it but it's not the point of this post). The point of this post is the architectural reason why Stripe metered, which is genuinely great at what it's designed for, doesn't fit a multi-product credit-pool product, and what the homegrown alternative actually looks like.

I'll keep this builder-to-builder. No pricing math, no markup talk, just the primitives.

What Stripe metered is great at

Stripe metered is designed around a clean model:

One product (or one subscription item per product).
A customer is on a subscription.
You report usage events as they happen.
At the end of the billing period, Stripe converts events into an invoice.
Pricing tiers (graduated, volume) are configured on the price object.

If you sell a single product priced per API call, per minute, per GB — this is the right answer. You should use it. You'll be done in a day, you'll get hosted invoices and dunning for free, and you'll never reimplement tax rates.

The problems start the moment you violate any of those assumptions.

Where it stops being a good fit

In my case, four assumptions were violated at once.

1. The pool is org-wide, not per-product

I have multiple agents (PRD writing, error auto-fix, test gen, activity ledger, and a workflow harness) that all draw from one shared credit balance attached to the customer's organization. There is no "Subscription for the test-gen agent" — there's an organization, and an org-wide pool that any agent can pull from.

You can model this in Stripe two ways and they both hurt. One subscription item per product means N usage streams to keep in sync, and the customer doesn't have a single balance — they have N usages that get reconciled at invoice time. One subscription item with a "credits" SKU means you've already started building a homegrown ledger; you've just put a thin Stripe wrapper on it.

The org-wide entity is the unit I actually want to lock and update atomically:

# billing-control-plane/backend/src/domain/entities/organization_credit.py
@dataclass
class OrganizationCredit:
    auth_org_id: UUID
    balance: Decimal
    reserved_balance: Decimal
    billing_paused: bool
    paused_reason: str | None = None
    updated_at: datetime | None = None

    @property
    def available_balance(self) -> Decimal:
        return self.balance - self.reserved_balance

Two columns: balance and reserved_balance. available_balance is just their difference. Every consume, reserve, refund, chargeback takes a SELECT ... FOR UPDATE on this row before doing anything. There is exactly one row per organization, and it's the truth.

You cannot get SELECT FOR UPDATE semantics out of Stripe. Stripe's API is eventually consistent from your perspective; you submit a usage event, you trust their backend to process it, you read it back via the dashboard or their API. That's fine for billing-period accounting. It's not fine for "did this agent's request just bring the org under their available balance, yes or no, right now."

2. Long-running tasks need reservations

Several of my agent operations take 30 seconds to 30 minutes. The naive "consume after the fact" pattern blows up: two concurrent requests both pass the balance check, both run, the second one was always going to overspend, and now I owe somebody an apology.

The fix is reservations. Before the work starts, reserve the credits. When the work finishes, consume the actual amount and release the rest.

# reserve_credit_use_case.py — abridged
credit = await self.credit_repo.get_by_auth_org_id_for_update(req.auth_org_id)
if credit is None:
    raise NotFound(f"organization {req.auth_org_id} not initialized")
if credit.billing_paused:
    raise BillingPaused(f"billing paused: {credit.paused_reason}")
if credit.available_balance < req.amount:
    raise InsufficientCredits(
        f"available={credit.available_balance} required={req.amount}"
    )

expires_at = datetime.now(UTC) + timedelta(seconds=req.expires_in_seconds)
reservation = Reservation(
    auth_org_id=req.auth_org_id,
    amount=req.amount,
    expires_at=expires_at,
    reserved_by_service=req.service_id,
    metadata=req.metadata or {},
)
reservation = await self.reservation_repo.create(reservation)

# balance is NOT changed; only reserved_balance moves
await self.credit_repo.update_balance(
    auth_org_id=req.auth_org_id,
    balance_delta=Decimal(0),
    reserved_balance_delta=req.amount,
)

The trick is that balance doesn't move during a reservation. Only reserved_balance does. So the org's total credit balance is unchanged ("the customer didn't lose money yet, the work hasn't been done"), but their available_balance (balance - reserved_balance) drops, which is what the next concurrent request sees. If a second agent tries to reserve more than what's left available, it fails before doing any work.

Then on success the reservation gets consumed for the actual amount, and the difference goes back:

# consume_reservation_use_case.py — abridged
new_balance = credit.balance - req.actual_amount
await self.credit_repo.update_balance(
    auth_org_id=reservation.auth_org_id,
    balance_delta=-req.actual_amount,
    reserved_balance_delta=-reservation.amount,
)

credit_back = reservation.amount - req.actual_amount

balance drops by the actual consumed amount. reserved_balance drops by the full reservation amount (so what got over-reserved is returned to available). One row update, atomic.

If the agent crashed instead, the matching call is release_reservation:

# release_reservation_use_case.py — abridged
await self.credit_repo.update_balance(
    auth_org_id=reservation.auth_org_id,
    balance_delta=Decimal(0),
    reserved_balance_delta=-reservation.amount,
)

balance unchanged, reserved_balance drops back. The credits are available again.

There is no Stripe metered primitive for this. The closest you can build is a "we'll send you a usage event later, maybe" pattern, which means the customer's available balance is computed in your code anyway, which means you're keeping a parallel ledger anyway, which means... you've built this.

3. Idempotency on retries needs payload checking, not just key checking

This is the one that bit me hardest in early prototypes. Every agent retries. If the network blips during a reserve call, the agent retries with the same idempotency key. You cannot double-reserve.

Stripe's Idempotency-Key header gets you a long way for Stripe API calls. But the key alone isn't enough for a credit ledger — you also need to verify that the payload is the same. If the agent reuses a key but with a different amount (almost always a bug), you should reject it loudly, not return the cached response for the wrong amount.

That's what the idempotency manager looks like:

# idempotency_manager.py — the core branching
async def acquire(
    self,
    auth_org_id: UUID,
    service_id: str,
    idempotency_key: str,
    payload_hash: str,
    lease_seconds: int = DEFAULT_LEASE_SECONDS,
    retention_days: int = DEFAULT_RETENTION_DAYS,
) -> tuple[IdempotencyOutcome, IdempotencyKey | None]:
    existing = await self.repo.get_for_update(
        auth_org_id=auth_org_id,
        service_id=service_id,
        idempotency_key=idempotency_key,
    )
    now = datetime.now(UTC)

    if existing is not None:
        if existing.payload_hash != payload_hash:
            raise IdempotencyConflict(
                f"idempotency_key={idempotency_key} payload_hash mismatch"
            )
        if existing.status == "completed":
            return (IdempotencyOutcome.REPLAY, existing)
        if existing.status == "failed_terminal":
            return (IdempotencyOutcome.FAILED_REPLAY, existing)
        # in_flight
        lease = existing.in_flight_lease_until
        if lease is not None and lease > now:
            retry_after = max(1, int((lease - now).total_seconds()))
            raise IdempotencyInFlight(retry_after_seconds=retry_after)

The four states matter:

Completed with matching payload hash: return the cached response, do not touch balance. This is a successful retry of a prior success.
Completed with mismatched payload hash: throw. Same key, different payload is a bug.
In-flight, lease still valid: another worker is doing this right now. Tell the caller to retry after N seconds. This prevents two workers from racing on the same key.
Failed terminal: cached error response. This stops retry storms on permanently-bad requests (insufficient credits, malformed payload).

The key observation: payload hash is computed by the use case and includes only the fields that should make the request unique. For reserve, that's {amount, expires_in_seconds}. For consume, that's {service_id, operation_type, input_tokens, output_tokens, model_name}. If the caller resubmits with different content but the same key, the lock catches it before any balance changes.

def compute_payload_hash(payload: dict[str, Any]) -> str:
    """Deterministic SHA-256 hash of the canonical JSON payload."""
    canonical = json.dumps(
        payload, sort_keys=True, separators=(",", ":"), default=str
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

The whole acquire → mutate balance → complete sequence runs inside one DB transaction. If the use case crashes between acquire and complete, the row is left in_flight with a lease. Lease expires, next attempt takes over. No double-spend.

I genuinely cannot see how to express this in terms of Stripe usage records. They aren't the right primitive — they're append-only, conditionally cancellable, and the cancellation isn't tied to a payload hash.

4. Refunds at the credit level vs the invoice level

In Stripe-metered land, a refund is an invoice-level concept. You issued an invoice for $X based on accumulated usage; you can refund some or all of $X. That's fine for "we billed you wrong, here's your money back."

In credit-pool land, two different things happen and they're not the same:

Real money refund. The customer paid for credits, decided not to use the product, charges back or asks for a refund. The money flows back via Stripe. The credits should be removed from the balance. But if they already spent some of them, you have an AR situation.
Operational credit-back. A reservation was bigger than the actual consumption, so credits go back to available. No money moves at all. This happens hundreds of times a day per active org.

These are different transaction types in the ledger:

# credit_transaction.py
VALID_TRANSACTION_TYPES = frozenset(
    {
        "consume",
        "grant",
        "purchase",
        "refund",
        "chargeback",
        "reservation_consume",
        "reservation_release",
        "ar_settlement",
        "migration_apply",
        "migration_rollback",
    }
)

reservation_consume and reservation_release are operational. refund and chargeback involve real money. Stripe metered conflates these because Stripe is invoice-shaped, not credit-shaped.

The refund use case interestingly does not change balance at all — it logs the Stripe refund event for accounting and audit, because the refund itself happens in Stripe:

# refund_credit_use_case.py — abridged
# Refund: balance NOT changed, affects_balance=FALSE
tx = CreditTransaction(
    auth_org_id=req.auth_org_id,
    service_id=req.service_id,
    transaction_type="refund",
    amount=-req.amount,
    balance_after=credit.balance,
    reserved_balance_after=credit.reserved_balance,
    affects_balance=False,
    related_transaction_id=req.original_transaction_id,
    metadata={
        "stripe_refund_id": req.stripe_refund_id,
        "reason": req.reason,
    },
)

That affects_balance=False flag is a small thing that does a lot of work — it lets the audit trail include the refund event without double-counting it against the balance, because the chargeback/refund recovery happens via a separate flow.

5. The audit trail you want is row-level, not invoice-level

When a customer asks "why did our balance drop by 4,000 last Tuesday," what you want to be able to answer is: this reservation, by this service, for this operation, with this idempotency key, that resolved into this consume transaction. Then you want to be able to replay the exact event sequence.

I write every state-changing operation through three things in the same transaction:

The balance update on organization_credit.
A credit_transaction row that records the delta, the type, the related reservation/transaction, and metadata.
An outbox_event that downstream consumers (notifications, analytics, the customer's audit dashboard) can replay.

# from reserve_credit_use_case.py
await self.outbox_repo.create(
    OutboxEvent(
        aggregate_id=req.auth_org_id,
        event_type="credit.reserved",
        payload={
            "reservation_id": str(reservation.reservation_id),
            "auth_org_id": str(req.auth_org_id),
            "amount": str(req.amount),
            "expires_at": expires_at.isoformat(),
            "service_id": req.service_id,
        },
    )
)

The transactional outbox pattern is the boring right answer for "I need to atomically update state and emit an event." If event delivery fails, the state still moved consistently and the event will be redelivered. If state failed, no event went out. You can rebuild any downstream view by replaying the outbox.

Stripe's webhooks are excellent — but they're Stripe's events, not yours. They tell you "an invoice was paid" or "a usage record was created." They don't tell you "service yellow-codens reserved 1200 credits at 14:02 on behalf of operation x for org Y." That's the granularity an org admin actually wants in their audit log.

What we kept Stripe for: the actual money

The thing I am very deliberately not building is a payment processor. I am not handling card numbers, I am not implementing 3DS, I am not arguing with chargebacks. Stripe is excellent at all of that.

So Stripe still owns the seam where money becomes credits. The package-purchase flow has two stages:

# purchase_package_use_case.py — abridged
session = await self.stripe.create_checkout_session(
    stripe_customer_id=org.stripe_customer_id,
    stripe_price_id=package.stripe_price_id,
    success_url=req.success_url,
    cancel_url=req.cancel_url,
    metadata={
        "auth_org_id": str(req.auth_org_id),
        "package_id": req.package_id,
        "credits": str(package.credits),
    },
)

That's stage one — PreparePurchaseUseCase creates a Stripe Checkout session for a credit package. No balance changes. The customer gets sent to Stripe's hosted checkout.

Stage two runs from the Stripe webhook handler when payment actually clears:

# from ApplyPurchasedCreditsUseCase
settlement = await self._settle.execute(
    auth_org_id=req.auth_org_id,
    incoming_credit=req.credits,
    service_id=req.service_id,
)
net_to_balance = settlement.remaining_credit
new_balance = credit.balance + net_to_balance

await self.credit_repo.update_balance(
    auth_org_id=req.auth_org_id,
    balance_delta=net_to_balance,
    reserved_balance_delta=Decimal(0),
)

Stripe webhook arrives, dedup'd by stripe_event_id. AR settlement happens first (if the org owed money from a chargeback, that gets paid down before new credits land in the balance). Whatever's left becomes available balance.

Note the seam: purchase amounts go through Stripe, internal pool movements go through BCP. Stripe is the source of truth for "did this customer pay." BCP is the source of truth for "what's the org's available balance right now and what is each agent allowed to spend." Neither tries to be the other.

What it cost to build

This is the honest part. The credit-pool service is around 4-6 weeks of focused work to do well enough to trust in production:

Domain modeling: organizations, credits, reservations, transactions, AR, idempotency keys, outbox events.
The five core use cases (reserve, consume, consume-reservation, release, refund) plus the edge cases (chargeback, grant, AR settlement, expired-reservation cleanup).
Tests, especially concurrency tests for the lock + reservation paths.
The Stripe webhook surface — checkout success, payment intent, dispute, refund, all idempotent on stripe_event_id.
An admin surface to inspect balances and reservations during incidents.
A migration path from a previous, simpler "just deduct" model.

You are not going to ship this in a weekend, even with the schema fully designed. The concurrency tests alone took me longer than I expected, because deadlocks under SELECT FOR UPDATE ordering are subtle and the bugs only show up under load.

If this isn't core to what your product is, don't build it.

When Stripe metered is fine

To be specific about when the build vs buy answer flips back the other way — Stripe metered is the right answer if all of these are true:

One product, one usage stream per customer. Per-seat or per-API-call pricing.
You're OK with usage being consumed as it's reported (no reservations).
You're OK with the customer's "balance" being computed at invoice time, not in real-time.
Your retries can be made strictly idempotent by the Stripe Idempotency-Key header alone, without payload-hash protection.
You don't need to atomically move money between sub-products on the same account.
Your refunds are invoice-level ("we billed you wrong"), not credit-level ("here's some make-good credits for the outage").

If even one of those isn't true, you're looking at a hybrid where you keep some ledger state on your side anyway. At which point: the ledger you're keeping is the actual source of truth, and Stripe is the payment layer. Be honest about which one's the system of record.

Three things I wish I'd known sooner

Reservations are the load-bearing primitive, not consumes. I started by modeling consumption as a single atomic deduct. Reservations got bolted on later when the long-running-task problem hit. If I were starting again, I would build reserve/consume/release as the only three primitives from day one, and treat "synchronous consume" as a sugar over "reserve then immediately consume."
Payload-hash idempotency is non-negotiable. Key-only idempotency lets bugs through that you'll only catch in customer-support tickets. The day you accidentally retry the same key with a different actual_amount and silently return the cached result, you've corrupted a balance. Make it a hard error.
Keep the money seam thin and explicit. It's tempting to let your ledger know about Stripe customers, prices, invoices, taxes. Don't. The boundary is: Stripe collects money and tells you (via webhooks dedup'd by event ID); your ledger turns money-in into credits and bookkeeps every internal movement. The simpler that seam stays, the easier the next payment provider integration is when it inevitably comes up.

I'm building this credit ledger inside a larger AI dev harness called Codens — the homegrown billing layer is the "Billing Control Plane," and it's what lets the agents share an org-wide pool with the reservation/release semantics described above. LP: https://www.codens.ai/

If you've built a credit ledger that survived contact with real customers and a few chargebacks, I'd love to hear which corners you cut and which ones bit you. Or if you're staring down this build vs buy decision right now and want to compare notes — also good.

I let my AI agents be advisory-only. Here's the rules-first PR risk engine I shipped instead.

Takayuki Kawazoe — Thu, 30 Apr 2026 01:24:23 +0000

A pattern I keep seeing in early AI-in-the-SDLC teams: someone wires an LLM into the PR-review pipeline as a quality gate, the LLM marks one perfectly fine PR as "risky" two weeks in, the team lead overrides it once and grumbles about it twice, and within a month the AI gate is silently disabled.

You can't recover from that. Once a TL has spent a Friday afternoon explaining to engineering why the AI thinks their PR is dangerous when it isn't, "AI dev tools" become a punchline in their next 1:1 with the CTO. And the AI was probably right some of the time — you just lost the chance to find out which times.

I'm building a SaaS that touches this problem (Codens, an AI dev harness — happy to talk about it but it's not the point of this post). When I designed the PR risk evaluation service for it, I started with five non-negotiable design rules that I think apply to any AI-in-workflow product:

AI is advisory only. Never auto-blocks a merge.
The TL owns code quality. Not the AI.
OK needs no reason. NG needs one line. Don't ask humans to defend approvals.
Raw events, derived facts, and AI suggestions are stored separately. Audit trail must survive any single layer being wrong.
Configuration > code. Risk rules are repository-editable, not in source control of the AI service.

This post is the implementation of those constraints — what the engine actually looks like, with the code that ships in production.

The architecture: rules first, AI second

The flow for a PR walks through three stages, in order:

PR webhook
    ↓
[1] Rules engine          ← pure deterministic, no LLM
    ↓ rule matches + risk score
[2] Threshold check       ← rule-driven "is this high-risk"
    ↓ if high-risk
[3] AI advisory layer     ← LLM looks at the matched rules and PR diff,
                            offers an opinion, never blocks

The TL still presses the merge button. The AI's role is at most "raise a concern in a comment that the TL can dismiss." The rules engine's role is "tell the TL to look closer at certain files." Nothing fans out from the AI to anything that gates a deploy.

The rules engine: 4 match types, and that's it

Every rule has one of four match types. I deliberately resisted adding a fifth (regex was tempting; I kept saying no).

class RuleMatchType(Enum):
    FILE_PATH_GLOB = "file_path_glob"   # auth/**, billing/**, migrations/**
    KEYWORD        = "keyword"          # "permission", "PII", "encryption"
    LABEL          = "label"            # GitHub label "risk:high"
    FILE_EXTENSION = "file_extension"   # .sql, .tf, .pem

@dataclass
class RiskRule:
    pattern:    str
    match_type: RuleMatchType
    risk_area:  RiskArea          # "auth" / "billing" / "data" / etc.
    weight:     float             # 0.5, 1.0, 2.0

Why only four? Because the people writing these rules are repository owners, not the AI service author. If I add regex, the rules become unmaintainable for anyone who isn't a regex hobbyist. With glob + keyword + label + extension, you cover roughly 90% of useful "this PR touches a sensitive area" detection, and the remaining 10% probably needs a human anyway.

The evaluation is dead simple — fnmatch for the path glob, lowercased substring match for keywords, label lookup for labels:

async def assess_pr(self, pr: GitHubPullRequest) -> AssessmentResult:
    rulesets = await self.ruleset_repository.get_applicable_rulesets(
        project_id=pr.project_id,
        repository_id=pr.repository_id,
    )

    matches: list[RuleMatch] = []
    for ruleset in rulesets:
        for rule in ruleset.rules:
            if rule.match_type == RuleMatchType.FILE_PATH_GLOB:
                matches.extend(self._match_file_paths(pr, rule))
            elif rule.match_type == RuleMatchType.KEYWORD:
                matches.extend(self._match_keywords(pr, rule))
            elif rule.match_type == RuleMatchType.LABEL:
                matches.extend(self._match_labels(pr, rule))
            elif rule.match_type == RuleMatchType.FILE_EXTENSION:
                matches.extend(self._match_file_extensions(pr, rule))

    total_score = sum(m.weight for m in matches)
    is_high_risk = total_score >= self.high_risk_threshold  # default 1.0

Path matching uses fnmatch.fnmatch with ** collapsed to *:

def _match_file_paths(self, pr: GitHubPullRequest, rule: RiskRule) -> list[RuleMatch]:
    matches: list[RuleMatch] = []
    glob_pattern = rule.pattern.replace("**", "*")
    for file_path in pr.changed_file_paths:
        if fnmatch.fnmatch(file_path, glob_pattern):
            matches.append(RuleMatch(
                rule_pattern=rule.pattern,
                match_type=rule.match_type.value,
                risk_area=rule.risk_area,
                matched_value=file_path,
                weight=rule.weight,
            ))
    return matches

Up to this point: zero LLM calls. A PR touching migrations/0042_add_user_pii.sql with a risk:high label and "permission" in the title gets a deterministic score derived from the rule weights, and that score is reproducible by anyone reading the ruleset.

High-risk → precheck profile, not AI judgment

When is_high_risk is true, the next thing that happens is not an LLM call. It's a precheck profile lookup:

if is_high_risk and assessment.risk_areas:
    precheck_profile = await self.profile_repository.get_best_profile_for_areas(
        risk_areas=assessment.risk_areas,
        project_id=pr.project_id,
    )
    if precheck_profile:
        assessment.precheck_results = [
            PrecheckItemResult(
                item_description=item.description,
                is_required=item.is_required,
            )
            for item in precheck_profile.items
        ]

A precheck profile is a checklist. For each risk area, the team has authored a list of "if you touched this area, did you remember to do X?" items. Examples:

auth area: "Did you isolate OAuth scope changes into a separate PR?"
billing area: "Does the price calculation route through bcp_price()?"
migrations area: "Did you verify the migration is reversible with --dry-run?"

The PR author or reviewer ticks ✅ / ❌ + a one-line reason for ❌. The AI doesn't tick these. Ever. It can't, by design — the precheck profile is a human checklist, and the AI has no API to mark items.

If the engineer wants to know "is my migration reversible," they run --dry-run. If they want a second opinion, they ask the TL. The AI sees this whole exchange, but its participation is bounded.

Where the AI finally shows up

After the rules-first stages run, the AI gets to chime in. Two narrow things only:

1. Per-precheck-item advisory. The AI can attach a comment to a checklist item: "Looking at the related files, I think items 2 and 3 are the ones worth reviewing closely." It cannot mark them done. It cannot add new items. It can only annotate.

2. Anomaly detection. Things like "this PR is 4× larger than the median PR for this area" or "the test count went down by 30% but the implementation grew." Statistical pattern matching on top of historical data, surfaced as a comment.

@dataclass
class AISuggestion:
    suggestion_type: str       # "anomaly" | "precheck_advisory" | "scope_warning"
    severity:        str       # "info" | "warning"  ← never "blocking"
    description:     str
    confidence:      float     # 0.0-1.0
    is_dismissed:    bool      # TL can dismiss → never re-shown for this PR

The is_dismissed flag does a lot of work. A TL who decides the AI's anomaly detection is too noisy on a particular repo can dismiss its suggestions for a few PRs in a row, and they stop appearing. The AI gets quieter automatically when the team finds it unhelpful, instead of making them open a settings page to silence it.

This is, I think, the single most important UX move in any AI-in-workflow product: let the human turn it down without leaving the workflow.

Three-table audit log

The other constraint that paid off in practice was separating the storage of three things that look similar but aren't:

GitHub webhook
    ↓
raw_events                 (immutable, the bytes that arrived)
    ↓ parser
gate_events / pr_stats     (derived facts: "PR opened", "review submitted")
    ↓ AI evaluation (non-blocking)
ai_suggestions             (the AI's commentary on the facts)

Why three tables instead of one row-per-PR-event with everything denormalized:

A bug in the parser (the path from raw_events → derived) shows up as wrong facts. You can re-run the parser against raw_events and regenerate gate_events without touching the AI's history. Recoverable.
A bad AI prompt (the path from facts → suggestions) shows up as wrong opinions. You can re-run the AI evaluator with a corrected prompt and regenerate ai_suggestions. The facts are unchanged, the events are unchanged. Recoverable.
A compliance question ("show me what the AI told the TL on PR #4912") returns from ai_suggestions directly. The fact that something was suggested, the fact that it was dismissed, the fact that the TL still merged anyway — all separately queryable.

If you put it all in one table, your "the AI was wrong here" investigation becomes "we lost the original event because we overwrote it with the corrected derivation". You can't get back to ground truth.

This costs more storage (raw events are big) and a slightly more complex backfill story, but I'd take that trade every time over having to apologize to a customer with "we don't actually know what happened."

What I'd push back on if I were rebuilding from scratch

A 5th match type. I still want regex sometimes. I haven't added it because the day I do, three teams will write regexes, two will be wrong, and one will write a catastrophically slow one that DoSes the rule engine. If I do add it, it'll be safe_regex with a strict timeout and a max-length cap, and I'll fight the urge to allow lookarounds.

More AI in the audit suggestions. Current prompts are conservative on purpose — short context, explicit "advisory only" framing, low temperature. There's a part of me that wants to let the AI chase suspicious patterns more aggressively across the codebase. But every time I've drafted that more-ambitious prompt, my mental simulation ends with the same false-positive-kills-trust failure mode that started this post.

Three takeaways

An AI in your workflow that can be dismissed at the workflow level (one click, no settings page) survives much longer than an AI that requires reconfiguration to ignore. The dismiss button is the trust-preserving move.
Rules + thresholds + checklists do most of the gating work. AI is an excellent annotator for "these specific files in this specific PR are worth a closer look", but it's a bad replacement for "should this merge." Don't promote it past advisory.
Separate raw events, derived facts, and AI suggestions in storage from day one. It's cheap to do upfront and impossible to retrofit when you discover your AI was systematically wrong about a class of PRs for two months and you need to figure out where to draw the rollback line.

I'm building this engine inside a larger AI dev harness called Codens — 5 specialized agents that share a credit pool across PRD writing, error auto-fix, test gen, and engineering activity tracking. The risk evaluation in this post is the "Yellow Codens" service. LP: https://www.codens.ai/

If you've shipped an AI gate that survived team contact for >6 months, I'd love to hear what kept it alive. Or if you've shipped one that died from a single false positive — even more interested in that.

Stop AI from hallucinating E2E test selectors — code analysis + live browser exploration via Claude Agent SDK and 2 MCP servers

Takayuki Kawazoe — Wed, 29 Apr 2026 09:58:33 +0000

Generating E2E tests with an LLM sounds great in a demo. You hand a Playwright test spec to Claude, ask it to produce TypeScript code, and the toy app passes.

Plug it into a real codebase and the wheels come off immediately. The AI confidently generates await page.click('#login-button') for a project where the actual element is <button data-testid="auth-submit">Sign in</button>. Selectors are invented from common patterns ("most projects use #login-button") rather than read from your code or your DOM. Result: ~every generated test fails on first run, because the selectors are pure hallucination.

This post is the architecture I shipped to fix that — how I make the AI read the codebase and drive an actual browser before it writes a single line of test code, and why the implementation choices look the way they do.

The architecture in one diagram

The agent gets two MCP (Model Context Protocol) servers wired in parallel:

              ┌──────────────────────────────────┐
              │     Claude Agent SDK Client      │
              │                                  │
   ┌──────────┤   mcp_servers = {                │
   │          │     "local":      <SDK MCP>,     │
   │          │     "playwright": <SDK MCP>      │
   │          │   }                              │
   │          └──────┬───────────────────────────┘
   │                 │
   ▼                 ▼
[ workspace        [ Playwright API
  file read ]        (Chromium running) ]

The "local" MCP server exposes file-read / list-directory tools backed by the developer's repo. The "playwright" MCP server exposes browser-control tools — navigate, snapshot, click, type — backed by an actual chromium.launch() instance.

Critically, both MCP servers run inside the same Python process as the agent client. Anthropic publishes a Docker image for an "official" Playwright MCP, but I deliberately don't use it — more on that below.

Here's the wiring (Python, with the Anthropic claude-agent-sdk):

# infrastructure/external_apis/exploratory_agent_client.py
from claude_agent_sdk import (
    ClaudeAgentOptions,
    ClaudeSDKClient,
    create_sdk_mcp_server,
    tool,
)
from playwright.async_api import async_playwright

@tool(
    "browser_navigate",
    "Navigate to a URL in the browser.",
    {"url": str},
)
async def browser_navigate_tool(args: dict[str, Any]) -> dict[str, Any]:
    url = args["url"]
    await page.goto(url, wait_until="domcontentloaded", timeout=30000)
    return {"status": "success", "url": page.url, "title": await page.title()}

Each @tool decorator turns a Python function into an MCP tool the agent can call. The Playwright page object is captured in the closure, so the tool drives a single, persistent browser session that the agent navigates step by step.

Why in-process MCP, not Docker MCP

Anthropic ships mcr.microsoft.com/playwright/mcp as a Docker image. The "right" thing on paper is to spawn that container as a subprocess of the agent and let stdio do the talking.

In practice, my agent runs inside a Celery worker on ECS Fargate. Docker-in-Docker on Fargate is a tax: Fargate doesn't expose /var/run/docker.sock by default, the workarounds (sysbox, Docker rootless inside a container, side-cars) all add operational complexity, and you lose the ability to attach event listeners to the Playwright browser from the agent process.

In-process MCP via create_sdk_mcp_server() sidesteps all of this. The Playwright page is just a Python variable. I can attach page.on("console", ...) and page.on("request", ...) listeners outside the tools, and have the tools query that captured state:

console_messages: list[dict] = []
network_requests: list[dict] = []

page.on("console", lambda msg: console_messages.append({
    "type": msg.type, "text": msg.text,
}))
page.on("request", lambda req: network_requests.append({
    "url": req.url, "method": req.method, "resource_type": req.resource_type,
}))

@tool(
    "browser_console_messages",
    "Get console messages captured from the browser (errors, warnings, logs).",
    {},
)
async def browser_console_messages_tool(args):
    return {"status": "success", "messages": console_messages[-50:]}

When the agent generates a test and the test causes a JS console error, the agent can call browser_console_messages to see it and write an expect(consoleErrors).toEqual([]) assertion guarding against it. With Docker MCP this is much harder — the listener state lives in a different process from the tool implementation.

The 3-phase flow

Just letting the agent loose on "URL + workspace, generate tests" produces sloppy results. I prompt it through three explicit phases.

Phase 1: Code analysis

Tools available: read_file, list_directory (the local MCP).

The system prompt tells the agent to:

Identify the framework (Next.js, React, Vue, etc.) and routing pattern
Locate page components, forms, key UI flows
Extract data-testid / aria-label patterns this project uses
Output a JSON summary of detected_pages, detected_routes, detected_forms

The point of this phase is to make the agent discover this project's selector conventions before it generates any test code. If the project uses data-testid="page-login-submit" consistently, the agent learns the pattern; later, when it sees the actual DOM, it gravitates toward selectors that match the convention.

Phase 2: Browser exploration

Tools available: browser_navigate, browser_snapshot, browser_click, browser_type, browser_console_messages, etc. (the Playwright in-process MCP).

browser_snapshot is the linchpin. Returning the raw HTML of the page would burn hundreds of thousands of tokens for a single SPA. Instead, the tool returns the accessibility tree as a list of refs:

@tool("browser_snapshot", "Get accessibility tree snapshot ...", {"full_page": bool})
async def browser_snapshot_tool(args):
    snapshot = await page.accessibility.snapshot()
    elements = []
    ref_counter = [0]

    def extract_elements(node, path=""):
        if node is None: return
        role = node.get("role", "")
        if role in {"button", "link", "textbox", "checkbox", "radio",
                    "combobox", "listbox", "option", "menuitem", "tab"}:
            ref_counter[0] += 1
            elements.append({
                "ref": f"e{ref_counter[0]}",
                "role": role,
                "name": node.get("name", ""),
                "path": path,
            })
        for i, child in enumerate(node.get("children", [])):
            extract_elements(child, f"{path}/{i}")

    if snapshot:
        extract_elements(snapshot)

    return {
        "status": "success",
        "elements": elements[:100],
        "element_count": len(elements),
    }

The agent sees something like:

[
  {"ref": "e1", "role": "textbox", "name": "Email"},
  {"ref": "e2", "role": "textbox", "name": "Password"},
  {"ref": "e3", "role": "button", "name": "Sign in"}
]

Then it can issue browser_type {ref: "e1", text: "test@example.com"} and the tool resolves the ref back to the live element. This is the same pattern the official Playwright MCP uses, and it's the right tradeoff: dramatically less token usage, and the agent operates on stable, semantic refs instead of pixel coordinates or fragile CSS.

Phase 3: Test synthesis

Now the agent has:

(from Phase 1) The repo's selector convention (e.g. data-testid="page-{name}-{element}")
(from Phase 2) The actual ref/role/name for every element on every page it explored
(also from Phase 2) The console-error history during exploration

I hand it all of that and ask for Playwright TypeScript code. The output is qualitatively different from a 1-shot generation:

Selectors match the project's convention instead of generic #button
Selectors target real DOM that exists, because the agent saw them in browser_snapshot
Console-error checkpoints get encoded as expect assertions
Steps that produced visible network errors during exploration become explicit await page.waitFor* calls

Lessons after running this in production

Three things that aren't obvious from the diagram:

1. Show the AI both the source code AND the live DOM. Spec-only prompts will hallucinate selectors. Code-analysis-only will guess at runtime behavior. Browser-exploration-only will ignore project conventions. You need both, fed in as separate phases so the agent has enough context to bridge them.

2. In-process MCP beats Docker MCP for a cloud-deployed agent. It's an operational convenience as much as a technical one — you keep the Playwright page object accessible from the same Python process that runs the agent, you can attach event listeners outside tools, and you don't fight Fargate over Docker-in-Docker.

3. Accessibility-tree refs are the right token-economy move. Returning raw HTML in browser_snapshot would 10x the cost per exploration and isn't even the most useful representation. Returning a flat list of {ref, role, name} is faster, cheaper, and aligns the agent's "click this element" mental model with what's stable across DOM changes.

What I'm still working on

The 100-element cap in browser_snapshot is a heuristic — pages with massive product grids overflow it and the agent can miss interactive elements past element 100. I'd like to make it pagination-aware (let the agent ask for "elements 100–200" if it suspects something is missing) without making the prompt longer. Suggestions welcome.

The agent is also still bad at flows that span an email round-trip (signup → magic link → continue). MCP-driven email-inbox access could close that gap, but I haven't built it yet.

This pattern is shipped in Codens — a workflow-based AI harness for software development that I've been building solo at Corevice for the last 6 months. The exploratory E2E generator above is the "Blue Codens" service, one of 5 specialized agents that share state across one workflow (Green for PRD, Purple for orchestration, Red for auto-fix, Blue for QA, Yellow for engineering activity ledger). LP: https://www.codens.ai/

Feedback on the architecture, especially from anyone who's shipped Claude Agent SDK to production, very welcome.

Forem: Takayuki Kawazoe

Claude Code Skills vs deterministic verify commands — same checks, very different ergonomics

What "Skill-based verification" looks like

What verify_commands looks like

Where each one is the better fit

The traps each side hits

My current division of labor (provisional)

The unresolved question

Closing

I shipped a regex to catch AI agents committing only half the work. Then I ripped it back out.

Background: what "partial implementation" looks like

What the shape guard actually was

Reason 1 we ripped it out: half the tickets had no file paths to extract

Reason 2: the regex caught file paths in "References" and "Notes" sections too

Reason 3: the agent's stdout fallback broke when the stdout got truncated

Reason 4: the lockfile + no-op marker collision broke the design assumption

The behavior-based alternative was already in the workflow

What was removed, what was kept

The general pattern: shape vs behavior validation of AI output

A side note worth keeping

Closing

3 MCP server failure modes that bit us in production, and how we ship around them

Failure 1: built-in tools shadow your MCP tools, silently

Failure 2: tool name collisions across MCP servers, with no useful error

Failure 3: the parent process's environment bleeds into the MCP server

What these three failure modes have in common

Closing

One SSE connection, N upstream agents — the stream relay we shipped

The shape of the problem

Stateless relay, leased ownership

The lock, in Redis, in nine lines

Pod rotation, in practice

The tailer: one upstream connection per run

Two heartbeats: liveness vs forward progress

The buffer: Redis Streams with a hard cap

Backpressure on slow clients

When this is overkill

What I'd build differently

Why we built our own credit ledger instead of using Stripe metered billing

What Stripe metered is great at

Where it stops being a good fit

1. The pool is org-wide, not per-product

2. Long-running tasks need reservations

3. Idempotency on retries needs payload checking, not just key checking

4. Refunds at the credit level vs the invoice level

5. The audit trail you want is row-level, not invoice-level

What we kept Stripe for: the actual money

What it cost to build

When Stripe metered is fine

Three things I wish I'd known sooner

I let my AI agents be advisory-only. Here's the rules-first PR risk engine I shipped instead.

The architecture: rules first, AI second

The rules engine: 4 match types, and that's it

High-risk → precheck profile, not AI judgment

Where the AI finally shows up

Three-table audit log

What I'd push back on if I were rebuilding from scratch

Three takeaways

Stop AI from hallucinating E2E test selectors — code analysis + live browser exploration via Claude Agent SDK and 2 MCP servers

The architecture in one diagram

Why in-process MCP, not Docker MCP

The 3-phase flow

Phase 1: Code analysis

Phase 2: Browser exploration

Phase 3: Test synthesis

Lessons after running this in production

What I'm still working on

What `verify_commands` looks like