Correlation Spine: separate-process Judge + Judge-owned correlation_id (+ emitter wiring) [TRO-149, TRO-152] - #20
Merged
Merged
Conversation
Orchestrator-owned correlation_id binding, fail-closed + immutable. Part of the Correlation Spine swarm (invariant #2). [TRO-149] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…binding Fail-closed (UnknownAttackError) + immutable (CorrelationConflictError), typed under a shared RegistryError base. The Judge will resolve correlation_id from this store instead of trusting the Red Team's echo. Part of the Correlation Spine swarm (invariant #2). [TRO-149] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Orchestrator-owned correlation_id binding, fail-closed + immutable. 13/13 frozen tests green; reviewer APPROVE. [TRO-149]
Bytes-in/bytes-out serialize-only boundary; Judge runs in a child PID. Part of the Correlation Spine swarm (invariant #2, separate process). [TRO-149] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Correlation-joined red_team+judge spans, NullEmitter default, PHI-free labels, build_emitter config-selection, fail-open. [TRO-152 slice] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
SubprocessJudge.adjudicate_bytes runs DeterministicJudge in a spawned child (ProcessPoolExecutor, module-level worker); only bytes cross the boundary; malformed envelopes raise JudgeBoundaryError. last_worker_pid proves separate-process execution. Invariant #2 (separate process). [TRO-149] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Verdict.correlation_id owned by CorrelationRegistry, forged embedded id ignored, unbound fails closed, adjudication unchanged by source. [TRO-149] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…on Spine Judge runs in a spawned child; bytes-only boundary. 7/7 green; reviewer APPROVE (distinct per-call worker PIDs, no in-parent adjudication). [TRO-149]
…owned by the registry Additive adjudicate(ctx,*,correlation_id=None,registry=None); registry path resolves by attack_id and ignores the Red Team's echoed correlation_id; unbound fails closed; legacy path unchanged. Invariant #2. [TRO-149] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Additive emitter=NullEmitter(); per-category red_team+judge AgentSpans joined by correlation_id (started_at ordered), measured cost/latency, PHI-free labels via phi_free_label, fail-open emit, build_emitter config selection. [TRO-152 slice] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Verdict.correlation_id resolved from the CorrelationRegistry by attack_id; forged embedded id ignored; unbound fails closed; legacy path unchanged. 7/7 green; reviewer APPROVE (ownership + fail-closed traced). [TRO-149]
Correlation-joined red_team+judge spans (started_at ordered), PHI-free labels, NullEmitter default, fail-open, build_emitter config selection. 7/7 green; reviewer APPROVE (2000-run ordering probe, invariant #2 + PHI safety confirmed). [TRO-152 slice]
Across the process line: forged correlation_id ignored (registry-owned by attack_id), forged observed_hints ignored (independent recompute -> FAIL), unbound attack_id fails closed, genuine oracle fire still SUCCEEDS. [TRO-149] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…orged id + hints Additive binding path on the subprocess boundary: the child builds a CorrelationRegistry from the request binding and adjudicates via the registry path, so a forged AttackResult.correlation_id and forged observed_hints cannot influence the Verdict; unbound attack_id fails closed as JudgeBoundaryError. Legacy correlation_id path unchanged. [TRO-149] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Across the process line the Red Team cannot set the outcome (forged hints ignored, independent recompute) or the correlation_id (registry-owned by attack_id); unbound fails closed. 5/5 green; reviewer APPROVE (child-process resolution, no hint trust, JSEP3 intact). [TRO-149]
📝 WalkthroughWalkthroughThe PR adds orchestrator-owned correlation bindings, a validated serialized subprocess judge boundary, and campaign observability wiring with configurable emitters, timing and cost spans, PHI-free labels, and fail-open emission behavior. ChangesJudge correlation and process integrity
Campaign observability wiring
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)Serialized subprocess adjudicationsequenceDiagram
participant Caller
participant SubprocessJudge
participant run_judge_worker
participant DeterministicJudge
Caller->>SubprocessJudge: adjudicate_bytes(request_bytes)
SubprocessJudge->>SubprocessJudge: validate JSON envelope
SubprocessJudge->>run_judge_worker: send serialized request to spawned worker
run_judge_worker->>DeterministicJudge: adjudicate parsed OracleContext
DeterministicJudge-->>run_judge_worker: return Verdict
run_judge_worker-->>SubprocessJudge: return verdict JSON bytes
SubprocessJudge-->>Caller: return verdict JSON bytes
Campaign span flowsequenceDiagram
participant run_campaign
participant TargetClientLike
participant JudgeLike
participant SpanEmitter
run_campaign->>TargetClientLike: execute input sequence
run_campaign->>SpanEmitter: emit red-team span
run_campaign->>JudgeLike: adjudicate OracleContext
JudgeLike-->>run_campaign: return Verdict
run_campaign->>SpanEmitter: emit judge span
SpanEmitter-->>run_campaign: swallow emission failure
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@agentforge/judge/process.py`:
- Around line 158-181: Review the lifecycle in SubprocessJudge.adjudicate_bytes
and confirm whether spawning a fresh ProcessPoolExecutor per call is acceptable
for expected campaign volume. If throughput requires reuse, replace the per-call
executor with a longer-lived single-worker pool while preserving spawn-based
caller isolation, typed JudgeBoundaryError handling, worker PID tracking, and
deterministic shutdown.
In `@agentforge/live_run.py`:
- Around line 14-48: Update _reported_cost to reject non-finite parsed values by
importing math and requiring math.isfinite(value) alongside the existing
non-negative check; return 0.0 for infinity or NaN while preserving valid
non-negative costs.
In `@agentforge/observability/trace.py`:
- Around line 206-217: Update the urllib.request.urlopen call in _urllib_post to
pass a finite timeout value, ensuring stalled Langfuse requests return control
to _safe_emit instead of blocking indefinitely.
In `@tests/test_observability_wiring.py`:
- Around line 22-23: Update the tests around the build_emitter(...) calls in the
observability wiring tests to patch urllib.request.urlopen and assert it is
never invoked during emitter construction. Keep the existing construction
assertions intact while explicitly enforcing the no-network constructor
contract.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 354e23ec-fd80-4240-914b-f297cba20bfd
📒 Files selected for processing (10)
agentforge/judge/deterministic.pyagentforge/judge/process.pyagentforge/judge/registry.pyagentforge/live_run.pyagentforge/observability/trace.pytests/test_correlation_registry.pytests/test_judge_boundary_integrity.pytests/test_judge_correlation_ownership.pytests/test_judge_process_boundary.pytests/test_observability_wiring.py
Comment on lines
+158
to
+181
| def adjudicate_bytes(self, request_bytes: bytes) -> bytes: | ||
| """Send serialized request bytes to a child Judge process and return the | ||
| serialized Verdict bytes. Raises :class:`JudgeBoundaryError` on a | ||
| malformed envelope or a child that fails to adjudicate.""" | ||
| # Validate parent-side, before spawning — a bad envelope is reported as a | ||
| # clean typed error instead of an opaque child crash. | ||
| self._validate_envelope(request_bytes) | ||
|
|
||
| try: | ||
| # A fresh single-worker pool per call guarantees a real, freshly | ||
| # spawned child interpreter and deterministic cleanup on exit. | ||
| with ProcessPoolExecutor(max_workers=1, mp_context=_SPAWN) as executor: | ||
| verdict_bytes, worker_pid = executor.submit( | ||
| run_judge_worker, request_bytes | ||
| ).result() | ||
| except JudgeBoundaryError: | ||
| raise | ||
| except Exception as exc: # noqa: BLE001 - surface any child failure as a typed boundary error | ||
| raise JudgeBoundaryError( | ||
| f"subprocess Judge failed to adjudicate the request: {exc!r}" | ||
| ) from exc | ||
|
|
||
| self.last_worker_pid = worker_pid | ||
| return verdict_bytes |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🔵 Trivial
Per-call process spawn is costly if this boundary sits in a hot campaign loop.
Every adjudicate_bytes() call spins up a brand-new ProcessPoolExecutor with a freshly spawned interpreter (spawn, not fork), submits exactly one job, and tears the pool down. This is a deliberate, documented isolation trade-off for invariant #2, but if SubprocessJudge is invoked once per attack result in a campaign (as the observability-wiring red_team/judge span pattern in the stack suggests), the interpreter-startup cost is paid on every single adjudication, which can dominate wall-clock time at campaign scale.
Worth confirming whether this cost is acceptable at the expected campaign volume, or whether a longer-lived worker pool (reused across calls, still isolated from the caller) would be worth trading a small amount of isolation purity for throughput.
🧰 Tools
🪛 Ruff (0.15.21)
[warning] 176-178: Avoid specifying long messages outside the exception class
(TRY003)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentforge/judge/process.py` around lines 158 - 181, Review the lifecycle in
SubprocessJudge.adjudicate_bytes and confirm whether spawning a fresh
ProcessPoolExecutor per call is acceptable for expected campaign volume. If
throughput requires reuse, replace the per-call executor with a longer-lived
single-worker pool while preserving spawn-based caller isolation, typed
JudgeBoundaryError handling, worker PID tracking, and deterministic shutdown.
Comment on lines
+14
to
+48
| import time | ||
| from dataclasses import dataclass | ||
| from datetime import datetime, timezone | ||
| from typing import Callable, Protocol | ||
|
|
||
| from agentforge.contracts.common import AttackCategory, OwaspMapping | ||
| from agentforge.contracts.directive import AuthorizedScope | ||
| from agentforge.contracts.result import AttackResult, InputTurn, TargetResponse | ||
| from agentforge.contracts.verdict import Verdict | ||
| from agentforge.judge.base import OracleContext | ||
| from agentforge.observability.trace import ( | ||
| AgentName, | ||
| AgentSpan, | ||
| NullEmitter, | ||
| SpanEmitter, | ||
| phi_free_label, | ||
| ) | ||
| from agentforge.redteam.agent import RedTeam | ||
|
|
||
|
|
||
| def _reported_cost(agent: object) -> float: | ||
| """The agent's own reported model cost in USD if it exposes one, else 0.0. | ||
|
|
||
| A span's ``cost_usd`` must never be ``None`` (it feeds per-agent cost | ||
| attribution), so any unreported/malformed/negative value collapses to 0.0. | ||
| No agent reports a cost today; this is the forward-compatible seam. | ||
| """ | ||
| reported = getattr(agent, "last_cost_usd", None) | ||
| if reported is None: | ||
| return 0.0 | ||
| try: | ||
| value = float(reported) | ||
| except (TypeError, ValueError): | ||
| return 0.0 | ||
| return value if value >= 0 else 0.0 |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
rg -n -C3 'def _reported_cost|last_cost_usd|isfinite' agentforge/live_run.pyRepository: troysatchell/agentforge
Length of output: 713
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
printf '\n== live_run.py around _reported_cost ==\n'
sed -n '34,60p' agentforge/live_run.py
printf '\n== usages of _reported_cost and cost_usd ==\n'
rg -n -C2 '_reported_cost\(|cost_usd' agentforgeRepository: troysatchell/agentforge
Length of output: 16789
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
printf '\n== observability/trace.py relevant span model + emitter ==\n'
sed -n '1,220p' agentforge/observability/trace.py
printf '\n== observability/cost.py ==\n'
sed -n '1,120p' agentforge/observability/cost.py
printf '\n== any explicit JSON serialization settings for pydantic models ==\n'
rg -n 'allow_inf_nan|model_dump_json|json\(|to_json|orjson|json.dumps|pydantic' agentforgeRepository: troysatchell/agentforge
Length of output: 14576
Reject non-finite reported costs. float("inf") still passes the non-negative check here, and that value can propagate into span totals and emitted cost_usd payloads. Require math.isfinite(value) before returning it.
🧰 Tools
🪛 Ruff (0.15.21)
[warning] 17-17: Import from collections.abc instead: Callable
Import from collections.abc
(UP035)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentforge/live_run.py` around lines 14 - 48, Update _reported_cost to reject
non-finite parsed values by importing math and requiring math.isfinite(value)
alongside the existing non-negative check; return 0.0 for infinity or NaN while
preserving valid non-negative costs.
Comment on lines
+206
to
+217
| def _urllib_post(url: str, headers: dict, body: dict) -> dict: | ||
| """Default HTTP POST transport for :class:`LangfuseEmitter`. | ||
|
|
||
| Stdlib ``urllib`` only (no new dependency). It is NEVER called at emitter | ||
| construction — only inside :meth:`LangfuseEmitter.emit`, i.e. once per real | ||
| span emission. Returns the decoded JSON response (``{}`` when empty). | ||
| """ | ||
| payload = json.dumps(body).encode("utf-8") | ||
| request = urllib.request.Request(url, data=payload, headers=headers, method="POST") | ||
| with urllib.request.urlopen(request) as response: # pragma: no cover - network | ||
| raw = response.read().decode("utf-8") | ||
| return json.loads(raw) if raw else {} |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
rg -n -C2 'urlopen\(' agentforge/observability/trace.pyRepository: troysatchell/agentforge
Length of output: 488
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
sed -n '1,320p' agentforge/observability/trace.pyRepository: troysatchell/agentforge
Length of output: 9413
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
rg -n -C3 '_safe_emit|run_campaign|emit\(' agentforgeRepository: troysatchell/agentforge
Length of output: 6353
Set a finite timeout on the Langfuse POST.
urllib.request.urlopen() blocks indefinitely here, and _safe_emit() only catches exceptions after the call returns. A stalled Langfuse endpoint can therefore hang campaign progress; pass a bounded timeout.
Proposed fix
- with urllib.request.urlopen(request) as response: # pragma: no cover - network
+ with urllib.request.urlopen(request, timeout=5.0) as response: # pragma: no cover - network📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def _urllib_post(url: str, headers: dict, body: dict) -> dict: | |
| """Default HTTP POST transport for :class:`LangfuseEmitter`. | |
| Stdlib ``urllib`` only (no new dependency). It is NEVER called at emitter | |
| construction — only inside :meth:`LangfuseEmitter.emit`, i.e. once per real | |
| span emission. Returns the decoded JSON response (``{}`` when empty). | |
| """ | |
| payload = json.dumps(body).encode("utf-8") | |
| request = urllib.request.Request(url, data=payload, headers=headers, method="POST") | |
| with urllib.request.urlopen(request) as response: # pragma: no cover - network | |
| raw = response.read().decode("utf-8") | |
| return json.loads(raw) if raw else {} | |
| def _urllib_post(url: str, headers: dict, body: dict) -> dict: | |
| """Default HTTP POST transport for :class:`LangfuseEmitter`. | |
| Stdlib ``urllib`` only (no new dependency). It is NEVER called at emitter | |
| construction — only inside :meth:`LangfuseEmitter.emit`, i.e. once per real | |
| span emission. Returns the decoded JSON response (``{}`` when empty). | |
| """ | |
| payload = json.dumps(body).encode("utf-8") | |
| request = urllib.request.Request(url, data=payload, headers=headers, method="POST") | |
| with urllib.request.urlopen(request, timeout=5.0) as response: # pragma: no cover - network | |
| raw = response.read().decode("utf-8") | |
| return json.loads(raw) if raw else {} |
🧰 Tools
🪛 ast-grep (0.44.1)
[warning] 214-214: Request-controlled URL passed to urlopen; validate against an allowlist to prevent SSRF.
Context: urllib.request.urlopen(request)
Note: [CWE-918] Server-Side Request Forgery (SSRF).
(urlopen-unsanitized-data)
[info] 212-212: use jsonify instead of json.dumps for JSON output
Context: json.dumps(body)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
🪛 Ruff (0.15.21)
[error] 214-214: Audit URL open for permitted schemes. Allowing use of file: or custom schemes is often unexpected.
(S310)
[error] 215-215: Audit URL open for permitted schemes. Allowing use of file: or custom schemes is often unexpected.
(S310)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentforge/observability/trace.py` around lines 206 - 217, Update the
urllib.request.urlopen call in _urllib_post to pass a finite timeout value,
ensuring stalled Langfuse requests return control to _safe_emit instead of
blocking indefinitely.
Comment on lines
+22
to
+23
| import inspect | ||
| import json |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
rg -n -C2 'build_emitter|urlopen|no network' tests/test_observability_wiring.pyRepository: troysatchell/agentforge
Length of output: 1470
Assert the no-network constructor contract. Patch urllib.request.urlopen around the build_emitter(...) calls and assert it is never invoked so a construction-time network regression can’t slip through.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_observability_wiring.py` around lines 22 - 23, Update the tests
around the build_emitter(...) calls in the observability wiring tests to patch
urllib.request.urlopen and assert it is never invoked during emitter
construction. Keep the existing construction assertions intact while explicitly
enforcing the no-network constructor contract.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Correlation Spine — separate-process Judge + Judge-owned
correlation_id(+ emitter wiring)Closes the gap where the operator console (
web/static/index.html:270) and the directive docstring (contracts/directive.py) assert invariant #2 — "different processes… no channel to influence a verdict" and "the Judge resolves correlation_id from the campaign store rather than trusting anything the Red Team echoes back" — while the code did not yet do it (one caller-suppliedcorrelation_idfed both agents; the Judge ran in-process). This makes the flagship invariant a tested reality.Built test-first via
/tdd-loop(5 tickets, each frozen tests → coding agent → independent review). 473 tests green (was 434; +39). All changes additive — no existing test modified; the pre-existing suite stays green.Tickets (TRO-149 + TRO-152 emitter slice)
CorrelationRegistry— Orchestrator-owned binding, fail-closed (UnknownAttackError) + immutable (CorrelationConflictError)judge/registry.pyadjudicate(ctx, *, correlation_id=None, registry=None)— Verdict.correlation_id resolved byattack_id, forged embedded id ignored, unbound fails closed; legacy path unchangedjudge/deterministic.pySubprocessJudgeserialize-only boundary — bytes in / bytes out, Judge adjudicates in a spawned child (last_worker_pid != parent)judge/process.pycorrelation_idand forgedobserved_hintsacross the process line; unboundattack_idfails closedjudge/process.pyrun_campaign— correlation-joinedred_team+judgespans (started_at ordered), PHI-free labels,NullEmitterdefault, fail-open,build_emitter(config)live_run.py,observability/trace.pyWhy it holds (defense notes)
SubprocessJudgerunsDeterministicJudgein aProcessPoolExecutor(spawn)child via a module-level worker; onlybytescross. Verified empirically (distinct per-call worker PIDs).observed_hintsare never consulted (grep-confirmed zero references injudge/). Malicious "success" hints on a clean body →FAIL.Verdict.correlation_id = registry.resolve(str(attack_id));AttackResult.correlation_idis structurally unreachable. Unboundattack_id→JudgeBoundaryError(fail-closed), never a Verdict with the attacker's id.AgentSpanrejects PHI inmodel/label; judge label =phi_free_label(oracle_results)).Not in scope (deliberately)
The live half of TRO-152 (deploy self-hosted Langfuse + wire emitter to real agent calls with real keys) is infra-gated and excluded — this PR lands the offline emitter-wiring +
build_emitterfactory only.🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Bug Fixes
Tests