Deploy Enterprise Contract Guardian — finale build

.env.example

@@ -0,0 +1,25 @@
+# Copy this file to `.env` and fill in your real values.
+# `.env` is gitignored — your token will NOT be committed.
+#
+#   cp .env.example .env
+#   # then edit .env with your token
+
+# ---- Required for inference ---------------------------------------------
+HF_TOKEN=hf_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+API_BASE_URL=https://router.huggingface.co/v1
+MODEL_NAME=Qwen/Qwen2.5-72B-Instruct
+
+# ---- Optional: where to write the per-task scores -----------------------
+# Set this to enable JSON output. Path is relative to where you run python.
+# SCORES_OUT_PATH=../baseline_scores.json
+
+# ---- Optional: env server URL (default = local Docker) ------------------
+# ENV_BASE_URL=http://localhost:7860
+# Or your deployed HF Space:
+# ENV_BASE_URL=https://YOUR_USERNAME-api-contract-validator.hf.space
+
+# ---- Optional: training-only ---------------------------------------------
+# WANDB_API_KEY=xxxxxxxxxxxxxxxxxxxxxxxxxxxx
+# WANDB_PROJECT=openenv-contract-guardian
+# BASE_MODEL=unsloth/Qwen2.5-1.5B-Instruct-bnb-4bit
+# PUSH_TO_HUB=YOUR_USERNAME/api-contract-validator-grpo

Dockerfile

@@ -25,5 +25,4 @@ HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
     CMD curl -f http://localhost:7860/health || exit 1
 
 # Run the FastAPI server
-ENV ENABLE_WEB_INTERFACE=true
 CMD ["uvicorn", "server.app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -8,47 +8,87 @@ app_port: 7860
 tags:
   - openenv
 pinned: false
-base_path: /web
 ---
 
-# API Contract Validator — OpenEnv Environment
+# Enterprise Contract Guardian — OpenEnv Environment
 
-An OpenEnv RL environment where AI agents learn to validate API request/response
-payloads against OpenAPI specifications. Agents identify type mismatches, missing
-required fields, invalid enum values, format constraint violations, breaking API
-changes, cross-field arithmetic errors, and authentication schema violations.
+> **Meta PyTorch OpenEnv Hackathon × Scaler School of Technology — Grand Finale Submission**
+> **Theme #3.1**: World Modeling → Professional Tasks · ⭐ **Scaler AI Labs bonus track**: Multi-App RL Environment for Enterprise Workflows
 
-## Why This Environment?
+An OpenEnv RL environment that trains agents to do what senior platform engineers do when an API breaks in production: **detect the violation, trace which downstream services are affected, propose a backward-compatible fix, and verify the fix doesn't cascade**.
 
-API contract violations are one of the **top causes of production incidents** in
-microservice architectures. Every API integration requires validating payloads
-against specs — a tedious, error-prone task that developers perform daily. This
-environment trains agents to automate this critical workflow.
+## The Story
+
+> An engineer ships a "small" change to the Users API on Friday evening. It passes local tests. On Monday, **four downstream teams break** — the Orders service, the Billing pipeline, the Notification worker, and the Analytics ETL. The root cause: a single field renamed in one spec, with no awareness of who consumed it.
+>
+> This environment teaches agents the full workflow — not just "find the bug," but **reason about blast radius, propose fixes that preserve compatibility, and verify the migration across every consumer.**
+
+## Why This Environment Matters (Theme #3.1 Alignment)
+
+Per `themes.md` Theme #3.1: *"environments that require real interaction with tools, APIs, or dynamic systems where the model is expected to do real hard work instead of exploiting short-cuts."*
+
+- ✅ **Real tools/APIs**: OpenAPI specs, payloads, consumer service graphs
+- ✅ **Partially observable world**: agent discovers the consumer graph through queries
+- ✅ **Persistent state**: violations found, consumers traced, fixes proposed build up across steps
+- ✅ **Multi-step orchestration**: `detect → trace → propose → validate`
+- ✅ **Enterprise workflow nuance**: versioning, deprecation, backward compatibility
+- ✅ **Verifiable reward**: every step has a deterministic, objective grader
+
+## Architecture: Phase 1 → Phase 2 → Phase 3
+
+| Phase | What the agent does | Task examples |
+|---|---|---|
+| **Phase 1 — Detection** (inherited from Round 1) | Read one OpenAPI spec + payload, report violations | `find_type_mismatches`, `validate_nested_objects`, `detect_breaking_changes` |
+| **Phase 2 — Impact Tracing** | Given a detected breaking change, identify all downstream consumers whose contracts are violated | `trace_downstream_blast_radius` |
+| **Phase 3 — Fix & Verify** | Propose a backward-compatible migration; verify against every consumer spec | `propose_backward_compat_fix`, `multi_service_cascade_fix` |
 
 **Real-world applications:**
-- CI/CD pipeline contract testing
-- API gateway request/response validation
-- SDK compatibility checking between API versions
-- OAuth2 and auth schema enforcement
-- Migration safety audits
+- CI/CD contract gate that blocks a PR with predicted downstream impact
+- Automated migration-plan generator for API versioning
+- Enterprise API gateway pre-deployment safety check
+- SDK compatibility auditor across microservices
+- OAuth2/auth schema change impact analysis
 
 ## How It Works
 
-Each episode presents the agent with:
-1. An **OpenAPI specification** defining expected types, required fields, and constraints
-2. A **payload** containing planted violations
+Each episode places the agent inside a **simulated enterprise** with 3–5 microservices, each owning an OpenAPI spec and declaring which other services consume it.
 
-The agent inspects both and reports violations **one per step**. The environment
-grades each report against ground-truth violations and provides immediate reward feedback.
+```
+Enterprise Service Graph
+┌──────────────┐        ┌──────────────┐        ┌──────────────┐
+│ UsersService │ ─────▶ │ OrdersService │ ─────▶ │BillingService│
+└──────────────┘        └──────────────┘        └──────────────┘
+       │                        │
+       ▼                        ▼
+┌───────────────────┐    ┌──────────────────┐
+│ NotificationsSvc  │    │  AnalyticsETL    │
+└──���────────────────┘    └──────────────────┘
+```
+
+### Episode Flow (Phase 2/3 tasks)
 
 ```
-reset()  →  Agent sees spec + payload with N hidden violations
-step(violation_report)  →  Correct? +1.0 | Wrong path, right type? +0.3 | False positive? -0.3 | Duplicate? -0.1
-step(HINT)  →  Receive a location clue at -0.5 cost
-step(DONE)  →  Episode ends with completeness bonus +0.5 × (found/total)
+reset()
+  →  Agent receives: a changed spec (producer) + partial service graph.
+
+Phase 1 — Detection
+  step(violation_report)    →  Correct? +1.0 | Proximity +0.3 | False positive -0.3 | Duplicate -0.1
+
+Phase 2 — Impact Tracing
+  step(trace_impact)        →  For each consumer correctly flagged: +reward; missed consumer: penalty
+
+Phase 3 — Fix Proposal
+  step(propose_fix)         →  Fix validates against ALL consumers: +big reward | breaks ≥1 consumer: penalty
+  step(validate_fix)        →  Deterministic cross-spec check confirms/rejects the fix
+
+step(DONE)  →  Completeness bonus = 0.5 × (correct_violations / total) × (consumers_traced / total) × fix_valid
 ```
 
-## Tasks (6 difficulty levels)
+Phase 1 tasks retain the simple single-spec flow (used as curriculum starters — `help_guide.md §6`).
+
+## Tasks
+
+### Phase 1 — Detection (curriculum starters, inherited from Round 1)
 
 | Task | Difficulty | Violations | Max Steps | What the Agent Must Find |
 |------|-----------|------------|-----------|--------------------------|
@@ -59,6 +99,19 @@ step(DONE)  →  Episode ends with completeness bonus +0.5 × (found/total)
 | `validate_cross_field_constraints` | Expert | 7 | 18 | Cross-field arithmetic and date ordering on Invoice API — line totals, subtotal sum, tax calculation, discount rules for trial accounts |
 | `validate_auth_request` | Expert | 6 | 14 | OAuth2 token and API key management violations — invalid grant types, bad scopes, MFA token patterns, IP format, rate limits. 2 variants |
 
+### Phase 2 — Impact Tracing (finale — multi-service)
+
+| Task | Difficulty | Max Steps | What the Agent Must Do |
+|---|---|---|---|
+| `trace_downstream_blast_radius` | Hard | 20 | Given a breaking change in a producer spec + a consumer service graph, identify every downstream service whose contract is violated. Graded on precision + recall against ground-truth consumer impact. |
+
+### Phase 3 — Fix & Verify (finale — full workflow)
+
+| Task | Difficulty | Max Steps | What the Agent Must Do |
+|---|---|---|---|
+| `propose_backward_compat_fix` | Expert | 25 | Given a detected breaking change, propose a migration (aliasing, deprecation, version bump). Graded by whether the fix validates against all consumer specs. |
+| `multi_service_cascade_fix` | Expert | 40 | Full workflow: `detect → trace → propose → validate` in one episode, across 3–5 services. Sparse reward with per-phase sub-rewards. |
+
 ### Randomised Episode Generation
 
 All tasks support seed-based randomisation, making the environment suitable for
@@ -72,15 +125,35 @@ All tasks support seed-based randomisation, making the environment suitable for
 
 ## Action Space
 
-Each step the agent submits a `ValidatorAction`:
+Each step the agent submits a `ValidatorAction`. The action type it sends depends on the current episode phase.
+
+### Detection actions (Phase 1)
 
 | Field | Type | Description |
 |-------|------|-------------|
+| `action_type` | `str` | `report_violation` |
 | `field_path` | `str` | Dot-notation path to the violated field (e.g. `customer.email`, `items[1].quantity`). Special values: `DONE` to end episode, `HINT` for a location clue |
 | `violation_type` | `str` | One of: `type_mismatch`, `missing_required`, `invalid_enum`, `format_error`, `extra_field`, `breaking_change`, `cross_field_constraint` |
 | `description` | `str` | Human-readable explanation of the violation |
 | `suggested_fix` | `str` | Optional suggested correction |
 
+### Impact tracing actions (Phase 2 — finale)
+
+| Field | Type | Description |
+|---|---|---|
+| `action_type` | `str` | `trace_impact` |
+| `affected_services` | `list[str]` | Names of downstream services the agent believes are impacted |
+| `reasoning` | `str` | Brief justification for each entry |
+
+### Fix-proposal actions (Phase 3 — finale)
+
+| Field | Type | Description |
+|---|---|---|
+| `action_type` | `str` | `propose_fix` or `validate_fix` |
+| `fix_strategy` | `str` | One of: `field_alias`, `version_bump`, `deprecation_window`, `dual_write`, `consumer_patch` |
+| `spec_patch` | `dict` | JSON patch to apply to the producer spec |
+| `rationale` | `str` | Why this preserves backward compatibility |
+
 ## Observation Space
 
 After each step the agent receives a `ValidatorObservation`:
@@ -100,7 +173,9 @@ After each step the agent receives a `ValidatorObservation`:
 
 ## Reward Function
 
-Partial progress signals — not binary end-of-episode scoring:
+Multiple **independent** reward signals (per `help_guide.md §7`) — reduces reward-hacking risk, provides rich training signal.
+
+### Detection rewards (Phase 1)
 
 | Event | Reward | Rationale |
 |-------|--------|-----------|
@@ -111,7 +186,32 @@ Partial progress signals — not binary end-of-episode scoring:
 | False positive | **−0.3** | Penalises guessing |
 | DONE signal | **+0.5 × (found/total)** | Completeness bonus |
 
-**Final score** = `correct_violations / total_violations` clamped to `(0.01, 0.99)`
+### Impact-tracing rewards (Phase 2)
+
+| Event | Reward | Rationale |
+|---|---|---|
+| Correctly identified affected consumer | **+0.8** | Reward recall |
+| Missed affected consumer | **−0.5** | Penalise under-reporting |
+| False-flag unaffected consumer | **−0.4** | Penalise over-reporting |
+
+### Fix-proposal rewards (Phase 3)
+
+| Event | Reward | Rationale |
+|---|---|---|
+| Fix validates against ALL consumers | **+2.0** | Major incentive — this is the goal |
+| Fix breaks 1+ consumer | **−1.0** | Must be backward compatible |
+| Malformed spec patch | **−0.5** | Format compliance |
+| Invalid strategy for this violation class | **−0.3** | Encourages strategy selection |
+
+### Cross-cutting signals
+
+| Signal | Reward | Rationale (`help_guide.md §7`) |
+|---|---|---|
+| **Step efficiency** | +0.05 per unused step at DONE | Discourages padding |
+| **Format compliance** | −0.2 for malformed actions | Enforces schema |
+| **Anti-hacking (spam)** | −1.0 if > 3× total violations reported | Prevents "report everything" exploit |
+
+**Final episode score** = weighted blend of phase scores; see `server/rewards.py`.
 
 ## Setup
 
@@ -154,18 +254,66 @@ python inference.py
 openenv validate
 ```
 
-## Baseline Scores
+## Training Results
+
+> **Training**: GRPO via TRL + Unsloth · **Hardware**: HuggingFace Jobs T4 GPU
+
+### Reward Curve
+
+*Training plots will be embedded here after onsite training (Apr 25–26).*
+
+<!-- After training, replace with:
+![Reward Curve](results/reward_curve.png)
+*Episode reward over training steps. Baseline (untrained) vs GRPO-trained agent. x-axis: training step, y-axis: episode reward (0–1).*
+
+![Before vs After](results/before_after.png)
+*Per-task score comparison. Baseline model (blue) vs trained checkpoint (green).*
+-->
+
+| Phase | WandB Run | Notebook |
+|---|---|---|
+| GRPO (Phase 1 + Phase 2/3) | *(link after training)* | [`training/grpo_colab.ipynb`](training/grpo_colab.ipynb) |
+
+See [`training/README.md`](training/README.md) for the three ways to run the pipeline (Colab / HF Jobs / local).
+
+### Baseline Scores (pre-training, Qwen2.5-72B-Instruct, recorded 2026-04-25)
+
+| Task | Phase | Score | Steps | Success |
+|---|---|---|---|---|
+| `find_type_mismatches` | 1 | 0.75 | 4 | ✅ |
+| `validate_nested_objects` | 1 | 0.99 | 12 | ✅ |
+| `detect_breaking_changes` | 1 | **0.01** | 20 | ⛔ |
+| `validate_response_schema` | 1 | 0.99 | 10 | ✅ |
+| `validate_cross_field_constraints` | 1 | 0.86 | 8 | ✅ |
+| `validate_auth_request` | 1 | 0.99 | 10 | ✅ |
+| `trace_downstream_blast_radius` | 2 | 0.67 | 1 | ✅ |
+| `propose_backward_compat_fix` | 3 | 0.99 | 1 | ✅ |
+| `multi_service_cascade_fix` | 2+3 | 0.99 | 2 | ✅ |
+
+Full per-step rewards in [`../baseline_scores.json`](../baseline_scores.json).
+
+**Headroom for training**: `detect_breaking_changes` at 0.01 is the biggest opportunity — the 72B model finds the right field paths (proximity hits) but never predicts `violation_type='breaking_change'` correctly. Phase 2 trace is also under-shooting recall. After GRPO training the trained-model row will go alongside this table.
+
+## Why This Matters
+
+API contract violations are the **#1 cause of production incidents in microservice architectures**. Every platform team deals with this weekly. No existing RL environment teaches agents to reason about multi-service contract impact.
+
+**Who benefits from an agent trained on this environment:**
+- Platform / API gateway teams — pre-merge contract safety checks
+- CI/CD pipelines — automated impact analysis before deploy
+- API versioning toolchains — backward-compat migration planning
+- Any engineering org operating ≥ 3 microservices
+
+This is a genuinely underexplored domain in RL/LLM training — no prior benchmarks exist for multi-service API contract reasoning. A model trained here would be publishable as a research artifact.
 
-| Task | Model | Score | Steps |
-|------|-------|-------|-------|
-| `find_type_mismatches` | Qwen2.5-72B-Instruct | ~0.75 | 5–7 |
-| `validate_nested_objects` | Qwen2.5-72B-Instruct | ~0.57 | 8–12 |
-| `detect_breaking_changes` | Qwen2.5-72B-Instruct | ~0.44 | 12–18 |
-| `validate_response_schema` | Qwen2.5-72B-Instruct | ~0.40 | 15–22 |
-| `validate_cross_field_constraints` | Qwen2.5-72B-Instruct | ~0.43 | 10–16 |
-| `validate_auth_request` | Qwen2.5-72B-Instruct | ~0.60 | 8–12 |
+## Links
 
-*Scores are approximate and may vary with temperature/sampling.*
+| Resource | URL |
+|---|---|
+| HuggingFace Space | *(deploy link — add after deployment)* |
+| Training Notebook (Colab) | *(add after onsite training)* |
+| Demo Video / HF Blog | *(add after recording)* |
+| WandB Training Run | *(add after training)* |
 
 ## Project Structure
 
@@ -174,18 +322,23 @@ api_contract_validator/
 ├── openenv.yaml              # OpenEnv manifest
 ├── pyproject.toml            # Python project metadata
 ├── Dockerfile                # Container definition
-├── inference.py              # Baseline inference script
+├── inference.py              # Baseline inference script (OpenAI client, phase-aware)
 ├── README.md                 # This file
-├── models.py                 # Pydantic models (Action, Observation, State)
+├── models.py                 # Pydantic models (Action, Observation, State — all 3 phases)
 ├── client.py                 # WebSocket client (EnvClient subclass)
 ├── __init__.py               # Package exports
+├── results/                  # Training plots (.png) — committed, embedded above
+├── tests/
+│   └── test_environment.py   # 28 tests across all 3 phases
 └── server/
-    ├── __init__.py
     ├── app.py                # FastAPI wiring (create_app)
-    ├── environment.py        # Core environment logic (reset/step/state)
-    ├── spec_generator.py     # Task scenarios with planted violations
-    ├── rewards.py            # Reward computation
-    └── requirements.txt      # Server dependencies
+    ├── environment.py        # Core environment — multi-phase orchestration
+    ├── logging_setup.py      # Structured JSON episode logging
+    ├── spec_generator.py     # Phase 1 — task scenarios with planted violations
+    ├── service_graph.py      # Phase 2 — simulated enterprise service graph
+    ├── impact_tracer.py      # Phase 2 — ground-truth consumer-impact computation
+    ├── fix_validator.py      # Phase 3 — cross-spec fix verification
+    └── rewards.py            # Composable reward rubrics (multi-phase, independent signals)
 ```
 
 ## License

client.py

@@ -44,10 +44,19 @@ class ValidatorEnv(
     def _step_payload(self, action: ValidatorAction) -> Dict[str, Any]:
         """Convert action to JSON payload for the step message."""
         return {
+            "action_type": action.action_type,
+            # Phase 1
             "field_path": action.field_path,
             "violation_type": action.violation_type,
             "description": action.description,
             "suggested_fix": action.suggested_fix,
+            # Phase 2
+            "affected_services": list(action.affected_services),
+            "reasoning": action.reasoning,
+            # Phase 3
+            "fix_strategy": action.fix_strategy,
+            "spec_patch": dict(action.spec_patch),
+            "rationale": action.rationale,
         }
 
     def _parse_result(
@@ -60,10 +69,17 @@ class ValidatorEnv(
             reward=payload.get("reward"),
             task_name=obs_data.get("task_name", ""),
             task_description=obs_data.get("task_description", ""),
+            phase=obs_data.get("phase", "detection"),
             api_spec=obs_data.get("api_spec", {}),
             payload=obs_data.get("payload", {}),
             violations_found=obs_data.get("violations_found", []),
             violations_remaining=obs_data.get("violations_remaining", 0),
+            service_graph=obs_data.get("service_graph", {}),
+            consumers_traced=obs_data.get("consumers_traced", []),
+            total_consumers=obs_data.get("total_consumers", 0),
+            detected_violation=obs_data.get("detected_violation", {}),
+            consumer_specs=obs_data.get("consumer_specs", {}),
+            fix_validation_results=obs_data.get("fix_validation_results", {}),
             feedback=obs_data.get("feedback", ""),
             max_steps=obs_data.get("max_steps", 0),
         )
@@ -79,9 +95,17 @@ class ValidatorEnv(
             episode_id=payload.get("episode_id"),
             step_count=payload.get("step_count", 0),
             task_name=payload.get("task_name", ""),
+            phase=payload.get("phase", "detection"),
             total_violations=payload.get("total_violations", 0),
             correct_reports=payload.get("correct_reports", 0),
             false_positives=payload.get("false_positives", 0),
             duplicate_reports=payload.get("duplicate_reports", 0),
+            total_consumers=payload.get("total_consumers", 0),
+            consumers_correctly_traced=payload.get("consumers_correctly_traced", 0),
+            consumers_missed=payload.get("consumers_missed", 0),
+            consumers_false_flagged=payload.get("consumers_false_flagged", 0),
+            fix_attempts=payload.get("fix_attempts", 0),
+            fix_validated=payload.get("fix_validated", False),
+            fix_breaks_consumers=payload.get("fix_breaks_consumers", 0),
             score=payload.get("score", 0.0),
         )

inference.py

@@ -20,8 +20,20 @@ import asyncio
 import json
 import os
 import textwrap
+from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+# Load .env file (if present) before reading any os.getenv values.
+# .env is gitignored — keeps HF_TOKEN out of source control.
+try:
+    from dotenv import load_dotenv
+
+    _ENV_FILE = Path(__file__).resolve().parent / ".env"
+    if _ENV_FILE.exists():
+        load_dotenv(_ENV_FILE)
+except ImportError:
+    pass  # python-dotenv not installed — fall back to OS env vars only
+
 from openai import OpenAI
 
 from client import ValidatorEnv
@@ -35,9 +47,10 @@ LOCAL_IMAGE_NAME = os.getenv("LOCAL_IMAGE_NAME")
 HF_TOKEN = os.getenv("HF_TOKEN")
 API_BASE_URL = os.getenv("API_BASE_URL", "https://router.huggingface.co/v1")
 MODEL_NAME = os.getenv("MODEL_NAME", "Qwen/Qwen2.5-72B-Instruct")
+SCORES_OUT_PATH = os.getenv("SCORES_OUT_PATH")  # e.g. baseline_scores.json
 
 BENCHMARK = "api_contract_validator"
-TASKS = [
+PHASE1_TASKS = [
     "find_type_mismatches",
     "validate_nested_objects",
     "detect_breaking_changes",
@@ -45,6 +58,11 @@ TASKS = [
     "validate_cross_field_constraints",
     "validate_auth_request",
 ]
+PHASE2_TASKS = ["trace_downstream_blast_radius"]
+PHASE3_TASKS = ["propose_backward_compat_fix"]
+CASCADE_TASKS = ["multi_service_cascade_fix"]
+TASKS = PHASE1_TASKS + PHASE2_TASKS + PHASE3_TASKS + CASCADE_TASKS
+
 MAX_STEPS_PER_TASK = {
     "find_type_mismatches": 10,
     "validate_nested_objects": 15,
@@ -52,6 +70,9 @@ MAX_STEPS_PER_TASK = {
     "validate_response_schema": 25,
     "validate_cross_field_constraints": 18,
     "validate_auth_request": 14,
+    "trace_downstream_blast_radius": 20,
+    "propose_backward_compat_fix": 25,
+    "multi_service_cascade_fix": 40,
 }
 MAX_CONSECUTIVE_FAILURES = 3  # stop retrying same field after this many -0.3 rewards
 TEMPERATURE = 0.2
@@ -94,7 +115,7 @@ def log_end(
 # Prompt construction
 # ---------------------------------------------------------------------------
 
-SYSTEM_PROMPT = textwrap.dedent("""\
+SYSTEM_PROMPT_PHASE1 = textwrap.dedent("""\
 You are an expert API contract validator. You will be given an OpenAPI \
 specification and an API payload. Your job is to find ALL violations in the \
 payload that do not conform to the spec.
@@ -102,6 +123,7 @@ payload that do not conform to the spec.
 Each turn you must respond with EXACTLY one JSON object (no markdown, no \
 explanation outside the JSON):
 {
+    "action_type": "report_violation",
     "field_path": "<dot-notation path to the violated field, or 'DONE' if finished>",
     "violation_type": "<type_mismatch|missing_required|invalid_enum|format_error|extra_field|breaking_change|cross_field_constraint>",
     "description": "<brief explanation of the violation>",
@@ -117,7 +139,7 @@ STRICT RULES:
    - missing_required: required field absent from payload
    - invalid_enum: value not in the allowed enum list
    - format_error: value violates format/pattern/min/max constraint
-   - breaking_change: API v1→v2 change that breaks existing clients (ALWAYS use this for breaking changes)
+   - breaking_change: API v1→v2 change that breaks existing clients
    - cross_field_constraint: arithmetic/date/conditional rule across multiple fields
 5. Do NOT repeat a violation already in 'Violations found so far'.
 6. If last feedback was 'False positive' or negative reward, that field is WRONG — move to a different field.
@@ -125,26 +147,149 @@ STRICT RULES:
 8. You may set field_path='HINT' for a location clue at -0.5 reward cost.
 """)
 
+SYSTEM_PROMPT_PHASE2 = textwrap.dedent("""\
+You are an enterprise API impact analyst. A producer microservice has made \
+a breaking change to its API. You will see:
+  * the breaking change (`violation`)
+  * a list of consumer services with the fields each consumer depends on
+
+Your job: identify EVERY downstream service whose contract is broken by \
+the change. Submit a SINGLE action listing all affected consumers.
+
+Respond with EXACTLY one JSON object:
+{
+    "action_type": "trace_impact",
+    "affected_services": ["ServiceA", "ServiceB", ...],
+    "reasoning": "<why these services are impacted>"
+}
+
+STRICT RULES:
+1. Include a service ONLY if its declared `fields_consumed` overlaps with the \
+   field affected by the breaking change.
+2. For enum-narrowing changes, a consumer is affected only if it emits one \
+   of the removed values.
+3. Do NOT include services that consume unrelated fields — false flags are \
+   penalised heavily.
+4. Service names are case-insensitive but must match those in the consumer list.
+""")
+
+SYSTEM_PROMPT_PHASE3 = textwrap.dedent("""\
+You are a senior platform engineer designing a backward-compatible \
+migration. You see a breaking change and a list of consumer specs.
+
+Your job: propose a fix (a `spec_patch`) that lets every consumer keep \
+working without redeploying.
+
+Respond with EXACTLY one JSON object:
+{
+    "action_type": "propose_fix",
+    "fix_strategy": "<field_alias|version_bump|deprecation_window|dual_write|consumer_patch>",
+    "spec_patch": { ... },
+    "rationale": "<why this preserves backward compat>"
+}
+
+Strategy contracts (the patch must contain these keys):
+  * field_alias        → spec_patch.aliases = { "<old_name>": "<new_name>", ... }
+  * version_bump       → spec_patch.versions = ["v1.0", "v2.0"]   (must keep both)
+  * deprecation_window → spec_patch.deprecated_fields = ["<old_field>"]
+                          OR spec_patch.deprecated_enum_values = ["<old_val>"]
+  * dual_write         → spec_patch.emit_fields = ["<old>", "<new>"]
+  * consumer_patch     → spec_patch.consumers_to_migrate = [<every affected consumer>]
+
+STRICT RULES:
+1. Pick the strategy that fits the change. For enum narrowing, prefer \
+   consumer_patch or version_bump (aliasing cannot restore enum values).
+2. The patch must apply for EVERY affected consumer, not just one.
+3. If the proposal fails, refine the patch and try again — do not repeat \
+   the same failing patch.
+""")
+
+SYSTEM_PROMPT_CASCADE = SYSTEM_PROMPT_PHASE2 + "\n\n" + SYSTEM_PROMPT_PHASE3 + (
+    "\n\nThe episode begins in Phase 2 (trace_impact). After every consumer "
+    "is correctly traced, the environment switches to Phase 3 (propose_fix)."
+)
+
+
+def _system_prompt_for_phase(phase: str, task_name: str) -> str:
+    if task_name in CASCADE_TASKS:
+        return SYSTEM_PROMPT_CASCADE
+    if phase == "tracing" or task_name in PHASE2_TASKS:
+        return SYSTEM_PROMPT_PHASE2
+    if phase == "fix_proposal" or task_name in PHASE3_TASKS:
+        return SYSTEM_PROMPT_PHASE3
+    return SYSTEM_PROMPT_PHASE1
+
 
 def build_user_prompt(
     observation: Dict[str, Any],
     step: int,
     history: List[str],
 ) -> str:
-    """Build the user prompt from the current observation."""
+    """Build a phase-aware user prompt from the current observation."""
+    phase = observation.get("phase", "detection")
+    task_name = observation.get("task_name", "")
+    history_block = "\n".join(history[-5:]) if history else "None"
+
+    if phase == "tracing" or task_name in PHASE2_TASKS:
+        graph = observation.get("service_graph", {})
+        return textwrap.dedent(f"""\
+Step: {step}
+Task: {task_name}
+Phase: 2 — Impact Tracing
+Instructions: {observation.get('task_description', '')}
+
+Breaking change:
+{json.dumps(graph.get('violation', observation.get('detected_violation', {})), indent=2)}
+
+Consumers (each declares fields_consumed):
+{json.dumps(graph.get('consumers', []), indent=2)}
+
+Last feedback: {observation.get('feedback', '')}
+Previous steps:
+{history_block}
+
+Respond with a single JSON object containing action_type='trace_impact'.
+""")
+
+    if phase == "fix_proposal" or task_name in PHASE3_TASKS:
+        violation = observation.get("detected_violation", {})
+        consumer_specs = observation.get("consumer_specs", {})
+        last_results = observation.get("fix_validation_results", {})
+        return textwrap.dedent(f"""\
+Step: {step}
+Task: {task_name}
+Phase: 3 — Fix & Verify
+Instructions: {observation.get('task_description', '')}
+
+Breaking change to fix:
+{json.dumps(violation, indent=2)}
+
+Consumer specs (every consumer must keep working):
+{json.dumps(consumer_specs, indent=2)}
+
+Last fix attempt result: {json.dumps(last_results, indent=2) if last_results else 'None'}
+Last feedback: {observation.get('feedback', '')}
+Previous steps:
+{history_block}
+
+Respond with a single JSON object containing action_type='propose_fix'.
+""")
+
+    # Default: Phase 1 — Detection
     violations_found = observation.get("violations_found", [])
-    found_summary = "None yet"
     if violations_found:
-        found_lines = []
-        for v in violations_found:
-            found_lines.append(f"  - {v['field_path']}: {v['violation_type']}")
+        found_lines = [
+            f"  - {v['field_path']}: {v['violation_type']}"
+            for v in violations_found
+        ]
         found_summary = "\n".join(found_lines)
-
-    history_block = "\n".join(history[-5:]) if history else "None"
+    else:
+        found_summary = "None yet"
 
     return textwrap.dedent(f"""\
 Step: {step}
-Task: {observation.get('task_name', '')}
+Task: {task_name}
+Phase: 1 — Detection
 Instructions: {observation.get('task_description', '')}
 
 API Specification:
@@ -171,10 +316,17 @@ Respond with a single JSON object for the next violation (or DONE).
 # ---------------------------------------------------------------------------
 
 
-def parse_llm_response(text: str) -> Dict[str, str]:
-    """Parse the LLM response into a violation dict.
+_VALID_VIOLATION_TYPES = {
+    "type_mismatch", "missing_required", "invalid_enum",
+    "format_error", "extra_field", "breaking_change", "cross_field_constraint",
+}
+
+
+def parse_llm_response(text: str) -> Dict[str, Any]:
+    """Parse the LLM response into an action dict for any phase.
 
-    Handles cases where the model wraps JSON in markdown fences.
+    Handles markdown fences and infers the action_type when the model
+    omits it but includes phase-specific fields.
     """
     cleaned = text.strip()
     if cleaned.startswith("```"):
@@ -182,60 +334,119 @@ def parse_llm_response(text: str) -> Dict[str, str]:
         lines = [l for l in lines if not l.strip().startswith("```")]
         cleaned = "\n".join(lines).strip()
 
-    _VALID_TYPES = {
-        "type_mismatch", "missing_required", "invalid_enum",
-        "format_error", "extra_field", "breaking_change", "cross_field_constraint",
-    }
-
     try:
         data = json.loads(cleaned)
-        field_path = str(data.get("field_path", "DONE"))
-        violation_type = str(data.get("violation_type", "unknown"))
-
-        # LLMs sometimes embed ":violation_type" inside field_path — strip it
-        if ":" in field_path:
-            parts = field_path.split(":")
-            # Only strip if the suffix looks like a violation type keyword
-            if any(vt in parts[-1].lower() for vt in _VALID_TYPES):
-                field_path = parts[0].strip()
-
-        return {
-            "field_path": field_path,
-            "violation_type": violation_type,
-            "description": str(data.get("description", "")),
-            "suggested_fix": str(data.get("suggested_fix", "")),
-        }
     except json.JSONDecodeError:
-        # If the model just says "DONE" or similar
         upper = cleaned.upper()
         if "DONE" in upper:
-            return {
-                "field_path": "DONE",
-                "violation_type": "",
-                "description": "",
-                "suggested_fix": "",
-            }
+            return {"action_type": "report_violation", "field_path": "DONE",
+                    "violation_type": "", "description": "", "suggested_fix": ""}
+        return {"action_type": "report_violation", "field_path": "DONE",
+                "violation_type": "unknown",
+                "description": f"Failed to parse: {cleaned[:100]}",
+                "suggested_fix": ""}
+
+    action_type = str(data.get("action_type", "")).strip()
+
+    # Infer action_type when omitted
+    if not action_type:
+        if "affected_services" in data:
+            action_type = "trace_impact"
+        elif "fix_strategy" in data or "spec_patch" in data:
+            action_type = "propose_fix"
+        else:
+            action_type = "report_violation"
+
+    if action_type == "trace_impact":
+        services = data.get("affected_services") or []
+        if not isinstance(services, list):
+            services = [str(services)]
+        return {
+            "action_type": "trace_impact",
+            "affected_services": [str(s) for s in services],
+            "reasoning": str(data.get("reasoning", "")),
+        }
+
+    if action_type in ("propose_fix", "validate_fix"):
+        patch = data.get("spec_patch") or {}
+        if not isinstance(patch, dict):
+            patch = {}
         return {
-            "field_path": "DONE",
-            "violation_type": "unknown",
-            "description": f"Failed to parse: {cleaned[:100]}",
-            "suggested_fix": "",
+            "action_type": action_type,
+            "fix_strategy": str(data.get("fix_strategy", "")),
+            "spec_patch": patch,
+            "rationale": str(data.get("rationale", "")),
         }
 
+    # Default: Phase 1 — report_violation
+    field_path = str(data.get("field_path", "DONE"))
+    violation_type = str(data.get("violation_type", "unknown"))
+    if ":" in field_path:
+        parts = field_path.split(":")
+        if any(vt in parts[-1].lower() for vt in _VALID_VIOLATION_TYPES):
+            field_path = parts[0].strip()
+    return {
+        "action_type": "report_violation",
+        "field_path": field_path,
+        "violation_type": violation_type,
+        "description": str(data.get("description", "")),
+        "suggested_fix": str(data.get("suggested_fix", "")),
+    }
+
+
+def _build_action(action_data: Dict[str, Any]) -> ValidatorAction:
+    """Materialise a ValidatorAction from a parsed-LLM dict."""
+    at = action_data.get("action_type", "report_violation")
+    if at == "trace_impact":
+        return ValidatorAction(
+            action_type="trace_impact",
+            affected_services=action_data.get("affected_services", []),
+            reasoning=action_data.get("reasoning", ""),
+        )
+    if at in ("propose_fix", "validate_fix"):
+        return ValidatorAction(
+            action_type=at,
+            fix_strategy=action_data.get("fix_strategy", ""),
+            spec_patch=action_data.get("spec_patch", {}),
+            rationale=action_data.get("rationale", ""),
+        )
+    return ValidatorAction(
+        action_type="report_violation",
+        field_path=action_data.get("field_path", "DONE"),
+        violation_type=action_data.get("violation_type", ""),
+        description=action_data.get("description", ""),
+        suggested_fix=action_data.get("suggested_fix", ""),
+    )
+
+
+def _action_summary(action_data: Dict[str, Any]) -> str:
+    at = action_data.get("action_type", "report_violation")
+    if at == "trace_impact":
+        return f"trace_impact:{','.join(action_data.get('affected_services', []))}"
+    if at in ("propose_fix", "validate_fix"):
+        return f"{at}:{action_data.get('fix_strategy','')}"
+    return (
+        f"{action_data.get('field_path','')}:"
+        f"{action_data.get('violation_type','')}"
+    )
+
 
 def query_llm(
     client: OpenAI,
     observation: Dict[str, Any],
     step: int,
     history: List[str],
-) -> Dict[str, str]:
+) -> Dict[str, Any]:
     """Send the current observation to the LLM and return a parsed action."""
+    phase = observation.get("phase", "detection")
+    task_name = observation.get("task_name", "")
     user_prompt = build_user_prompt(observation, step, history)
+    system_prompt = _system_prompt_for_phase(phase, task_name)
     try:
         completion = client.chat.completions.create(
             model=MODEL_NAME,
             messages=[
-                {"role": "system", "content": SYSTEM_PROMPT},
+                {"role": "system", "content": system_prompt},
                 {"role": "user", "content": user_prompt},
             ],
             temperature=TEMPERATURE,
@@ -246,12 +457,16 @@ def query_llm(
         return parse_llm_response(raw_text)
     except Exception as exc:
         print(f"[DEBUG] LLM request failed: {exc}", flush=True)
-        return {
-            "field_path": "DONE",
-            "violation_type": "",
-            "description": f"LLM error: {exc}",
-            "suggested_fix": "",
-        }
+        # Safe fallback action for any phase
+        if phase == "tracing" or task_name in PHASE2_TASKS:
+            return {"action_type": "trace_impact", "affected_services": [],
+                    "reasoning": f"LLM error: {exc}"}
+        if phase == "fix_proposal" or task_name in PHASE3_TASKS:
+            return {"action_type": "propose_fix", "fix_strategy": "",
+                    "spec_patch": {}, "rationale": f"LLM error: {exc}"}
+        return {"action_type": "report_violation", "field_path": "DONE",
+                "violation_type": "", "description": f"LLM error: {exc}",
+                "suggested_fix": ""}
 
 
 # ---------------------------------------------------------------------------
@@ -263,30 +478,40 @@ async def run_single_task(
     client: OpenAI,
     env: ValidatorEnv,
     task_name: str,
-) -> None:
-    """Run a single task episode and emit structured logs."""
+) -> Dict[str, Any]:
+    """Run a single task episode and emit structured logs.
+
+    Returns a dict with the per-task summary so the caller can aggregate
+    a baseline_scores.json or trained_scores.json file.
+    """
     max_steps = MAX_STEPS_PER_TASK.get(task_name, 15)
     history: List[str] = []
     rewards: List[float] = []
     steps_taken = 0
-    score = 0.01  # default: strictly > 0 as required by evaluator
+    score = 0.01
     success = False
     consecutive_failures = 0
     last_failed_path = ""
+    is_phase1_task = task_name in PHASE1_TASKS
 
     log_start(task=task_name, env=BENCHMARK, model=MODEL_NAME)
 
     try:
         result = await env.reset(task_name=task_name)
-        obs_dict = result.observation.model_dump() if hasattr(result.observation, 'model_dump') else result.observation.__dict__
+        obs_dict = (
+            result.observation.model_dump()
+            if hasattr(result.observation, "model_dump")
+            else result.observation.__dict__
+        )
 
         for step in range(1, max_steps + 1):
             if result.done:
                 break
 
-            # If stuck on same wrong field too many times, request a HINT
-            if consecutive_failures >= MAX_CONSECUTIVE_FAILURES:
+            # Phase 1 only: trigger HINT after repeated failures on same field
+            if is_phase1_task and consecutive_failures >= MAX_CONSECUTIVE_FAILURES:
                 action_data = {
+                    "action_type": "report_violation",
                     "field_path": "HINT",
                     "violation_type": "",
                     "description": "",
@@ -297,64 +522,85 @@ async def run_single_task(
             else:
                 action_data = query_llm(client, obs_dict, step, history)
 
-            action = ValidatorAction(
-                field_path=action_data["field_path"],
-                violation_type=action_data["violation_type"],
-                description=action_data["description"],
-                suggested_fix=action_data["suggested_fix"],
-            )
-
+            action = _build_action(action_data)
             result = await env.step(action)
-            obs_dict = result.observation.model_dump() if hasattr(result.observation, 'model_dump') else result.observation.__dict__
+            obs_dict = (
+                result.observation.model_dump()
+                if hasattr(result.observation, "model_dump")
+                else result.observation.__dict__
+            )
 
             reward = result.reward or 0.0
             done = result.done
-            error = None
-
             rewards.append(reward)
             steps_taken = step
 
-            action_str = f"{action_data['field_path']}:{action_data['violation_type']}"
-            log_step(step=step, action=action_str, reward=reward, done=done, error=error)
-
-            # Track consecutive failures on same field to trigger HINT
-            if reward < 0 and action_data["field_path"] not in ("DONE", "HINT"):
-                if action_data["field_path"] == last_failed_path:
+            action_str = _action_summary(action_data)
+            log_step(step=step, action=action_str, reward=reward, done=done, error=None)
+
+            # Phase 1 only: track stuck-on-same-field
+            if (
+                is_phase1_task
+                and reward < 0
+                and action_data.get("field_path") not in ("DONE", "HINT")
+            ):
+                fp = action_data.get("field_path", "")
+                if fp == last_failed_path:
                     consecutive_failures += 1
                 else:
                     consecutive_failures = 1
-                    last_failed_path = action_data["field_path"]
+                    last_failed_path = fp
             else:
                 consecutive_failures = 0
                 last_failed_path = ""
 
             history.append(
-                f"Step {step}: {action_str} → reward {reward:+.2f} "
-                f"({'correct' if reward >= 1.0 else 'WRONG - do not retry this field' if reward < 0 else 'partial'})"
+                f"Step {step}: {action_str} → reward {reward:+.2f}"
             )
 
             if done:
                 break
 
-        # Compute final score
-        if rewards:
-            correct_count = sum(1 for r in rewards if r >= 1.0)
-            total_violations = obs_dict.get("violations_remaining", 0) + len(
-                obs_dict.get("violations_found", [])
-            )
-            score = (
-                correct_count / total_violations if total_violations > 0 else 0.0
-            )
-        # Clamp to strictly (0, 1) — evaluator rejects 0.0 and 1.0 exactly
+        # Final score: trust env-side score (most accurate); fall back to
+        # Phase 1 heuristic for back-compat with older Phase 1 evaluators.
+        if is_phase1_task:
+            if rewards:
+                correct_count = sum(1 for r in rewards if r >= 1.0)
+                total_violations = obs_dict.get("violations_remaining", 0) + len(
+                    obs_dict.get("violations_found", [])
+                )
+                score = (
+                    correct_count / total_violations
+                    if total_violations > 0
+                    else 0.0
+                )
+        else:
+            # Phase 2/3 score is already computed by the environment.
+            try:
+                state = await env.state()
+                score = getattr(state, "score", None) or 0.01
+            except Exception:
+                score = 0.01
+
         score = min(max(score, 0.01), 0.99)
         success = score >= SUCCESS_SCORE_THRESHOLD
 
     finally:
         log_end(success=success, steps=steps_taken, score=score, rewards=rewards)
 
+    return {
+        "task": task_name,
+        "score": round(score, 4),
+        "steps": steps_taken,
+        "success": success,
+        "rewards": [round(r, 4) for r in rewards],
+    }
+
 
 async def main() -> None:
-    """Run the inference agent against all tasks."""
+    """Run the inference agent against all tasks and optionally save scores."""
+    from datetime import datetime, timezone
+
     openai_client = OpenAI(base_url=API_BASE_URL, api_key=HF_TOKEN)
 
     if LOCAL_IMAGE_NAME:
@@ -363,15 +609,29 @@ async def main() -> None:
         env_url = os.getenv("ENV_BASE_URL", "http://localhost:7860")
         env = ValidatorEnv(base_url=env_url)
 
+    results: List[Dict[str, Any]] = []
     try:
         for task_name in TASKS:
-            await run_single_task(openai_client, env, task_name)
+            res = await run_single_task(openai_client, env, task_name)
+            results.append(res)
     finally:
         try:
             await env.close()
         except Exception as exc:
             print(f"[DEBUG] env.close() error: {exc}", flush=True)
 
+    if SCORES_OUT_PATH:
+        scores_obj = {
+            "model": MODEL_NAME,
+            "benchmark": BENCHMARK,
+            "date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
+            "scores": {r["task"]: r["score"] for r in results},
+            "details": results,
+        }
+        with open(SCORES_OUT_PATH, "w", encoding="utf-8") as fh:
+            json.dump(scores_obj, fh, indent=2)
+        print(f"[INFO] wrote {SCORES_OUT_PATH}", flush=True)
+
 
 if __name__ == "__main__":
     asyncio.run(main())

models.py

@@ -2,7 +2,11 @@
 Data models for the API Contract Validator Environment.
 
 Defines typed Action, Observation, and State models that form the
-contract between the agent and the environment.
+contract between the agent and the environment across three phases:
+
+    Phase 1 — Detection      action_type='report_violation'
+    Phase 2 — Impact Tracing action_type='trace_impact'
+    Phase 3 — Fix & Verify   action_type='propose_fix' | 'validate_fix'
 """
 
 from typing import Any, Dict, List, Optional
@@ -11,19 +15,47 @@ from openenv.core.env_server.types import Action, Observation, State
 from pydantic import Field
 
 
+# ── Action types ──────────────────────────────────────────────────────────
+
+ACTION_REPORT_VIOLATION = "report_violation"
+ACTION_TRACE_IMPACT = "trace_impact"
+ACTION_PROPOSE_FIX = "propose_fix"
+ACTION_VALIDATE_FIX = "validate_fix"
+
+
 # ---------------------------------------------------------------------------
 # Action — what the agent submits each step
 # ---------------------------------------------------------------------------
 
 class ValidatorAction(Action):
-    """A single violation report submitted by the agent.
+    """A single agent action.
 
-    The agent inspects an API spec + payload pair and reports one violation
-    per step.  Submitting ``field_path="DONE"`` signals the agent is finished.
+    The ``action_type`` field selects which phase the action belongs to:
+
+      * ``report_violation`` (Phase 1, default) — uses ``field_path`` and
+        ``violation_type``. Special ``field_path`` values: ``DONE`` ends the
+        episode, ``HINT`` requests a location clue at -0.5 reward.
+      * ``trace_impact`` (Phase 2) — uses ``affected_services`` and
+        ``reasoning``.
+      * ``propose_fix`` / ``validate_fix`` (Phase 3) — uses
+        ``fix_strategy``, ``spec_patch``, ``rationale``.
+
+    All fields are optional so a single dataclass can carry every action
+    type. Phase 1 callers that only set ``field_path`` + ``violation_type``
+    continue to work without modification.
     """
 
+    action_type: str = Field(
+        default=ACTION_REPORT_VIOLATION,
+        description=(
+            "One of 'report_violation' (Phase 1), 'trace_impact' (Phase 2), "
+            "'propose_fix' (Phase 3), 'validate_fix' (Phase 3)."
+        ),
+    )
+
+    # ── Phase 1 — detection ──────────────────────────────────────────
     field_path: str = Field(
-        ...,
+        default="",
         description=(
             "Dot-notation path to the violated field, e.g. 'user.email'. "
             "Use 'DONE' to signal no more violations. "
@@ -31,7 +63,7 @@ class ValidatorAction(Action):
         ),
     )
     violation_type: str = Field(
-        ...,
+        default="",
         description=(
             "Category of violation: type_mismatch | missing_required | "
             "invalid_enum | format_error | extra_field | breaking_change | "
@@ -47,26 +79,68 @@ class ValidatorAction(Action):
         description="Optional suggested correction for the violation.",
     )
 
+    # ── Phase 2 — impact tracing ─────────────────────────────────────
+    affected_services: List[str] = Field(
+        default_factory=list,
+        description=(
+            "Phase 2 — names of downstream services the agent believes "
+            "are impacted by the breaking change."
+        ),
+    )
+    reasoning: str = Field(
+        default="",
+        description="Phase 2 — brief justification for the impact assessment.",
+    )
+
+    # ── Phase 3 — fix & verify ───────────────────────────────────────
+    fix_strategy: str = Field(
+        default="",
+        description=(
+            "Phase 3 — one of: field_alias | version_bump | "
+            "deprecation_window | dual_write | consumer_patch."
+        ),
+    )
+    spec_patch: Dict[str, Any] = Field(
+        default_factory=dict,
+        description="Phase 3 — JSON-shaped patch to apply to the producer spec.",
+    )
+    rationale: str = Field(
+        default="",
+        description="Phase 3 — why the proposed fix preserves backward compatibility.",
+    )
+
 
 # ---------------------------------------------------------------------------
 # Observation — what the agent sees after each step
 # ---------------------------------------------------------------------------
 
 class ValidatorObservation(Observation):
-    """What the environment returns after each agent action.
+    """Environment response after each agent action.
 
     Inherits ``done: bool`` and ``reward: Optional[float]`` from the
     ``Observation`` base class.
     """
 
-    task_name: str = Field(
-        default="",
-        description="Identifier for the current task.",
-    )
+    # ── universal ────────────────────────────────────────────────────
+    task_name: str = Field(default="", description="Current task identifier.")
     task_description: str = Field(
         default="",
         description="Natural-language instructions for the agent.",
     )
+    phase: str = Field(
+        default="detection",
+        description="Current episode phase: detection | tracing | fix_proposal.",
+    )
+    feedback: str = Field(
+        default="",
+        description="Result of the last submitted action.",
+    )
+    max_steps: int = Field(
+        default=0,
+        description="Maximum steps allowed for the current episode.",
+    )
+
+    # ── Phase 1 — detection ──────────────────────────────────────────
     api_spec: Dict[str, Any] = Field(
         default_factory=dict,
         description="The OpenAPI specification (or spec diff for hard tasks).",
@@ -83,13 +157,39 @@ class ValidatorObservation(Observation):
         default=0,
         description="Number of planted violations still undetected.",
     )
-    feedback: str = Field(
-        default="",
-        description="Result of the agent's last submitted report.",
+
+    # ── Phase 2 — impact tracing ─────────────────────────────────────
+    service_graph: Dict[str, Any] = Field(
+        default_factory=dict,
+        description=(
+            "Phase 2 — enterprise service graph: {producer: spec, "
+            "consumers: {name: {spec_excerpt, fields_consumed}}}."
+        ),
     )
-    max_steps: int = Field(
+    consumers_traced: List[str] = Field(
+        default_factory=list,
+        description="Phase 2 — affected services the agent has correctly identified.",
+    )
+    total_consumers: int = Field(
         default=0,
-        description="Maximum steps allowed for the current episode.",
+        description="Phase 2 — total number of services in the graph.",
+    )
+
+    # ── Phase 3 — fix & verify ───────────────────────────────────────
+    detected_violation: Dict[str, Any] = Field(
+        default_factory=dict,
+        description="Phase 3 — the breaking change that needs a fix.",
+    )
+    consumer_specs: Dict[str, Any] = Field(
+        default_factory=dict,
+        description="Phase 3 — consumer specs to validate the fix against.",
+    )
+    fix_validation_results: Dict[str, Any] = Field(
+        default_factory=dict,
+        description=(
+            "Phase 3 — per-consumer validation results from the last "
+            "validate_fix call."
+        ),
     )
 
 
@@ -105,8 +205,23 @@ class ValidatorState(State):
     """
 
     task_name: str = ""
+    phase: str = "detection"
+
+    # Phase 1
     total_violations: int = 0
     correct_reports: int = 0
     false_positives: int = 0
     duplicate_reports: int = 0
+
+    # Phase 2
+    total_consumers: int = 0
+    consumers_correctly_traced: int = 0
+    consumers_missed: int = 0
+    consumers_false_flagged: int = 0
+
+    # Phase 3
+    fix_attempts: int = 0
+    fix_validated: bool = False
+    fix_breaks_consumers: int = 0
+
     score: float = 0.01

pyproject.toml

@@ -10,6 +10,7 @@ requires-python = ">=3.10"
 dependencies = [
     "openenv-core[core]>=0.2.2",
     "openai>=1.0.0",
+    "python-dotenv>=1.0.0",
 ]
 
 [project.optional-dependencies]

results/baseline_table.md

@@ -0,0 +1,11 @@
+| Task | Score | Steps | Success |
+|---|---|---|---|
+| `find_type_mismatches` | 0.75 | 4 | ✅ |
+| `validate_nested_objects` | 0.99 | 12 | ✅ |
+| `detect_breaking_changes` | 0.01 | 20 | ⛔ |
+| `validate_response_schema` | 0.99 | 10 | ✅ |
+| `validate_cross_field_constraints` | 0.86 | 8 | ✅ |
+| `validate_auth_request` | 0.99 | 10 | ✅ |
+| `trace_downstream_blast_radius` | 0.67 | 1 | ✅ |
+| `propose_backward_compat_fix` | 0.99 | 1 | ✅ |
+| `multi_service_cascade_fix` | 0.99 | 2 | ✅ |

server/app.py

@@ -23,13 +23,16 @@ except Exception as exc:
 try:
     from ..models import ValidatorAction, ValidatorObservation
     from .environment import ValidatorEnvironment
+    from .logging_setup import configure_logging
 except (ImportError, ModuleNotFoundError):
     import sys
     import os
     sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
     from models import ValidatorAction, ValidatorObservation
     from server.environment import ValidatorEnvironment
+    from server.logging_setup import configure_logging
 
+configure_logging()
 
 app = create_app(
     ValidatorEnvironment,

server/environment.py

@@ -11,24 +11,58 @@ Special field_path values:
   'HINT' — receive a location hint (costs -0.5 reward)
 """
 
+import json
+import logging
+from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional, Set
 from uuid import uuid4
 
 from openenv.core.env_server.interfaces import Environment
 from openenv.core.env_server.types import State
 
+logger = logging.getLogger(__name__)
+
 try:
-    from ..models import ValidatorAction, ValidatorObservation, ValidatorState
+    from ..models import (
+        ACTION_PROPOSE_FIX,
+        ACTION_REPORT_VIOLATION,
+        ACTION_TRACE_IMPACT,
+        ACTION_VALIDATE_FIX,
+        ValidatorAction,
+        ValidatorObservation,
+        ValidatorState,
+    )
 except (ImportError, ModuleNotFoundError):
     import sys
     import os
     sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-    from models import ValidatorAction, ValidatorObservation, ValidatorState
-
+    from models import (
+        ACTION_PROPOSE_FIX,
+        ACTION_REPORT_VIOLATION,
+        ACTION_TRACE_IMPACT,
+        ACTION_VALIDATE_FIX,
+        ValidatorAction,
+        ValidatorObservation,
+        ValidatorState,
+    )
+
+from .fix_validator import validate_fix
+from .impact_tracer import trace_impact
 from .rewards import (
     RewardBreakdown,
     compute_episode_score,
     compute_step_reward,
+    phase2_episode_score,
+    phase2_trace_rubric,
+    phase3_episode_score,
+    phase3_fix_rubric,
+)
+from .service_graph import (
+    CASCADE_SCENARIO_IDS,
+    CascadeScenario,
+    consumer_specs_for_fix,
+    get_cascade_scenario,
+    public_observation,
 )
 from .spec_generator import (
     AVAILABLE_TASKS,
@@ -38,6 +72,23 @@ from .spec_generator import (
 )
 
 
+# ── Phase 2 / Phase 3 task names ─────────────────────────────────────────
+
+PHASE2_TASKS: Set[str] = {"trace_downstream_blast_radius"}
+PHASE3_TASKS: Set[str] = {"propose_backward_compat_fix"}
+CASCADE_TASKS: Set[str] = {"multi_service_cascade_fix"}
+ALL_TASKS: List[str] = (
+    AVAILABLE_TASKS
+    + sorted(PHASE2_TASKS)
+    + sorted(PHASE3_TASKS)
+    + sorted(CASCADE_TASKS)
+)
+
+PHASE_DETECTION = "detection"
+PHASE_TRACING = "tracing"
+PHASE_FIX = "fix_proposal"
+
+
 def _normalise_path(path: str) -> str:
     """Lower-case and strip whitespace for fuzzy path matching."""
     return path.strip().lower().replace(" ", "")
@@ -120,6 +171,10 @@ def _hint_section(field_path: str) -> str:
     return path
 
 
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
 class ValidatorEnvironment(Environment):
     """API Contract Validator — an OpenEnv RL environment.
 
@@ -150,6 +205,13 @@ class ValidatorEnvironment(Environment):
         self._reported_violations: List[Dict[str, str]] = []
         self._task_index: int = 0
 
+        # Phase 2 / Phase 3 episode state
+        self._cascade: Optional[CascadeScenario] = None
+        self._phase: str = PHASE_DETECTION
+        self._consumers_traced: Set[str] = set()
+        self._last_fix_results: Dict[str, Any] = {}
+        self._cascade_max_steps: int = 0
+
     # ── reset ─────────────────────────────────────────────────────────
 
     def reset(
@@ -160,23 +222,51 @@ class ValidatorEnvironment(Environment):
     ) -> ValidatorObservation:
         """Start a new episode.
 
-        If ``task_name`` is passed in *kwargs* it selects a specific task;
-        otherwise the environment cycles through all available tasks.
+        Dispatches to the right setup path based on ``task_name``:
+
+          * Phase 1 detection tasks (default) → spec + payload
+          * Phase 2 trace task → service graph + breaking change
+          * Phase 3 fix task → detected violation + consumer specs
+          * Cascade task → all three phases in one episode
         """
         task_name = kwargs.get("task_name") or AVAILABLE_TASKS[
             self._task_index % len(AVAILABLE_TASKS)
         ]
         self._task_index += 1
 
-        self._scenario = generate_scenario_for_task(task_name, seed=seed)
+        # Reset shared episode bookkeeping
         self._matched_paths = set()
         self._proximity_paths = set()
         self._reported_violations = []
+        self._consumers_traced = set()
+        self._last_fix_results = {}
+        self._cascade = None
+        self._scenario = None
+
+        if task_name in PHASE2_TASKS:
+            return self._reset_phase2(task_name, seed, episode_id)
+        if task_name in PHASE3_TASKS:
+            return self._reset_phase3(task_name, seed, episode_id)
+        if task_name in CASCADE_TASKS:
+            return self._reset_cascade(task_name, seed, episode_id)
+        return self._reset_phase1(task_name, seed, episode_id)
+
+    # ── Phase 1 reset (unchanged behaviour) ──────────────────────────
+
+    def _reset_phase1(
+        self,
+        task_name: str,
+        seed: Optional[int],
+        episode_id: Optional[str],
+    ) -> ValidatorObservation:
+        self._phase = PHASE_DETECTION
+        self._scenario = generate_scenario_for_task(task_name, seed=seed)
 
         self._state = ValidatorState(
             episode_id=episode_id or str(uuid4()),
             step_count=0,
             task_name=self._scenario.task_name,
+            phase=PHASE_DETECTION,
             total_violations=len(self._scenario.violations),
             correct_reports=0,
             false_positives=0,
@@ -184,11 +274,22 @@ class ValidatorEnvironment(Environment):
             score=0.0,
         )
 
+        logger.info(json.dumps({
+            "event": "episode_start",
+            "episode_id": self._state.episode_id,
+            "task": self._state.task_name,
+            "phase": self._phase,
+            "total_violations": self._state.total_violations,
+            "max_steps": self._scenario.max_steps,
+            "ts": _now(),
+        }))
+
         return ValidatorObservation(
             done=False,
             reward=0.0,
             task_name=self._scenario.task_name,
             task_description=self._scenario.task_description,
+            phase=PHASE_DETECTION,
             api_spec=self._scenario.api_spec,
             payload=self._scenario.payload,
             violations_found=[],
@@ -197,6 +298,183 @@ class ValidatorEnvironment(Environment):
             max_steps=self._scenario.max_steps,
         )
 
+    # ── Phase 2 reset — impact tracing ───────────────────────────────
+
+    def _reset_phase2(
+        self,
+        task_name: str,
+        seed: Optional[int],
+        episode_id: Optional[str],
+    ) -> ValidatorObservation:
+        self._phase = PHASE_TRACING
+        self._cascade = get_cascade_scenario(seed=seed)
+        max_steps = 20
+        self._cascade_max_steps = max_steps
+
+        self._state = ValidatorState(
+            episode_id=episode_id or str(uuid4()),
+            step_count=0,
+            task_name=task_name,
+            phase=PHASE_TRACING,
+            total_consumers=len(self._cascade.consumers),
+            consumers_correctly_traced=0,
+            consumers_missed=len(self._cascade.ground_truth_affected),
+            consumers_false_flagged=0,
+            score=0.01,
+        )
+
+        logger.info(json.dumps({
+            "event": "episode_start",
+            "episode_id": self._state.episode_id,
+            "task": task_name,
+            "phase": self._phase,
+            "scenario": self._cascade.scenario_id,
+            "consumers": [c.name for c in self._cascade.consumers],
+            "max_steps": max_steps,
+            "ts": _now(),
+        }))
+
+        return ValidatorObservation(
+            done=False,
+            reward=0.0,
+            task_name=task_name,
+            task_description=(
+                f"{self._cascade.description} Submit a single trace_impact "
+                f"action listing every downstream service whose contract is "
+                f"broken by the change."
+            ),
+            phase=PHASE_TRACING,
+            service_graph=public_observation(self._cascade),
+            consumers_traced=[],
+            total_consumers=len(self._cascade.consumers),
+            feedback=(
+                "Phase 2 — Impact Tracing. Inspect the service graph and "
+                "submit action_type='trace_impact' with affected_services."
+            ),
+            max_steps=max_steps,
+        )
+
+    # ── Phase 3 reset — fix proposal ─────────────────────────────────
+
+    def _reset_phase3(
+        self,
+        task_name: str,
+        seed: Optional[int],
+        episode_id: Optional[str],
+    ) -> ValidatorObservation:
+        self._phase = PHASE_FIX
+        self._cascade = get_cascade_scenario(seed=seed)
+        max_steps = 25
+        self._cascade_max_steps = max_steps
+
+        self._state = ValidatorState(
+            episode_id=episode_id or str(uuid4()),
+            step_count=0,
+            task_name=task_name,
+            phase=PHASE_FIX,
+            total_consumers=len(self._cascade.consumers),
+            fix_attempts=0,
+            fix_validated=False,
+            score=0.01,
+        )
+
+        logger.info(json.dumps({
+            "event": "episode_start",
+            "episode_id": self._state.episode_id,
+            "task": task_name,
+            "phase": self._phase,
+            "scenario": self._cascade.scenario_id,
+            "acceptable_strategies": self._cascade.acceptable_fix_strategies,
+            "max_steps": max_steps,
+            "ts": _now(),
+        }))
+
+        return ValidatorObservation(
+            done=False,
+            reward=0.0,
+            task_name=task_name,
+            task_description=(
+                f"{self._cascade.description} Submit propose_fix with a "
+                f"fix_strategy and spec_patch that keeps every consumer "
+                f"working."
+            ),
+            phase=PHASE_FIX,
+            detected_violation=self._cascade.violation,
+            consumer_specs=consumer_specs_for_fix(self._cascade),
+            service_graph=public_observation(self._cascade),
+            feedback=(
+                "Phase 3 — Fix & Verify. Submit action_type='propose_fix' "
+                f"with fix_strategy in "
+                f"{self._cascade.acceptable_fix_strategies} and a "
+                "spec_patch object."
+            ),
+            max_steps=max_steps,
+        )
+
+    # ── Cascade reset — full workflow ────────────────────────────────
+
+    def _reset_cascade(
+        self,
+        task_name: str,
+        seed: Optional[int],
+        episode_id: Optional[str],
+    ) -> ValidatorObservation:
+        """Full detect → trace → fix workflow in one episode.
+
+        Starts in tracing phase since the violation is given to the agent
+        upfront (cascade scenarios already include the breaking change).
+        Phase 3 begins after the agent submits a successful trace_impact.
+        """
+        self._phase = PHASE_TRACING
+        self._cascade = get_cascade_scenario(seed=seed)
+        max_steps = 40
+        self._cascade_max_steps = max_steps
+
+        self._state = ValidatorState(
+            episode_id=episode_id or str(uuid4()),
+            step_count=0,
+            task_name=task_name,
+            phase=PHASE_TRACING,
+            total_consumers=len(self._cascade.consumers),
+            consumers_correctly_traced=0,
+            consumers_missed=len(self._cascade.ground_truth_affected),
+            fix_attempts=0,
+            fix_validated=False,
+            score=0.01,
+        )
+
+        logger.info(json.dumps({
+            "event": "episode_start",
+            "episode_id": self._state.episode_id,
+            "task": task_name,
+            "phase": self._phase,
+            "scenario": self._cascade.scenario_id,
+            "max_steps": max_steps,
+            "ts": _now(),
+        }))
+
+        return ValidatorObservation(
+            done=False,
+            reward=0.0,
+            task_name=task_name,
+            task_description=(
+                "Multi-phase cascade: first trace_impact to identify "
+                "affected consumers, then propose_fix with a backward-"
+                "compatible migration. Episode ends when the fix passes "
+                "all consumers or the step budget runs out."
+            ),
+            phase=PHASE_TRACING,
+            service_graph=public_observation(self._cascade),
+            detected_violation=self._cascade.violation,
+            consumer_specs=consumer_specs_for_fix(self._cascade),
+            total_consumers=len(self._cascade.consumers),
+            feedback=(
+                "Cascade episode started in Phase 2. Submit trace_impact "
+                "first, then move on to propose_fix."
+            ),
+            max_steps=max_steps,
+        )
+
     # ── step ──────────────────────────────────────────────────────────
 
     def step(
@@ -205,10 +483,35 @@ class ValidatorEnvironment(Environment):
         timeout_s: Optional[float] = None,
         **kwargs: Any,
     ) -> ValidatorObservation:
-        """Process one violation report from the agent."""
-        if self._scenario is None:
+        """Dispatch one agent action to the matching phase handler."""
+        if self._scenario is None and self._cascade is None:
             raise RuntimeError("Call reset() before step().")
 
+        # Phase 2 — single-step trace
+        if (
+            action.action_type == ACTION_TRACE_IMPACT
+            and self._cascade is not None
+        ):
+            return self._step_trace_impact(action)
+
+        # Phase 3 — fix proposal / validation
+        if (
+            action.action_type in (ACTION_PROPOSE_FIX, ACTION_VALIDATE_FIX)
+            and self._cascade is not None
+        ):
+            return self._step_fix(action)
+
+        # Default — Phase 1 detection (handles report_violation, DONE, HINT)
+        if self._scenario is None:
+            return self._build_observation_phase2(
+                reward=-0.5,
+                done=False,
+                feedback=(
+                    f"Action type '{action.action_type}' is not valid in "
+                    f"phase '{self._phase}'."
+                ),
+            )
+
         self._state.step_count += 1
         signal = action.field_path.strip().upper()
 
@@ -401,6 +704,223 @@ class ValidatorEnvironment(Environment):
 
     # ── helpers ────────────────────────────────────────────────────────
 
+    # ── Phase 2 step — trace_impact ──────────────────────────────────
+
+    def _step_trace_impact(
+        self, action: ValidatorAction
+    ) -> ValidatorObservation:
+        """Grade a single trace_impact action against ground truth."""
+        assert self._cascade is not None
+
+        self._state.step_count += 1
+
+        result = trace_impact(self._cascade, action.affected_services)
+        rubric = phase2_trace_rubric(result)
+        reward = rubric.total
+
+        self._consumers_traced.update(result.correct_hits)
+        self._state.consumers_correctly_traced = len(result.correct_hits)
+        self._state.consumers_missed = len(result.missed)
+        self._state.consumers_false_flagged = len(result.false_flags)
+
+        # In a pure Phase-2 task, one trace ends the episode.
+        # In cascade, a fully-correct trace transitions to Phase 3.
+        is_cascade = self._state.task_name in CASCADE_TASKS
+        all_correct = not result.missed and not result.false_flags
+        steps_exhausted = self._state.step_count >= self._cascade_max_steps
+
+        if is_cascade and all_correct and not steps_exhausted:
+            self._phase = PHASE_FIX
+            self._state.phase = PHASE_FIX
+            done = False
+            feedback = (
+                "All consumers correctly traced. Phase 3 unlocked — submit "
+                "propose_fix with a backward-compatible spec_patch."
+            )
+        else:
+            done = True
+            self._state.score = phase2_episode_score(result)
+            feedback = (
+                f"Phase 2 result — precision {result.precision:.2f}, "
+                f"recall {result.recall:.2f}, f1 {result.f1:.2f}. "
+                f"correct={result.correct_hits} | missed={result.missed} | "
+                f"false-flagged={result.false_flags}"
+            )
+
+        if steps_exhausted and not done:
+            done = True
+            self._state.score = phase2_episode_score(result)
+            feedback += " Step budget exhausted."
+
+        return self._build_observation_phase2(
+            reward=round(reward, 4),
+            done=done,
+            feedback=feedback,
+            rubric_components=rubric.to_dict(),
+        )
+
+    # ── Phase 3 step — propose_fix / validate_fix ────────────────────
+
+    def _step_fix(self, action: ValidatorAction) -> ValidatorObservation:
+        """Grade a fix proposal against every consumer in the scenario."""
+        assert self._cascade is not None
+
+        self._state.step_count += 1
+        self._state.fix_attempts += 1
+
+        fix_result = validate_fix(
+            self._cascade, action.fix_strategy, action.spec_patch
+        )
+        rubric = phase3_fix_rubric(fix_result)
+        reward = rubric.total
+
+        self._state.fix_validated = fix_result.all_consumers_pass
+        self._state.fix_breaks_consumers = len(fix_result.consumers_failing)
+        self._last_fix_results = {
+            "strategy": fix_result.strategy,
+            "consumers_passing": fix_result.consumers_passing,
+            "consumers_failing": fix_result.consumers_failing,
+            "failure_reasons": fix_result.failure_reasons,
+            "notes": fix_result.notes,
+        }
+
+        steps_exhausted = self._state.step_count >= self._cascade_max_steps
+        done = fix_result.all_consumers_pass or steps_exhausted
+
+        if done:
+            self._state.score = phase3_episode_score(fix_result)
+
+        if fix_result.all_consumers_pass:
+            feedback = (
+                f"Fix accepted — strategy '{fix_result.strategy}' "
+                f"validates against all "
+                f"{len(fix_result.consumers_passing)} consumer(s). "
+                f"Episode complete."
+            )
+        elif not fix_result.is_well_formed:
+            feedback = (
+                f"Malformed fix proposal: "
+                f"{'; '.join(fix_result.notes) or 'see field requirements'}."
+            )
+        else:
+            feedback = (
+                f"Fix breaks {len(fix_result.consumers_failing)} consumer(s): "
+                f"{fix_result.consumers_failing}. "
+                f"Refine the spec_patch and try again."
+            )
+            if steps_exhausted:
+                feedback += " Step budget exhausted."
+
+        return self._build_observation_phase3(
+            reward=round(reward, 4),
+            done=done,
+            feedback=feedback,
+            fix_validation_results=self._last_fix_results,
+            rubric_components=rubric.to_dict(),
+        )
+
+    # ── Phase 2 observation builder ──────────────────────────────────
+
+    def _build_observation_phase2(
+        self,
+        *,
+        reward: float,
+        done: bool,
+        feedback: str,
+        rubric_components: Optional[Dict[str, Any]] = None,
+    ) -> ValidatorObservation:
+        assert self._cascade is not None
+        logger.debug(json.dumps({
+            "event": "step",
+            "episode_id": self._state.episode_id,
+            "task": self._state.task_name,
+            "phase": self._phase,
+            "step": self._state.step_count,
+            "reward": reward,
+            "done": done,
+            "rubric": rubric_components,
+            "ts": _now(),
+        }))
+        if done:
+            logger.info(json.dumps({
+                "event": "episode_end",
+                "episode_id": self._state.episode_id,
+                "task": self._state.task_name,
+                "phase": self._phase,
+                "score": round(self._state.score, 4),
+                "steps": self._state.step_count,
+                "consumers_correctly_traced": self._state.consumers_correctly_traced,
+                "consumers_missed": self._state.consumers_missed,
+                "consumers_false_flagged": self._state.consumers_false_flagged,
+                "ts": _now(),
+            }))
+        return ValidatorObservation(
+            done=done,
+            reward=reward,
+            task_name=self._state.task_name,
+            task_description="",
+            phase=self._phase,
+            service_graph=public_observation(self._cascade),
+            consumers_traced=sorted(self._consumers_traced),
+            total_consumers=len(self._cascade.consumers),
+            detected_violation=self._cascade.violation,
+            consumer_specs=consumer_specs_for_fix(self._cascade),
+            feedback=feedback,
+            max_steps=self._cascade_max_steps,
+        )
+
+    # ── Phase 3 observation builder ──────────────────────────────────
+
+    def _build_observation_phase3(
+        self,
+        *,
+        reward: float,
+        done: bool,
+        feedback: str,
+        fix_validation_results: Dict[str, Any],
+        rubric_components: Optional[Dict[str, Any]] = None,
+    ) -> ValidatorObservation:
+        assert self._cascade is not None
+        logger.debug(json.dumps({
+            "event": "step",
+            "episode_id": self._state.episode_id,
+            "task": self._state.task_name,
+            "phase": self._phase,
+            "step": self._state.step_count,
+            "reward": reward,
+            "done": done,
+            "fix_validation": fix_validation_results,
+            "rubric": rubric_components,
+            "ts": _now(),
+        }))
+        if done:
+            logger.info(json.dumps({
+                "event": "episode_end",
+                "episode_id": self._state.episode_id,
+                "task": self._state.task_name,
+                "phase": self._phase,
+                "score": round(self._state.score, 4),
+                "steps": self._state.step_count,
+                "fix_validated": self._state.fix_validated,
+                "fix_attempts": self._state.fix_attempts,
+                "ts": _now(),
+            }))
+        return ValidatorObservation(
+            done=done,
+            reward=reward,
+            task_name=self._state.task_name,
+            task_description="",
+            phase=self._phase,
+            service_graph=public_observation(self._cascade),
+            consumers_traced=sorted(self._consumers_traced),
+            total_consumers=len(self._cascade.consumers),
+            detected_violation=self._cascade.violation,
+            consumer_specs=consumer_specs_for_fix(self._cascade),
+            fix_validation_results=fix_validation_results,
+            feedback=feedback,
+            max_steps=self._cascade_max_steps,
+        )
+
     def _build_observation(
         self,
         *,
@@ -411,11 +931,39 @@ class ValidatorEnvironment(Environment):
         """Construct an observation from current state."""
         assert self._scenario is not None
         remaining = self._state.total_violations - self._state.correct_reports
+
+        logger.debug(json.dumps({
+            "event": "step",
+            "episode_id": self._state.episode_id,
+            "task": self._state.task_name,
+            "step": self._state.step_count,
+            "reward": round(reward, 4),
+            "correct_so_far": self._state.correct_reports,
+            "total_violations": self._state.total_violations,
+            "done": done,
+            "ts": _now(),
+        }))
+
+        if done:
+            logger.info(json.dumps({
+                "event": "episode_end",
+                "episode_id": self._state.episode_id,
+                "task": self._state.task_name,
+                "score": round(self._state.score, 4),
+                "steps": self._state.step_count,
+                "correct": self._state.correct_reports,
+                "total": self._state.total_violations,
+                "false_positives": self._state.false_positives,
+                "duplicates": self._state.duplicate_reports,
+                "ts": _now(),
+            }))
+
         return ValidatorObservation(
             done=done,
             reward=reward,
             task_name=self._scenario.task_name,
             task_description=self._scenario.task_description,
+            phase=PHASE_DETECTION,
             api_spec=self._scenario.api_spec,
             payload=self._scenario.payload,
             violations_found=list(self._reported_violations),

server/fix_validator.py

@@ -0,0 +1,227 @@
+"""
+Phase 3 — backward-compatibility fix validation.
+
+Five strategies are accepted, each grading rule is independent so the
+reward layer can compose a rubric:
+
+    field_alias        — keep the old field name as an alias to the new one
+    version_bump       — expose the change behind a new API version
+    deprecation_window — keep the old field, mark it deprecated, document removal
+    dual_write         — emit both the old and new field for a transition period
+    consumer_patch     — coordinate consumer updates (only valid when the
+                          producer cannot retain backward compat, e.g. enum
+                          narrowing where the old values are illegal upstream)
+
+A fix is graded against every consumer in the scenario's service graph.
+The result includes per-consumer pass/fail so the reward layer can
+penalise partial fixes (breaks ≥1 consumer) without false-rewarding
+fixes that happen to satisfy the easy consumer but break the hard one.
+"""
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List
+
+from .service_graph import CascadeScenario, ConsumerDeclaration
+
+
+VALID_STRATEGIES = {
+    "field_alias",
+    "version_bump",
+    "deprecation_window",
+    "dual_write",
+    "consumer_patch",
+}
+
+
+@dataclass
+class FixValidationResult:
+    """Outcome of validating one fix proposal against all consumers."""
+
+    strategy: str
+    is_well_formed: bool
+    is_strategy_acceptable: bool
+    consumers_passing: List[str] = field(default_factory=list)
+    consumers_failing: List[str] = field(default_factory=list)
+    failure_reasons: Dict[str, str] = field(default_factory=dict)
+    notes: List[str] = field(default_factory=list)
+
+    @property
+    def all_consumers_pass(self) -> bool:
+        return (
+            self.is_well_formed
+            and self.is_strategy_acceptable
+            and len(self.consumers_failing) == 0
+            and len(self.consumers_passing) > 0
+        )
+
+
+# ── Strategy-specific consumer checks ────────────────────────────────────
+
+
+def _check_field_alias(
+    scenario: CascadeScenario,
+    spec_patch: Dict[str, Any],
+    consumer: ConsumerDeclaration,
+) -> tuple[bool, str]:
+    """Pass if the patch reintroduces every consumed field as an alias."""
+    aliases = spec_patch.get("aliases") or spec_patch.get("field_aliases") or {}
+    if not isinstance(aliases, dict) or not aliases:
+        return False, "patch missing 'aliases' map"
+
+    if scenario.scenario_id == "user_email_rename":
+        # Patch must alias the old name back to the new name
+        if "email" not in aliases:
+            return False, "no alias for 'email'"
+        target = aliases["email"]
+        if target != "email_address":
+            return False, f"alias points to '{target}', expected 'email_address'"
+        # Consumer passes if it still consumes 'email' (alias covers it)
+        if "email" in consumer.fields_consumed:
+            return True, ""
+        return True, "consumer not affected"
+
+    if scenario.scenario_id == "orders_status_narrowed":
+        # Aliasing doesn't help an enum narrowing — reject for affected consumers
+        affected = consumer.name in scenario.ground_truth_affected
+        if affected:
+            return False, "field_alias cannot restore removed enum values"
+        return True, "consumer not affected"
+
+    return False, "unknown scenario for field_alias"
+
+
+def _check_version_bump(
+    scenario: CascadeScenario,
+    spec_patch: Dict[str, Any],
+    consumer: ConsumerDeclaration,
+) -> tuple[bool, str]:
+    """Pass if the patch declares both v1 (legacy) and v2 endpoints."""
+    versions = spec_patch.get("versions") or []
+    if not isinstance(versions, list) or len(versions) < 2:
+        return False, "patch missing two-version declaration"
+    has_legacy = any("v1" in str(v).lower() or "1.0" in str(v) for v in versions)
+    has_new = any("v2" in str(v).lower() or "2.0" in str(v) for v in versions)
+    if not (has_legacy and has_new):
+        return False, "patch must keep v1 alongside v2"
+    return True, ""
+
+
+def _check_deprecation_window(
+    scenario: CascadeScenario,
+    spec_patch: Dict[str, Any],
+    consumer: ConsumerDeclaration,
+) -> tuple[bool, str]:
+    """Pass if the patch keeps the old field/enum and marks it deprecated."""
+    if scenario.scenario_id == "user_email_rename":
+        deprecated = spec_patch.get("deprecated_fields") or []
+        if "email" not in deprecated:
+            return False, "must list 'email' under deprecated_fields"
+        return True, ""
+
+    if scenario.scenario_id == "orders_status_narrowed":
+        deprecated_values = spec_patch.get("deprecated_enum_values") or []
+        for value in ("cancelled", "refunded"):
+            if value not in deprecated_values:
+                return False, f"must keep '{value}' as deprecated enum"
+        return True, ""
+
+    return False, "unknown scenario for deprecation_window"
+
+
+def _check_dual_write(
+    scenario: CascadeScenario,
+    spec_patch: Dict[str, Any],
+    consumer: ConsumerDeclaration,
+) -> tuple[bool, str]:
+    """Pass if the patch emits both old and new field names simultaneously."""
+    fields = spec_patch.get("emit_fields") or []
+    if scenario.scenario_id == "user_email_rename":
+        if "email" not in fields or "email_address" not in fields:
+            return False, "must emit both 'email' and 'email_address'"
+        return True, ""
+    return False, "dual_write not supported for this scenario"
+
+
+def _check_consumer_patch(
+    scenario: CascadeScenario,
+    spec_patch: Dict[str, Any],
+    consumer: ConsumerDeclaration,
+) -> tuple[bool, str]:
+    """Pass if the patch lists every truly-affected consumer to migrate."""
+    migrate = spec_patch.get("consumers_to_migrate") or []
+    if not isinstance(migrate, list):
+        return False, "consumers_to_migrate must be a list"
+    if consumer.name in scenario.ground_truth_affected:
+        if consumer.name not in migrate:
+            return False, "affected consumer missing from migration list"
+        return True, ""
+    return True, ""
+
+
+_STRATEGY_CHECKERS = {
+    "field_alias": _check_field_alias,
+    "version_bump": _check_version_bump,
+    "deprecation_window": _check_deprecation_window,
+    "dual_write": _check_dual_write,
+    "consumer_patch": _check_consumer_patch,
+}
+
+
+# ── Public entry point ────────────────────────────────────────────────────
+
+
+def validate_fix(
+    scenario: CascadeScenario,
+    strategy: str,
+    spec_patch: Dict[str, Any],
+) -> FixValidationResult:
+    """Validate a fix proposal against all consumers in the scenario."""
+    notes: List[str] = []
+
+    if not isinstance(spec_patch, dict):
+        return FixValidationResult(
+            strategy=strategy,
+            is_well_formed=False,
+            is_strategy_acceptable=False,
+            notes=["spec_patch must be a JSON object"],
+        )
+
+    if not strategy or strategy not in VALID_STRATEGIES:
+        return FixValidationResult(
+            strategy=strategy,
+            is_well_formed=False,
+            is_strategy_acceptable=False,
+            notes=[
+                f"strategy '{strategy}' not in {sorted(VALID_STRATEGIES)}"
+            ],
+        )
+
+    is_acceptable = strategy in scenario.acceptable_fix_strategies
+    if not is_acceptable:
+        notes.append(
+            f"'{strategy}' not in acceptable strategies "
+            f"{scenario.acceptable_fix_strategies} for this scenario"
+        )
+
+    checker = _STRATEGY_CHECKERS[strategy]
+    passing: List[str] = []
+    failing: List[str] = []
+    reasons: Dict[str, str] = {}
+
+    for consumer in scenario.consumers:
+        ok, reason = checker(scenario, spec_patch, consumer)
+        if ok:
+            passing.append(consumer.name)
+        else:
+            failing.append(consumer.name)
+            reasons[consumer.name] = reason
+
+    return FixValidationResult(
+        strategy=strategy,
+        is_well_formed=True,
+        is_strategy_acceptable=is_acceptable,
+        consumers_passing=passing,
+        consumers_failing=failing,
+        failure_reasons=reasons,
+        notes=notes,
+    )

server/impact_tracer.py

@@ -0,0 +1,95 @@
+"""
+Phase 2 — ground-truth impact tracing.
+
+Given an agent's predicted list of affected consumers, compare against
+the cascade scenario's ground truth and emit a precision/recall-style
+``ImpactTraceResult`` so the reward layer can score each consumer
+decision independently (composable rubric).
+"""
+
+from dataclasses import dataclass, field
+from typing import List
+
+from .service_graph import CascadeScenario
+
+
+@dataclass
+class ImpactTraceResult:
+    """Per-consumer outcome of one ``trace_impact`` action."""
+
+    correct_hits: List[str] = field(default_factory=list)
+    missed: List[str] = field(default_factory=list)
+    false_flags: List[str] = field(default_factory=list)
+    unknown_services: List[str] = field(default_factory=list)
+    total_consumers: int = 0
+
+    @property
+    def precision(self) -> float:
+        flagged = len(self.correct_hits) + len(self.false_flags)
+        if flagged == 0:
+            return 0.0
+        return len(self.correct_hits) / flagged
+
+    @property
+    def recall(self) -> float:
+        truly_affected = len(self.correct_hits) + len(self.missed)
+        if truly_affected == 0:
+            return 1.0  # nothing to find
+        return len(self.correct_hits) / truly_affected
+
+    @property
+    def f1(self) -> float:
+        p, r = self.precision, self.recall
+        if p + r == 0:
+            return 0.0
+        return 2 * p * r / (p + r)
+
+
+def _normalise(name: str) -> str:
+    return name.strip().lower()
+
+
+def trace_impact(
+    scenario: CascadeScenario,
+    predicted_affected: List[str],
+) -> ImpactTraceResult:
+    """Compare agent's predicted consumer list against ground truth.
+
+    ``predicted_affected`` is matched case-insensitively. Names that match
+    no consumer in the scenario are reported in ``unknown_services`` and
+    treated as false flags for reward purposes.
+    """
+    truth_lookup = {_normalise(n): n for n in scenario.ground_truth_affected}
+    known_consumers = {_normalise(c.name): c.name for c in scenario.consumers}
+
+    seen: set = set()
+    correct_hits: List[str] = []
+    false_flags: List[str] = []
+    unknown_services: List[str] = []
+
+    for raw in predicted_affected:
+        key = _normalise(raw)
+        if key in seen:
+            continue
+        seen.add(key)
+
+        if key in truth_lookup:
+            correct_hits.append(truth_lookup[key])
+        elif key in known_consumers:
+            false_flags.append(known_consumers[key])
+        else:
+            unknown_services.append(raw)
+
+    missed = [
+        name
+        for name in scenario.ground_truth_affected
+        if _normalise(name) not in {_normalise(c) for c in correct_hits}
+    ]
+
+    return ImpactTraceResult(
+        correct_hits=correct_hits,
+        missed=missed,
+        false_flags=false_flags,
+        unknown_services=unknown_services,
+        total_consumers=len(scenario.consumers),
+    )

server/logging_setup.py

@@ -0,0 +1,49 @@
+"""Structured JSON logging for API Contract Validator.
+
+Outputs one JSON object per log record to stdout (visible in docker logs)
+and, when writable, to logs/episodes.jsonl for persistent episode history.
+"""
+
+import json
+import logging
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+class _JsonFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        payload: dict = {
+            "ts": datetime.now(timezone.utc).isoformat(),
+            "level": record.levelname,
+            "logger": record.name,
+            "msg": record.getMessage(),
+        }
+        if record.exc_info:
+            payload["exc"] = self.formatException(record.exc_info)
+        return json.dumps(payload)
+
+
+def configure_logging() -> None:
+    """Configure root logger: JSON to stdout + logs/episodes.jsonl."""
+    log_level = os.getenv("LOG_LEVEL", "INFO").upper()
+
+    root = logging.getLogger()
+    if root.handlers:
+        return  # already configured (e.g. during testing)
+
+    root.setLevel(log_level)
+
+    sh = logging.StreamHandler(sys.stdout)
+    sh.setFormatter(_JsonFormatter())
+    root.addHandler(sh)
+
+    log_dir = Path(os.getenv("LOG_DIR", "logs"))
+    try:
+        log_dir.mkdir(parents=True, exist_ok=True)
+        fh = logging.FileHandler(log_dir / "episodes.jsonl", encoding="utf-8")
+        fh.setFormatter(_JsonFormatter())
+        root.addHandler(fh)
+    except OSError:
+        pass  # read-only filesystem (HF Spaces free tier) — stdout only

server/rewards.py

@@ -1,28 +1,113 @@
 """
-Reward computation for the API Contract Validator Environment.
-
-Provides partial-progress reward signals rather than binary end-of-episode
-scoring, creating a richer gradient for RL training:
-
-  - Correct violation found      → +1.0   primary incentive, one reward per unique violation
-  - Path-only match (wrong type) → +0.3   proximity signal — agent found the right field,
-                                           now needs to classify it correctly; each path
-                                           earns proximity reward at most once
-  - HINT requested               → -0.5   informative but expensive; use sparingly
-  - Duplicate report             → -0.1   light penalty — agent should track findings
-  - False positive               → -0.3   penalises random guessing
-  - DONE signal                  → +0.5 × (found/total)   completeness bonus on exit
-
-Final episode score = correct / total, clamped to (0.01, 0.99) so it is
-always strictly between 0 and 1 as required by the OpenEnv eval pipeline.
+Multi-phase reward computation for the API Contract Validator.
+
+This module follows the OpenEnv "composable rubric" pattern: each reward
+signal is an independent ``RubricComponent`` and the total step reward is
+the sum of its components. Independent components have two key benefits:
+
+  1. Reduces reward-hacking risk — an agent that maximises one component
+     while degrading another shows up immediately in per-component logs.
+  2. Gives RL training a richer gradient — partial progress on one
+     component still produces signal even when others are zero.
+
+Phase reward signals
+--------------------
+
+Phase 1 — Detection (legacy ``compute_step_reward`` helper kept for
+back-compat with the existing tests; new code should use the rubric API):
+
+  - correct violation         +1.0
+  - proximity match           +0.3
+  - hint requested            -0.5
+  - duplicate report          -0.1
+  - false positive            -0.3
+  - DONE bonus                +0.5 * (correct / total)
+
+Phase 2 — Impact Tracing:
+
+  - correct consumer hit      +0.8 each
+  - missed consumer           -0.5 each
+  - false-flag consumer       -0.4 each
+  - unknown service name      -0.2 each (sub-rule of false-flag)
+
+Phase 3 — Fix & Verify:
+
+  - fix passes ALL consumers  +2.0
+  - fix breaks 1+ consumer    -1.0
+  - malformed spec patch      -0.5
+  - unacceptable strategy     -0.3
+
+Cross-cutting:
+
+  - format compliance         -0.2 for malformed action JSON
+  - anti-hacking (spam)       -1.0 if total reports > 3 * planted violations
 """
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from typing import List
+
+from .fix_validator import FixValidationResult
+from .impact_tracer import ImpactTraceResult
+
+
+# ── Rubric primitives ────────────────────────────────────────────────────
+
+
+@dataclass
+class RubricComponent:
+    """A single named reward signal."""
+
+    name: str
+    score: float
+    explanation: str = ""
+
+
+@dataclass
+class Rubric:
+    """Composition of independent reward signals.
+
+    The ``total`` property sums every component. Components are kept
+    individually so logs and training analysis can show which signal
+    moved across episodes (the key requirement for the "Pipeline 10%"
+    judging criterion).
+    """
+
+    components: List[RubricComponent] = field(default_factory=list)
+
+    def add(self, name: str, score: float, explanation: str = "") -> "Rubric":
+        self.components.append(
+            RubricComponent(name=name, score=score, explanation=explanation)
+        )
+        return self
+
+    @property
+    def total(self) -> float:
+        return sum(c.score for c in self.components)
+
+    def to_dict(self) -> dict:
+        return {
+            "total": round(self.total, 4),
+            "components": [
+                {
+                    "name": c.name,
+                    "score": round(c.score, 4),
+                    "explanation": c.explanation,
+                }
+                for c in self.components
+            ],
+        }
+
+
+# ── Phase 1 — detection (legacy scalar helper kept for backwards compat) ─
 
 
 @dataclass
 class RewardBreakdown:
-    """Detailed breakdown of a single-step reward."""
+    """Detailed breakdown of a single Phase 1 step reward.
+
+    Kept for backwards compatibility with the existing Phase 1 tests and
+    inference loop. New phases use ``Rubric`` directly.
+    """
 
     reward: float
     is_correct: bool
@@ -34,14 +119,30 @@ class RewardBreakdown:
     explanation: str
 
 
-# ── Per-step reward values ────────────────────────────────────────────────
-
+# Phase 1 reward constants
 CORRECT_VIOLATION_REWARD = 1.0
-PATH_MATCH_REWARD = 0.3          # right field, wrong violation_type
-HINT_PENALTY = -0.5              # cost of requesting a location hint
+PATH_MATCH_REWARD = 0.3
+HINT_PENALTY = -0.5
 DUPLICATE_PENALTY = -0.1
 FALSE_POSITIVE_PENALTY = -0.3
-DONE_BONUS_MULTIPLIER = 0.5      # bonus = multiplier * (correct / total)
+DONE_BONUS_MULTIPLIER = 0.5
+
+# Phase 2 reward constants
+CORRECT_CONSUMER_REWARD = 0.8
+MISSED_CONSUMER_PENALTY = -0.5
+FALSE_FLAG_PENALTY = -0.4
+UNKNOWN_SERVICE_PENALTY = -0.2
+
+# Phase 3 reward constants
+FIX_PASSES_ALL_REWARD = 2.0
+FIX_BREAKS_CONSUMER_PENALTY = -1.0
+MALFORMED_PATCH_PENALTY = -0.5
+UNACCEPTABLE_STRATEGY_PENALTY = -0.3
+
+# Cross-cutting
+MALFORMED_ACTION_PENALTY = -0.2
+SPAM_PENALTY = -1.0
+SPAM_THRESHOLD_MULTIPLIER = 3
 
 
 def compute_step_reward(
@@ -54,30 +155,8 @@ def compute_step_reward(
     correct_so_far: int,
     total_violations: int,
 ) -> RewardBreakdown:
-    """Compute reward for a single agent step.
-
-    Parameters
-    ----------
-    is_correct:
-        Whether the report fully matches a ground-truth violation (path + type).
-    is_path_match:
-        Whether the field_path matches a violation but violation_type is wrong.
-    is_duplicate:
-        Whether the agent already reported this violation.
-    is_done_signal:
-        Whether the agent submitted ``field_path='DONE'``.
-    is_hint:
-        Whether the agent submitted ``field_path='HINT'``.
-    correct_so_far:
-        Number of unique correct violations found before this step.
-    total_violations:
-        Total planted violations in the current scenario.
-
-    Returns
-    -------
-    RewardBreakdown
-        Contains the scalar reward and a human-readable explanation.
-    """
+    """Compute a Phase 1 detection step reward (legacy scalar API)."""
+
     if is_hint:
         return RewardBreakdown(
             reward=HINT_PENALTY,
@@ -93,8 +172,6 @@ def compute_step_reward(
     if is_done_signal:
         completeness = correct_so_far / max(total_violations, 1)
         bonus = DONE_BONUS_MULTIPLIER * completeness
-        # Clamp to strictly (0, 1) — evaluator requires score never equals 0.0 or 1.0
-        # Use 0.01/0.99 so value is still non-zero/non-one after :.2f formatting
         bonus = round(max(0.01, min(0.99, bonus)), 4)
         return RewardBreakdown(
             reward=bonus,
@@ -105,9 +182,8 @@ def compute_step_reward(
             is_done_signal=True,
             is_hint=False,
             explanation=(
-                f"Agent signalled DONE. "
-                f"Completeness {correct_so_far}/{total_violations} "
-                f"→ bonus {bonus:.2f}"
+                f"Agent signalled DONE. Completeness "
+                f"{correct_so_far}/{total_violations} → bonus {bonus:.2f}"
             ),
         )
 
@@ -145,12 +221,12 @@ def compute_step_reward(
             is_done_signal=False,
             is_hint=False,
             explanation=(
-                "Correct field location! The field_path matches a violation, "
-                "but the violation_type is wrong. Try again with the right type."
+                "Correct field location! The field_path matches a "
+                "violation, but the violation_type is wrong. Try again "
+                "with the right type."
             ),
         )
 
-    # False positive
     return RewardBreakdown(
         reward=FALSE_POSITIVE_PENALTY,
         is_correct=False,
@@ -164,13 +240,135 @@ def compute_step_reward(
 
 
 def compute_episode_score(correct_count: int, total_violations: int) -> float:
-    """Compute the final normalised score for the episode.
-
-    Returns a float strictly in ``(0.0, 1.0)`` — endpoints excluded — as
-    required by the OpenEnv evaluation pipeline.
-    """
+    """Final Phase 1 normalised score, strictly in (0, 1)."""
     if total_violations == 0:
         return 0.5
     raw = correct_count / total_violations
-    # Use 0.01/0.99 so value is still non-zero/non-one after :.2f formatting
     return round(max(0.01, min(0.99, raw)), 4)
+
+
+# ── Phase 2 — impact tracing (Rubric API) ────────────────────────────────
+
+
+def phase2_trace_rubric(result: ImpactTraceResult) -> Rubric:
+    """Build a per-consumer Rubric from a Phase 2 impact-trace result."""
+    rubric = Rubric()
+
+    for hit in result.correct_hits:
+        rubric.add(
+            name=f"consumer_correct:{hit}",
+            score=CORRECT_CONSUMER_REWARD,
+            explanation=f"Correctly identified affected consumer '{hit}'.",
+        )
+    for missed in result.missed:
+        rubric.add(
+            name=f"consumer_missed:{missed}",
+            score=MISSED_CONSUMER_PENALTY,
+            explanation=f"Missed affected consumer '{missed}'.",
+        )
+    for flagged in result.false_flags:
+        rubric.add(
+            name=f"consumer_false_flag:{flagged}",
+            score=FALSE_FLAG_PENALTY,
+            explanation=(
+                f"False-flagged unaffected consumer '{flagged}'."
+            ),
+        )
+    for unknown in result.unknown_services:
+        rubric.add(
+            name=f"unknown_service:{unknown}",
+            score=UNKNOWN_SERVICE_PENALTY,
+            explanation=f"'{unknown}' is not a known service in this graph.",
+        )
+    return rubric
+
+
+def phase2_episode_score(result: ImpactTraceResult) -> float:
+    """Phase 2 final score = F1, clamped to (0.01, 0.99)."""
+    return round(max(0.01, min(0.99, result.f1)), 4)
+
+
+# ── Phase 3 — fix validation (Rubric API) ────────────────────────────────
+
+
+def phase3_fix_rubric(result: FixValidationResult) -> Rubric:
+    """Build a Rubric from a Phase 3 fix-validation result."""
+    rubric = Rubric()
+
+    if not result.is_well_formed:
+        rubric.add(
+            name="malformed_patch",
+            score=MALFORMED_PATCH_PENALTY,
+            explanation="; ".join(result.notes) or "Malformed spec patch.",
+        )
+        return rubric
+
+    if not result.is_strategy_acceptable:
+        rubric.add(
+            name="strategy_unacceptable",
+            score=UNACCEPTABLE_STRATEGY_PENALTY,
+            explanation=(
+                f"Strategy '{result.strategy}' is not appropriate for this "
+                f"scenario."
+            ),
+        )
+
+    if result.all_consumers_pass:
+        rubric.add(
+            name="fix_passes_all_consumers",
+            score=FIX_PASSES_ALL_REWARD,
+            explanation=(
+                f"Fix using strategy '{result.strategy}' validates against "
+                f"all {len(result.consumers_passing)} consumer(s)."
+            ),
+        )
+    else:
+        for consumer, reason in result.failure_reasons.items():
+            rubric.add(
+                name=f"fix_breaks_consumer:{consumer}",
+                score=FIX_BREAKS_CONSUMER_PENALTY,
+                explanation=(
+                    f"Fix breaks consumer '{consumer}': {reason}."
+                ),
+            )
+
+    return rubric
+
+
+def phase3_episode_score(result: FixValidationResult) -> float:
+    """Phase 3 final score: 0.99 if all consumers pass else proportional."""
+    if not result.is_well_formed:
+        return 0.01
+    total = len(result.consumers_passing) + len(result.consumers_failing)
+    if total == 0:
+        return 0.01
+    raw = len(result.consumers_passing) / total
+    return round(max(0.01, min(0.99, raw)), 4)
+
+
+# ── Cross-cutting signals ────────────────────────────────────────────────
+
+
+def malformed_action_component() -> RubricComponent:
+    """Penalty for an action JSON that fails schema validation."""
+    return RubricComponent(
+        name="malformed_action",
+        score=MALFORMED_ACTION_PENALTY,
+        explanation="Action did not match the expected schema.",
+    )
+
+
+def spam_penalty_component(reports: int, planted: int) -> RubricComponent | None:
+    """Anti-hacking: agent reporting > 3× planted violations is spamming."""
+    if planted <= 0:
+        return None
+    if reports > SPAM_THRESHOLD_MULTIPLIER * planted:
+        return RubricComponent(
+            name="spam_penalty",
+            score=SPAM_PENALTY,
+            explanation=(
+                f"Reported {reports} violations against {planted} planted "
+                f"— exceeds {SPAM_THRESHOLD_MULTIPLIER}× threshold."
+            ),
+        )
+    return None

server/service_graph.py

@@ -0,0 +1,447 @@
+"""
+Enterprise service graph for Phase 2 / Phase 3 tasks.
+
+Defines a small fictional microservice ecosystem inspired by real
+e-commerce platforms: a producer service (UserService) whose API change
+ripples through several consumers (OrdersService, BillingService,
+NotificationsService, AnalyticsETL).
+
+Each scenario contains:
+  * ``producer_spec_v1``    — original OpenAPI spec
+  * ``producer_spec_v2``    — spec after the breaking change
+  * ``violation``           — the breaking-change record being analysed
+  * ``consumers``           — declarations of which fields each consumer
+                              depends on plus their own contract specs
+  * ``ground_truth_affected`` — names of consumers whose contract is broken
+
+The graph is intentionally compact: judges can read the whole graph in
+under a minute, but the dependency structure is rich enough to surface
+multi-hop impact (e.g. AnalyticsETL only breaks because BillingService
+forwards a renamed field).
+"""
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+
+# ── Data classes ─────────────────────────────────────────────────────────
+
+
+@dataclass
+class ConsumerDeclaration:
+    """How one consumer depends on the producer."""
+
+    name: str
+    description: str
+    fields_consumed: List[str]
+    spec_excerpt: Dict[str, Any]
+
+
+@dataclass
+class CascadeScenario:
+    """A complete Phase 2/Phase 3 scenario."""
+
+    scenario_id: str
+    producer_name: str
+    producer_spec_v1: Dict[str, Any]
+    producer_spec_v2: Dict[str, Any]
+    violation: Dict[str, Any]
+    consumers: List[ConsumerDeclaration]
+    ground_truth_affected: List[str]
+    description: str
+    acceptable_fix_strategies: List[str] = field(
+        default_factory=lambda: [
+            "field_alias",
+            "version_bump",
+            "deprecation_window",
+            "dual_write",
+        ]
+    )
+
+
+# ── Scenario A — UserService renames `email` to `email_address` ──────────
+
+
+def _scenario_user_email_rename() -> CascadeScenario:
+    """A producer renames a field that three consumers read directly."""
+
+    producer_v1 = {
+        "openapi": "3.0.0",
+        "info": {"title": "UserService", "version": "1.0.0"},
+        "paths": {
+            "/users/{id}": {
+                "get": {
+                    "responses": {
+                        "200": {
+                            "content": {
+                                "application/json": {
+                                    "schema": {
+                                        "type": "object",
+                                        "required": ["id", "email", "created_at"],
+                                        "properties": {
+                                            "id": {"type": "string"},
+                                            "email": {
+                                                "type": "string",
+                                                "format": "email",
+                                            },
+                                            "name": {"type": "string"},
+                                            "created_at": {
+                                                "type": "string",
+                                                "format": "date-time",
+                                            },
+                                        },
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        },
+    }
+
+    producer_v2 = {
+        "openapi": "3.0.0",
+        "info": {"title": "UserService", "version": "2.0.0"},
+        "paths": {
+            "/users/{id}": {
+                "get": {
+                    "responses": {
+                        "200": {
+                            "content": {
+                                "application/json": {
+                                    "schema": {
+                                        "type": "object",
+                                        "required": [
+                                            "id",
+                                            "email_address",
+                                            "created_at",
+                                        ],
+                                        "properties": {
+                                            "id": {"type": "string"},
+                                            "email_address": {
+                                                "type": "string",
+                                                "format": "email",
+                                            },
+                                            "name": {"type": "string"},
+                                            "created_at": {
+                                                "type": "string",
+                                                "format": "date-time",
+                                            },
+                                        },
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        },
+    }
+
+    violation = {
+        "field_path": "GET /users/{id}.email",
+        "violation_type": "breaking_change",
+        "description": (
+            "Field 'email' was renamed to 'email_address' and the original "
+            "'email' was removed from required fields and properties."
+        ),
+        "from": "email",
+        "to": "email_address",
+    }
+
+    consumers = [
+        ConsumerDeclaration(
+            name="OrdersService",
+            description="Reads user.email to attach customer email to orders.",
+            fields_consumed=["id", "email"],
+            spec_excerpt={
+                "expects": {
+                    "id": {"type": "string"},
+                    "email": {"type": "string", "format": "email"},
+                }
+            },
+        ),
+        ConsumerDeclaration(
+            name="BillingService",
+            description=(
+                "Reads user.email to send invoices; forwards email to "
+                "AnalyticsETL through its own response."
+            ),
+            fields_consumed=["id", "email"],
+            spec_excerpt={
+                "expects": {
+                    "id": {"type": "string"},
+                    "email": {"type": "string", "format": "email"},
+                }
+            },
+        ),
+        ConsumerDeclaration(
+            name="NotificationsService",
+            description="Sends transactional emails to user.email.",
+            fields_consumed=["email"],
+            spec_excerpt={
+                "expects": {
+                    "email": {"type": "string", "format": "email"},
+                }
+            },
+        ),
+        ConsumerDeclaration(
+            name="AnalyticsETL",
+            description=(
+                "Reads only id and created_at from UserService directly. "
+                "(Tempting false-flag — does NOT consume 'email'.)"
+            ),
+            fields_consumed=["id", "created_at"],
+            spec_excerpt={
+                "expects": {
+                    "id": {"type": "string"},
+                    "created_at": {"type": "string", "format": "date-time"},
+                }
+            },
+        ),
+    ]
+
+    return CascadeScenario(
+        scenario_id="user_email_rename",
+        producer_name="UserService",
+        producer_spec_v1=producer_v1,
+        producer_spec_v2=producer_v2,
+        violation=violation,
+        consumers=consumers,
+        ground_truth_affected=[
+            "OrdersService",
+            "BillingService",
+            "NotificationsService",
+        ],
+        description=(
+            "UserService renamed 'email' to 'email_address'. Identify which "
+            "downstream services break, and propose a fix that keeps all "
+            "consumers working without forcing them to redeploy."
+        ),
+        acceptable_fix_strategies=[
+            "field_alias",
+            "version_bump",
+            "deprecation_window",
+            "dual_write",
+        ],
+    )
+
+
+# ── Scenario B — OrdersService narrows `status` enum ─────────────────────
+
+
+def _scenario_orders_status_narrowed() -> CascadeScenario:
+    """A producer removes enum values that two consumers still emit."""
+
+    producer_v1 = {
+        "openapi": "3.0.0",
+        "info": {"title": "OrdersService", "version": "1.0.0"},
+        "paths": {
+            "/orders/{id}/status": {
+                "put": {
+                    "requestBody": {
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "type": "object",
+                                    "required": ["status"],
+                                    "properties": {
+                                        "status": {
+                                            "type": "string",
+                                            "enum": [
+                                                "pending",
+                                                "confirmed",
+                                                "shipped",
+                                                "delivered",
+                                                "cancelled",
+                                                "refunded",
+                                            ],
+                                        }
+                                    },
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        },
+    }
+
+    producer_v2 = {
+        "openapi": "3.0.0",
+        "info": {"title": "OrdersService", "version": "2.0.0"},
+        "paths": {
+            "/orders/{id}/status": {
+                "put": {
+                    "requestBody": {
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "type": "object",
+                                    "required": ["status"],
+                                    "properties": {
+                                        "status": {
+                                            "type": "string",
+                                            "enum": [
+                                                "pending",
+                                                "confirmed",
+                                                "shipped",
+                                                "delivered",
+                                            ],
+                                        }
+                                    },
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        },
+    }
+
+    violation = {
+        "field_path": "PUT /orders/{id}/status.status",
+        "violation_type": "breaking_change",
+        "description": (
+            "Enum narrowed: 'cancelled' and 'refunded' were removed from "
+            "the allowed status values."
+        ),
+        "removed_values": ["cancelled", "refunded"],
+    }
+
+    consumers = [
+        ConsumerDeclaration(
+            name="ReturnsService",
+            description=(
+                "Sets status='refunded' when processing a return. Will "
+                "fail on v2 because 'refunded' is no longer accepted."
+            ),
+            fields_consumed=["status"],
+            spec_excerpt={
+                "emits": {"status": {"enum": ["refunded"]}}
+            },
+        ),
+        ConsumerDeclaration(
+            name="SupportPortal",
+            description=(
+                "Allows agents to mark orders as 'cancelled'. Will fail on "
+                "v2 because 'cancelled' is no longer accepted."
+            ),
+            fields_consumed=["status"],
+            spec_excerpt={
+                "emits": {"status": {"enum": ["cancelled"]}}
+            },
+        ),
+        ConsumerDeclaration(
+            name="ShippingService",
+            description=(
+                "Only emits 'shipped' and 'delivered' — both still valid. "
+                "(False-flag candidate.)"
+            ),
+            fields_consumed=["status"],
+            spec_excerpt={
+                "emits": {"status": {"enum": ["shipped", "delivered"]}}
+            },
+        ),
+    ]
+
+    return CascadeScenario(
+        scenario_id="orders_status_narrowed",
+        producer_name="OrdersService",
+        producer_spec_v1=producer_v1,
+        producer_spec_v2=producer_v2,
+        violation=violation,
+        consumers=consumers,
+        ground_truth_affected=["ReturnsService", "SupportPortal"],
+        description=(
+            "OrdersService narrowed the order status enum, removing "
+            "'cancelled' and 'refunded'. Identify which consumers can no "
+            "longer call this endpoint."
+        ),
+        acceptable_fix_strategies=[
+            "version_bump",
+            "deprecation_window",
+            "consumer_patch",
+        ],
+    )
+
+
+# ── Public registry ───────────────────────────────────────────────────────
+
+
+_SCENARIOS: Dict[str, CascadeScenario] = {
+    "user_email_rename": _scenario_user_email_rename(),
+    "orders_status_narrowed": _scenario_orders_status_narrowed(),
+}
+
+
+def get_cascade_scenario(
+    scenario_id: Optional[str] = None,
+    seed: Optional[int] = None,
+) -> CascadeScenario:
+    """Return a cascade scenario by id, or pick one deterministically by seed.
+
+    Parameters
+    ----------
+    scenario_id:
+        Explicit scenario name. Takes precedence over ``seed``.
+    seed:
+        If provided (and ``scenario_id`` is not), selects ``user_email_rename``
+        for even seeds and ``orders_status_narrowed`` for odd seeds.
+
+    Returns
+    -------
+    CascadeScenario
+        The (immutable) selected scenario.
+    """
+    if scenario_id is not None:
+        if scenario_id not in _SCENARIOS:
+            raise ValueError(
+                f"Unknown cascade scenario '{scenario_id}'. "
+                f"Available: {list(_SCENARIOS)}"
+            )
+        return _SCENARIOS[scenario_id]
+
+    keys = sorted(_SCENARIOS.keys())
+    if seed is None:
+        return _SCENARIOS[keys[0]]
+    return _SCENARIOS[keys[seed % len(keys)]]
+
+
+def public_observation(scenario: CascadeScenario) -> Dict[str, Any]:
+    """Return the portion of the scenario the agent is allowed to see.
+
+    The ground-truth ``ground_truth_affected`` list is held back so the
+    agent must reason about impact from the consumer declarations rather
+    than read the answer.
+    """
+    return {
+        "producer": scenario.producer_name,
+        "producer_spec_v1": scenario.producer_spec_v1,
+        "producer_spec_v2": scenario.producer_spec_v2,
+        "violation": scenario.violation,
+        "consumers": [
+            {
+                "name": c.name,
+                "description": c.description,
+                "fields_consumed": c.fields_consumed,
+                "spec_excerpt": c.spec_excerpt,
+            }
+            for c in scenario.consumers
+        ],
+    }
+
+
+def consumer_specs_for_fix(scenario: CascadeScenario) -> Dict[str, Any]:
+    """Return only the consumer specs needed for Phase 3 fix validation."""
+    return {
+        c.name: {
+            "spec_excerpt": c.spec_excerpt,
+            "fields_consumed": c.fields_consumed,
+        }
+        for c in scenario.consumers
+    }
+
+
+CASCADE_SCENARIO_IDS: List[str] = sorted(_SCENARIOS.keys())

tests/test_environment.py

@@ -214,3 +214,153 @@ def test_auth_task_variants_differ():
 def test_easy_pool_has_twelve_variants():
     from server.spec_generator import _EASY_POOL
     assert len(_EASY_POOL) == 12, f"Expected 12 pool entries, got {len(_EASY_POOL)}"
+
+
+# ── Phase 2 — impact tracing ───────────────────────────────────────────────
+
+
+def test_phase2_reset_returns_service_graph(env):
+    obs = env.reset(task_name="trace_downstream_blast_radius", seed=1)
+    assert obs.phase == "tracing"
+    assert obs.total_consumers >= 3
+    assert "consumers" in obs.service_graph
+    assert obs.feedback
+
+
+def test_phase2_perfect_trace_scores_high(env):
+    env.reset(task_name="trace_downstream_blast_radius", seed=1)
+    action = ValidatorAction(
+        action_type="trace_impact",
+        affected_services=[
+            "OrdersService",
+            "BillingService",
+            "NotificationsService",
+        ],
+    )
+    result = env.step(action)
+    assert result.done is True
+    assert result.reward > 2.0  # 3 hits @ +0.8 each
+    assert env.state.score > 0.9
+
+
+def test_phase2_false_flag_penalty(env):
+    env.reset(task_name="trace_downstream_blast_radius", seed=1)
+    action = ValidatorAction(
+        action_type="trace_impact",
+        affected_services=["OrdersService", "AnalyticsETL"],  # one false flag
+    )
+    result = env.step(action)
+    assert result.done is True
+    # 1 hit (+0.8) + 2 missed (-0.5 each) + 1 false (-0.4) = -0.6
+    assert result.reward < 0
+
+
+def test_phase2_unknown_service_treated_as_false_flag(env):
+    env.reset(task_name="trace_downstream_blast_radius", seed=1)
+    action = ValidatorAction(
+        action_type="trace_impact",
+        affected_services=["NonexistentService"],
+    )
+    result = env.step(action)
+    assert result.done is True
+    assert result.reward < 0
+
+
+# ── Phase 3 — fix proposal ─────────────────────────────────────────────────
+
+
+def test_phase3_reset_returns_violation_and_consumers(env):
+    obs = env.reset(task_name="propose_backward_compat_fix", seed=1)
+    assert obs.phase == "fix_proposal"
+    assert obs.detected_violation
+    assert obs.consumer_specs
+
+
+def test_phase3_good_field_alias_passes_all_consumers(env):
+    env.reset(task_name="propose_backward_compat_fix", seed=1)
+    action = ValidatorAction(
+        action_type="propose_fix",
+        fix_strategy="field_alias",
+        spec_patch={"aliases": {"email": "email_address"}},
+        rationale="Keep old field name as alias",
+    )
+    result = env.step(action)
+    assert result.done is True
+    assert result.reward >= 2.0
+    assert env.state.fix_validated is True
+
+
+def test_phase3_malformed_strategy_penalty(env):
+    env.reset(task_name="propose_backward_compat_fix", seed=1)
+    action = ValidatorAction(
+        action_type="propose_fix",
+        fix_strategy="not_a_real_strategy",
+        spec_patch={},
+    )
+    result = env.step(action)
+    assert result.reward < 0
+    assert env.state.fix_validated is False
+
+
+def test_phase3_breaking_consumer_penalised(env):
+    env.reset(task_name="propose_backward_compat_fix", seed=1)
+    # dual_write but missing the new field — breaks all consumers
+    action = ValidatorAction(
+        action_type="propose_fix",
+        fix_strategy="dual_write",
+        spec_patch={"emit_fields": ["email"]},
+    )
+    result = env.step(action)
+    assert result.reward < 0
+    assert env.state.fix_validated is False
+
+
+# ── Cascade — full workflow ───────────────────────────────────────────────
+
+
+def test_cascade_starts_in_tracing_phase(env):
+    obs = env.reset(task_name="multi_service_cascade_fix", seed=1)
+    assert obs.phase == "tracing"
+
+
+def test_cascade_transitions_to_fix_after_correct_trace(env):
+    env.reset(task_name="multi_service_cascade_fix", seed=1)
+    trace = ValidatorAction(
+        action_type="trace_impact",
+        affected_services=[
+            "OrdersService",
+            "BillingService",
+            "NotificationsService",
+        ],
+    )
+    obs = env.step(trace)
+    assert obs.done is False
+    assert obs.phase == "fix_proposal"
+
+    fix = ValidatorAction(
+        action_type="propose_fix",
+        fix_strategy="field_alias",
+        spec_patch={"aliases": {"email": "email_address"}},
+    )
+    obs = env.step(fix)
+    assert obs.done is True
+    assert env.state.fix_validated is True
+
+
+# ── Determinism ────────────────────────────────────────────────────────────
+
+
+def test_phase2_seed_determinism(env):
+    obs1 = env.reset(task_name="trace_downstream_blast_radius", seed=1)
+    obs2 = env.reset(task_name="trace_downstream_blast_radius", seed=1)
+    services1 = sorted(c["name"] for c in obs1.service_graph["consumers"])
+    services2 = sorted(c["name"] for c in obs2.service_graph["consumers"])
+    assert services1 == services2
+
+
+def test_different_seeds_pick_different_scenarios(env):
+    obs_even = env.reset(task_name="trace_downstream_blast_radius", seed=0)
+    obs_odd = env.reset(task_name="trace_downstream_blast_radius", seed=1)
+    name_even = obs_even.service_graph["producer"]
+    name_odd = obs_odd.service_graph["producer"]
+    assert name_even != name_odd

training/README.md

@@ -0,0 +1,80 @@
+# Training — Enterprise Contract Guardian
+
+Re-runnable training pipeline for the API Contract Validator environment, using GRPO from TRL with LoRA adapters via Unsloth.
+
+## Files
+
+| File | Purpose |
+|---|---|
+| `baseline.py` | Run the untrained model on every task; write `baseline_scores.json` |
+| `train.py` | GRPO training loop — connects to env, rolls out, trains LoRA |
+| `plot.py` | Build `reward_curve.png` and `before_after.png` for the README |
+| `grpo_colab.ipynb` | One-click Colab notebook (open in Colab → Runtime → Run all) |
+
+## Three ways to run training
+
+### A. Colab notebook (easiest, free GPU)
+
+Open `grpo_colab.ipynb` in Colab. Set `HF_TOKEN` and `WANDB_API_KEY` in the secrets pane. Hit **Runtime → Run all**.
+
+### B. HF Jobs (best for the onsite — uses your $30 credit)
+
+```bash
+hf jobs uv run \
+    --with trl --with unsloth --with openenv-core --with wandb \
+    --flavor t4-small \
+    -s HF_TOKEN -s WANDB_API_KEY \
+    -- python training/train.py
+```
+
+### C. Local GPU
+
+```bash
+pip install trl unsloth wandb matplotlib datasets
+export HF_TOKEN="hf_..."
+export WANDB_API_KEY="..."
+export ENV_URL="http://localhost:7860"   # or your HF Space URL
+
+# 1. Start the env server in another terminal
+uvicorn server.app:app --host 0.0.0.0 --port 7860
+
+# 2. Baseline
+python training/baseline.py
+
+# 3. Train
+python training/train.py
+
+# 4. Inference with trained adapter
+export MODEL_NAME="<your-username>/api-contract-validator-grpo"
+export SCORES_OUT_PATH="trained_scores.json"
+python inference.py
+
+# 5. Plots
+python training/plot.py
+```
+
+## Key environment variables
+
+| Variable | Default | Notes |
+|---|---|---|
+| `HF_TOKEN` | — | Required. Used for both inference (router) and Hub push |
+| `WANDB_API_KEY` | — | Optional. If set, training logs go to WandB |
+| `BASE_MODEL` | `unsloth/Qwen2.5-1.5B-Instruct-bnb-4bit` | Small enough for T4 |
+| `ENV_URL` | `http://localhost:7860` | Local server or deployed HF Space |
+| `MAX_STEPS` | `200` | GRPO steps. ~45 min on T4 |
+| `NUM_GENERATIONS` | `4` | Completions per prompt for relative ranking |
+| `LORA_R` | `16` | LoRA rank |
+| `PUSH_TO_HUB` | — | `<username>/<repo>` — push trained adapter |
+
+## What the judges look at
+
+After running, **commit** these to the repo:
+
+```
+baseline_scores.json                            # repo root
+trained_scores.json                             # repo root
+api_contract_validator/results/reward_curve.png
+api_contract_validator/results/before_after.png
+```
+
+The README's "Training Results" section reads from these files. The plots are evidence of the "Improvement in Rewards" 20% criterion.

training/baseline.py

@@ -0,0 +1,109 @@
+"""
+Baseline runner — measure the untrained model's score on every task.
+
+Writes two artefacts that the finale judges look at:
+
+    baseline_scores.json   — top-level scores per task (committed to repo)
+    api_contract_validator/results/baseline_table.md   — markdown table for README
+
+Usage:
+
+    # Make sure the env server is up first
+    docker run -d -p 7860:7860 --name baseline-env api-contract-validator
+
+    # Then run from the api_contract_validator/ directory
+    export HF_TOKEN="hf_xxxxx"
+    export API_BASE_URL="https://router.huggingface.co/v1"
+    export MODEL_NAME="Qwen/Qwen2.5-72B-Instruct"
+    python training/baseline.py
+"""
+
+import asyncio
+import json
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+# Make api_contract_validator importable when this script is run directly
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+# Load .env from api_contract_validator/ before reading os.getenv values
+try:
+    from dotenv import load_dotenv
+    _ENV_FILE = ROOT / ".env"
+    if _ENV_FILE.exists():
+        load_dotenv(_ENV_FILE)
+except ImportError:
+    pass
+
+from openai import OpenAI  # noqa: E402
+
+from client import ValidatorEnv  # noqa: E402
+from inference import (  # noqa: E402
+    BENCHMARK,
+    TASKS,
+    MODEL_NAME,
+    HF_TOKEN,
+    API_BASE_URL,
+    LOCAL_IMAGE_NAME,
+    run_single_task,
+)
+
+
+OUT_PATH = Path(os.getenv("BASELINE_OUT", ROOT.parent / "baseline_scores.json"))
+TABLE_PATH = ROOT / "results" / "baseline_table.md"
+
+
+async def main() -> None:
+    if not HF_TOKEN:
+        sys.exit("HF_TOKEN not set. Export it before running.")
+
+    openai_client = OpenAI(base_url=API_BASE_URL, api_key=HF_TOKEN)
+
+    if LOCAL_IMAGE_NAME:
+        env = await ValidatorEnv.from_docker_image(LOCAL_IMAGE_NAME)
+    else:
+        env_url = os.getenv("ENV_BASE_URL", "http://localhost:7860")
+        env = ValidatorEnv(base_url=env_url)
+
+    results = []
+    try:
+        for task in TASKS:
+            res = await run_single_task(openai_client, env, task)
+            results.append(res)
+    finally:
+        try:
+            await env.close()
+        except Exception:
+            pass
+
+    out = {
+        "model": MODEL_NAME,
+        "benchmark": BENCHMARK,
+        "date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
+        "scores": {r["task"]: r["score"] for r in results},
+        "details": results,
+    }
+    OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
+    OUT_PATH.write_text(json.dumps(out, indent=2))
+    print(f"[INFO] wrote {OUT_PATH}", flush=True)
+
+    # Markdown table for README embedding
+    lines = [
+        "| Task | Score | Steps | Success |",
+        "|---|---|---|---|",
+    ]
+    for r in results:
+        lines.append(
+            f"| `{r['task']}` | {r['score']:.2f} | {r['steps']} | "
+            f"{'✅' if r['success'] else '⛔'} |"
+        )
+    TABLE_PATH.parent.mkdir(parents=True, exist_ok=True)
+    TABLE_PATH.write_text("\n".join(lines) + "\n")
+    print(f"[INFO] wrote {TABLE_PATH}", flush=True)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

training/grpo_colab.ipynb

@@ -0,0 +1,215 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Enterprise Contract Guardian — GRPO Training\n",
+    "\n",
+    "**Meta PyTorch OpenEnv Hackathon × Scaler School of Technology — Grand Finale**\n",
+    "\n",
+    "This notebook trains a small open-weight model (Qwen2.5-1.5B by default) on the API Contract Validator environment using GRPO from TRL. The reward signal comes directly from the deployed environment, not from a static dataset — the model learns by interacting with the env on every training step.\n",
+    "\n",
+    "**Hardware**: T4 GPU (15 GB VRAM) is enough.  Colab free tier or HF Jobs `--flavor t4-small` both work.\n",
+    "\n",
+    "**Estimated runtime**: ~45 min for 200 steps on Qwen2.5-1.5B at LoRA r=16."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 1. Install dependencies"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "!pip install -q --upgrade pip\n",
+    "!pip install -q openenv-core==0.2.2 trl unsloth wandb matplotlib datasets\n",
+    "!pip install -q --no-deps bitsandbytes triton xformers"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 2. Authenticate to HuggingFace and WandB"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import os\n",
+    "from huggingface_hub import login as hf_login\n",
+    "import wandb\n",
+    "\n",
+    "# Either paste your tokens here, or `Runtime → Secrets` in Colab\n",
+    "os.environ['HF_TOKEN'] = os.getenv('HF_TOKEN') or 'hf_paste_yours_here'\n",
+    "os.environ['WANDB_API_KEY'] = os.getenv('WANDB_API_KEY') or 'paste_yours_here'\n",
+    "\n",
+    "hf_login(token=os.environ['HF_TOKEN'])\n",
+    "wandb.login(key=os.environ['WANDB_API_KEY'])"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 3. Clone the project repo and start the env server\n",
+    "\n",
+    "We start a local FastAPI server in the background. If you already have an HF Space deployed, set `ENV_URL` to the Space URL instead and skip the server start."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "!git clone https://github.com/kumarpushpam17-personal/Hackathon hack && cd hack/api_contract_validator && pip install -e .\n",
+    "%cd hack/api_contract_validator\n",
+    "\n",
+    "# Start the env server in the background\n",
+    "import subprocess, time\n",
+    "server = subprocess.Popen(\n",
+    "    ['uvicorn', 'server.app:app', '--host', '0.0.0.0', '--port', '7860'],\n",
+    "    stdout=subprocess.PIPE, stderr=subprocess.STDOUT,\n",
+    ")\n",
+    "time.sleep(8)\n",
+    "!curl -s http://localhost:7860/health"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 4. Run the baseline (untrained) to establish before-numbers\n",
+    "\n",
+    "Writes `baseline_scores.json` at the repo root. This is one of the two files the judges grade the \"Improvement in Rewards\" 20% criterion against."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "%env API_BASE_URL=https://router.huggingface.co/v1\n",
+    "%env MODEL_NAME=Qwen/Qwen2.5-72B-Instruct\n",
+    "%env BASELINE_OUT=/content/hack/baseline_scores.json\n",
+    "\n",
+    "!python training/baseline.py"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 5. Train with GRPO\n",
+    "\n",
+    "The reward function rolls out one env step per generated completion and uses the env's grader as the reward. GRPO compares the `num_generations` completions per prompt and pushes the model toward the higher-reward ones."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "%env BASE_MODEL=unsloth/Qwen2.5-1.5B-Instruct-bnb-4bit\n",
+    "%env ENV_URL=http://localhost:7860\n",
+    "%env MAX_STEPS=200\n",
+    "%env NUM_GENERATIONS=4\n",
+    "%env WANDB_PROJECT=openenv-contract-guardian\n",
+    "%env WANDB_RUN=grpo-onsite\n",
+    "%env PUSH_TO_HUB=YOUR_USERNAME/api-contract-validator-grpo\n",
+    "\n",
+    "!python training/train.py"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 6. Run inference with the trained adapter\n",
+    "\n",
+    "Set `MODEL_NAME` to the adapter you just pushed, then re-run inference and write `trained_scores.json`."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "%env MODEL_NAME=YOUR_USERNAME/api-contract-validator-grpo\n",
+    "%env SCORES_OUT_PATH=/content/hack/trained_scores.json\n",
+    "\n",
+    "!python inference.py"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 7. Generate the plots judges look at\n",
+    "\n",
+    "Writes `results/reward_curve.png` and `results/before_after.png`. Commit these to the repo — they're embedded in the README results section."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "!python training/plot.py\n",
+    "from IPython.display import Image, display\n",
+    "display(Image('results/reward_curve.png'))\n",
+    "display(Image('results/before_after.png'))"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 8. Commit the artefacts back to your fork\n",
+    "\n",
+    "From your laptop after the run is done:\n",
+    "\n",
+    "```bash\n",
+    "git add baseline_scores.json trained_scores.json \\\n",
+    "    api_contract_validator/results/reward_curve.png \\\n",
+    "    api_contract_validator/results/before_after.png\n",
+    "git commit -m 'Add baseline, trained scores and reward plots'\n",
+    "git push\n",
+    "```"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "name": "python",
+   "version": "3.10"
+  },
+  "colab": {
+   "provenance": [],
+   "gpuType": "T4"
+  },
+  "accelerator": "GPU"
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

training/plot.py

@@ -0,0 +1,109 @@
+"""
+Plot generators for the README results section.
+
+Produces two PNG files in ``results/``:
+
+    reward_curve.png   — per-step training reward (from training_state.json)
+    before_after.png   — per-task baseline vs trained score bar chart
+
+Run after both ``baseline_scores.json`` and ``trained_scores.json``
+exist::
+
+    python training/plot.py
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from typing import Dict, List
+
+import matplotlib.pyplot as plt
+import numpy as np
+
+
+ROOT = Path(__file__).resolve().parent.parent
+REPO_ROOT = ROOT.parent
+RESULTS = ROOT / "results"
+BASELINE = REPO_ROOT / "baseline_scores.json"
+TRAINED = REPO_ROOT / "trained_scores.json"
+TRAIN_STATE = RESULTS / "training_state.json"
+
+
+def _load_scores(path: Path) -> Dict[str, float]:
+    if not path.exists():
+        sys.exit(f"missing {path}")
+    return json.loads(path.read_text())["scores"]
+
+
+def plot_reward_curve() -> None:
+    if not TRAIN_STATE.exists():
+        print(f"[WARN] {TRAIN_STATE} not found — skipping reward_curve.png")
+        return
+    history = json.loads(TRAIN_STATE.read_text())
+    rows = [h for h in history if "reward" in h and "step" in h]
+    if not rows:
+        print("[WARN] no reward entries in training_state.json")
+        return
+
+    steps = [h["step"] for h in rows]
+    rewards = [h["reward"] for h in rows]
+
+    plt.figure(figsize=(8, 5))
+    plt.plot(steps, rewards, linewidth=2, color="#2563eb",
+             label="GRPO training reward")
+    if "loss" in rows[0]:
+        plt.twinx().plot(
+            steps, [h.get("loss", 0) for h in rows],
+            linewidth=1, color="#9ca3af", linestyle="--", label="loss",
+        )
+    plt.xlabel("Training step")
+    plt.ylabel("Mean episode reward")
+    plt.title("Enterprise Contract Guardian — GRPO Training")
+    plt.grid(alpha=0.3)
+    plt.legend(loc="lower right")
+    plt.tight_layout()
+    out = RESULTS / "reward_curve.png"
+    plt.savefig(out, dpi=150)
+    plt.close()
+    print(f"[INFO] wrote {out}")
+
+
+def plot_before_after() -> None:
+    baseline = _load_scores(BASELINE)
+    trained = _load_scores(TRAINED)
+
+    tasks = sorted(set(baseline) | set(trained))
+    base_vals = [baseline.get(t, 0.0) for t in tasks]
+    train_vals = [trained.get(t, 0.0) for t in tasks]
+
+    x = np.arange(len(tasks))
+    width = 0.4
+
+    plt.figure(figsize=(11, 5.5))
+    plt.bar(x - width / 2, base_vals, width, label="Baseline (untrained)",
+            color="#9ca3af")
+    plt.bar(x + width / 2, train_vals, width, label="GRPO-trained",
+            color="#16a34a")
+    plt.xticks(x, tasks, rotation=30, ha="right", fontsize=9)
+    plt.ylabel("Episode score (0–1)")
+    plt.ylim(0, 1.0)
+    plt.title("Per-task score: baseline vs trained agent")
+    plt.legend()
+    plt.grid(alpha=0.3, axis="y")
+    plt.tight_layout()
+    out = RESULTS / "before_after.png"
+    plt.savefig(out, dpi=150)
+    plt.close()
+    print(f"[INFO] wrote {out}")
+
+
+def main() -> None:
+    RESULTS.mkdir(parents=True, exist_ok=True)
+    plot_reward_curve()
+    plot_before_after()
+
+
+if __name__ == "__main__":
+    main()

training/train.py

@@ -0,0 +1,300 @@
+"""
+GRPO training script for the API Contract Validator environment.
+
+Designed to be re-runnable by judges from a Colab notebook OR via HF
+Jobs:
+
+    # HF Jobs (T4 small, ~$0.50/hr — uses your $30 credit)
+    hf jobs uv run \
+        --with trl --with unsloth --with openenv-core --with wandb \
+        --flavor t4-small \
+        -s HF_TOKEN -s WANDB_API_KEY \
+        -- python training/train.py
+
+The script:
+
+  1. Connects to a deployed HF Space (or local docker) running the env
+  2. Loads a small base model with Unsloth 4-bit quantisation
+  3. Applies LoRA adapters
+  4. Rolls out episodes through the env, collecting (prompt, completion,
+     reward) tuples
+  5. Trains the LoRA adapters with GRPO from TRL
+  6. Logs reward curves to WandB and writes results/reward_curve.png
+  7. Pushes the trained adapter to the HuggingFace Hub
+
+The reward function uses the env's grader directly — no synthetic
+shaping. This is the key difference from a static-dataset SFT run:
+the model learns from the env's verifiable signal, which is exactly
+what the hackathon's "Improvement in Rewards" criterion rewards.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, List
+
+import matplotlib.pyplot as plt
+
+# Ensure api_contract_validator is importable
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+# Load .env from api_contract_validator/ before reading os.getenv values
+try:
+    from dotenv import load_dotenv
+    _ENV_FILE = ROOT / ".env"
+    if _ENV_FILE.exists():
+        load_dotenv(_ENV_FILE)
+except ImportError:
+    pass
+
+
+# ── Configuration ────────────────────────────────────────────────────────
+
+
+@dataclass
+class TrainConfig:
+    base_model: str = os.getenv(
+        "BASE_MODEL", "unsloth/Qwen2.5-1.5B-Instruct-bnb-4bit"
+    )
+    env_url: str = os.getenv("ENV_URL", "http://localhost:7860")
+    push_to_hub_id: str | None = os.getenv("PUSH_TO_HUB", None)
+    output_dir: str = os.getenv("OUTPUT_DIR", "checkpoints/grpo")
+    results_dir: str = os.getenv(
+        "RESULTS_DIR", str(ROOT / "results")
+    )
+    seed: int = int(os.getenv("SEED", "42"))
+
+    # LoRA
+    lora_r: int = int(os.getenv("LORA_R", "16"))
+    lora_alpha: int = int(os.getenv("LORA_ALPHA", "32"))
+
+    # GRPO
+    max_seq_length: int = int(os.getenv("MAX_SEQ_LEN", "2048"))
+    num_generations: int = int(os.getenv("NUM_GENERATIONS", "4"))
+    max_steps: int = int(os.getenv("MAX_STEPS", "200"))
+    learning_rate: float = float(os.getenv("LR", "5e-6"))
+    per_device_batch_size: int = int(os.getenv("BATCH_SIZE", "1"))
+    grad_accum: int = int(os.getenv("GRAD_ACCUM", "4"))
+
+    # Tasks to train on (subset speeds up onsite training)
+    train_tasks: List[str] | None = None
+
+    # WandB
+    wandb_project: str = os.getenv("WANDB_PROJECT", "openenv-contract-guardian")
+    wandb_run: str = os.getenv("WANDB_RUN", "grpo-onsite")
+
+
+# ── Reward function: rolls out one step against the live env ─────────────
+
+
+def make_reward_fn(env_client, task_pool: List[str]):
+    """Return a TRL-compatible reward_fn that grades each completion via env.
+
+    For every (prompt, completion) pair, we ask the env to score it.
+    GRPO will then promote the higher-reward completion among the
+    ``num_generations`` samples per prompt.
+    """
+    from inference import _build_action, parse_llm_response  # noqa: WPS433
+    import asyncio
+
+    def reward_fn(prompts, completions, **kwargs):  # noqa: ARG001
+        rewards: List[float] = []
+        loop = asyncio.get_event_loop()
+        for completion in completions:
+            text = completion if isinstance(completion, str) else completion[0]["content"]
+            try:
+                action_data = parse_llm_response(text)
+                action = _build_action(action_data)
+                # one-step roll-out: reset → step → grade → reset
+                # Each prompt is associated with a fresh episode in train_dataset,
+                # so we use the env's most recent reset state as scoring context.
+                step_result = loop.run_until_complete(env_client.step(action))
+                rewards.append(float(step_result.reward or 0.0))
+            except Exception as exc:  # noqa: BLE001
+                print(f"[WARN] reward_fn error: {exc}")
+                rewards.append(-0.5)
+        return rewards
+
+    return reward_fn
+
+
+# ── Dataset: one prompt per env reset ────────────────────────────────────
+
+
+def build_train_dataset(env_client, tasks: List[str], episodes_per_task: int = 50):
+    """Roll out reset() to capture initial observations as training prompts.
+
+    Each row is one episode start. During training, GRPO samples
+    ``num_generations`` completions per prompt and uses the env to
+    grade them.
+    """
+    import asyncio
+    from datasets import Dataset  # type: ignore
+
+    from inference import build_user_prompt, _system_prompt_for_phase  # noqa
+
+    rows: List[Dict[str, Any]] = []
+    loop = asyncio.get_event_loop()
+
+    for task in tasks:
+        for ep in range(episodes_per_task):
+            seed = ep
+            result = loop.run_until_complete(
+                env_client.reset(task_name=task, seed=seed)
+            )
+            obs = result.observation.model_dump()
+            phase = obs.get("phase", "detection")
+            system = _system_prompt_for_phase(phase, task)
+            user = build_user_prompt(obs, step=1, history=[])
+            rows.append({
+                "prompt": [
+                    {"role": "system", "content": system},
+                    {"role": "user", "content": user},
+                ],
+                "task": task,
+                "seed": seed,
+            })
+
+    return Dataset.from_list(rows)
+
+
+# ── Main entry point ─────────────────────────────────────────────────────
+
+
+def main() -> None:
+    cfg = TrainConfig()
+
+    # ---- Imports happen inside main so the script can be inspected
+    # ----  without the heavy deps installed.
+    import asyncio
+    from openenv.core.client_types import StepResult  # noqa: F401
+
+    try:
+        from unsloth import FastLanguageModel  # type: ignore
+    except ImportError as exc:
+        sys.exit(
+            "unsloth not installed. Install with: pip install unsloth trl wandb. "
+            f"({exc})"
+        )
+
+    from trl import GRPOConfig, GRPOTrainer  # type: ignore
+    import wandb  # type: ignore
+
+    from client import ValidatorEnv  # noqa: WPS433
+
+    if os.getenv("WANDB_API_KEY"):
+        wandb.init(
+            project=cfg.wandb_project,
+            name=cfg.wandb_run,
+            config=cfg.__dict__,
+        )
+
+    # 1. Connect to the env
+    env = ValidatorEnv(base_url=cfg.env_url)
+    print(f"[INFO] connected to env at {cfg.env_url}")
+
+    # 2. Choose tasks
+    train_tasks = cfg.train_tasks or [
+        "find_type_mismatches",
+        "validate_nested_objects",
+        "detect_breaking_changes",
+        "validate_response_schema",
+        "trace_downstream_blast_radius",
+        "propose_backward_compat_fix",
+    ]
+
+    # 3. Build dataset
+    print(f"[INFO] building dataset for tasks={train_tasks}")
+    train_dataset = build_train_dataset(env, train_tasks)
+
+    # 4. Load model + LoRA
+    print(f"[INFO] loading model: {cfg.base_model}")
+    model, tokenizer = FastLanguageModel.from_pretrained(
+        model_name=cfg.base_model,
+        max_seq_length=cfg.max_seq_length,
+        load_in_4bit=True,
+    )
+    model = FastLanguageModel.get_peft_model(
+        model,
+        r=cfg.lora_r,
+        lora_alpha=cfg.lora_alpha,
+        target_modules=[
+            "q_proj", "k_proj", "v_proj", "o_proj",
+            "gate_proj", "up_proj", "down_proj",
+        ],
+        random_state=cfg.seed,
+    )
+
+    # 5. GRPO trainer
+    grpo_cfg = GRPOConfig(
+        output_dir=cfg.output_dir,
+        learning_rate=cfg.learning_rate,
+        per_device_train_batch_size=cfg.per_device_batch_size,
+        gradient_accumulation_steps=cfg.grad_accum,
+        num_generations=cfg.num_generations,
+        max_steps=cfg.max_steps,
+        max_prompt_length=cfg.max_seq_length // 2,
+        max_completion_length=cfg.max_seq_length // 2,
+        logging_steps=1,
+        save_steps=50,
+        report_to="wandb" if os.getenv("WANDB_API_KEY") else "none",
+        bf16=True,
+    )
+
+    reward_fn = make_reward_fn(env, train_tasks)
+
+    trainer = GRPOTrainer(
+        model=model,
+        processing_class=tokenizer,
+        reward_funcs=[reward_fn],
+        args=grpo_cfg,
+        train_dataset=train_dataset,
+    )
+
+    # 6. Train
+    print("[INFO] starting GRPO training")
+    trainer.train()
+
+    # 7. Save reward curve
+    results_dir = Path(cfg.results_dir)
+    results_dir.mkdir(parents=True, exist_ok=True)
+
+    history = [
+        h for h in trainer.state.log_history if "reward" in h
+    ]
+    if history:
+        steps = [h["step"] for h in history]
+        rewards = [h["reward"] for h in history]
+        plt.figure(figsize=(8, 5))
+        plt.plot(steps, rewards, label="train reward", linewidth=2)
+        plt.xlabel("Training step")
+        plt.ylabel("Mean episode reward")
+        plt.title("GRPO Training — Enterprise Contract Guardian")
+        plt.grid(alpha=0.3)
+        plt.legend()
+        plt.tight_layout()
+        out = results_dir / "reward_curve.png"
+        plt.savefig(out, dpi=150)
+        print(f"[INFO] wrote {out}")
+
+    # 8. Save trainer state to JSON for plot.py to consume later
+    state_path = results_dir / "training_state.json"
+    state_path.write_text(json.dumps(trainer.state.log_history, indent=2))
+    print(f"[INFO] wrote {state_path}")
+
+    # 9. Push checkpoint
+    if cfg.push_to_hub_id:
+        print(f"[INFO] pushing adapter to {cfg.push_to_hub_id}")
+        model.push_to_hub(cfg.push_to_hub_id, token=os.getenv("HF_TOKEN"))
+
+    asyncio.get_event_loop().run_until_complete(env.close())
+    print("[INFO] done.")
+
+
+if __name__ == "__main__":
+    main()