Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,11 @@ is not a deepfake classifier and not cryptographic proof.
Entry for the OpenCV AI Competition 2026 (OpenCV 5 + AWS). The specification is in [SPEC.md](SPEC.md); current
status is in [docs/PROGRESS.md](docs/PROGRESS.md).

- Endpoint: see [docs/JUDGE_GUIDE.md](docs/JUDGE_GUIDE.md)
- Live app (AWS, HTTPS): https://5u7suz00bi.execute-api.us-east-1.amazonaws.com — guide in [docs/JUDGE_GUIDE.md](docs/JUDGE_GUIDE.md)
- Technical report: [docs/TECHNICAL_REPORT.md](docs/TECHNICAL_REPORT.md)
- Architecture: [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) · diagrams in [docs/diagrams/](docs/diagrams/)
- Responsible use: [docs/RESPONSIBLE_USE.md](docs/RESPONSIBLE_USE.md)
- Video: (not recorded yet — human task H8)
- Video: the link is added at submission (at most 5 minutes)

## What is in the repository

Expand All @@ -30,7 +30,7 @@ status is in [docs/PROGRESS.md](docs/PROGRESS.md).
| Camera calibration rig | `calibration_rig/` |
| AWS infrastructure (CDK v2) | `infra/` |
| Simulator and evaluation | `eval/` |
| COOL benchmark | `bench/` |
| Graviton vs x86 benchmark (COOL-ready) | `bench/` |

## Quick start (local)

Expand Down
62 changes: 57 additions & 5 deletions agent/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import json
import os
import sys
from contextlib import AsyncExitStack
from pathlib import Path
from typing import Any, Protocol

Expand All @@ -32,16 +33,65 @@ class ToolCaller(Protocol):
async def call(self, tool: str, args: dict[str, Any]) -> dict[str, Any]: ...


class ConnectionLost(Exception):
"""The MCP server process stopped answering (e.g. it was killed); the call can be retried on a new server."""


class McpCaller:
def __init__(self, session: ClientSession):
self.session = session

async def call(self, tool: str, args: dict[str, Any]) -> dict[str, Any]:
res = await self.session.call_tool(tool, args)
try:
res = await self.session.call_tool(tool, args)
except Exception as e:
raise ConnectionLost(f"{type(e).__name__}: {e}") from e
text = "".join(getattr(c, "text", "") for c in res.content)
if res.isError:
raise RuntimeError(text or f"{tool} failed")
return json.loads(text) if text else {}
if not text: # every engine tool returns a JSON object; an empty reply means the server is gone
raise ConnectionLost(f"{tool}: empty reply from MCP server")
return dict(json.loads(text))


class RestartingMcpCaller:
"""McpCaller that starts the MCP server on demand and, if the server process is lost, starts a new one and
retries the call once. Job state lives in the job store, so a new server continues the same job."""

def __init__(self, params: StdioServerParameters):
self.params = params
self.stack: AsyncExitStack | None = None
self.caller: McpCaller | None = None
self.restarts = 0

async def _start(self) -> McpCaller:
await self.aclose()
self.stack = AsyncExitStack()
read, write = await self.stack.enter_async_context(stdio_client(self.params))
session = await self.stack.enter_async_context(ClientSession(read, write))
await session.initialize()
self.caller = McpCaller(session)
return self.caller

async def call(self, tool: str, args: dict[str, Any]) -> dict[str, Any]:
for attempt in range(2):
caller = self.caller or await self._start()
try:
return await caller.call(tool, args)
except ConnectionLost:
self.caller = None
self.restarts += 1
if attempt:
raise
raise AssertionError("unreachable")

async def aclose(self) -> None:
stack, self.stack, self.caller = self.stack, None, None
if stack is not None:
try:
await stack.aclose()
except Exception: # noqa: BLE001 - the old server is already gone; nothing to clean up
pass


_PREREQ_FREE = {"probe_video", "compute_verdict", "request_user_input", "request_human_review", "render_report"}
Expand Down Expand Up @@ -126,9 +176,11 @@ async def run_with_mcp(job_id: str, planner: Planner, max_iterations: int, max_t
params = StdioServerParameters(command=sys.executable, args=["-m", "mcp_server.server"],
env={**os.environ, **env, "PYTHONPATH": f"{REPO}{os.pathsep}{REPO / 'engine'}"},
cwd=str(REPO))
async with stdio_client(params) as (read, write), ClientSession(read, write) as session:
await session.initialize()
return await run_job(job_id, planner, McpCaller(session), max_iterations, max_tool_calls, trace_path)
caller = RestartingMcpCaller(params)
try:
return await run_job(job_id, planner, caller, max_iterations, max_tool_calls, trace_path)
finally:
await caller.aclose()


def run_sync(job_id: str, planner: Planner, max_iterations: int, max_tool_calls: int, env: dict[str, str],
Expand Down
33 changes: 33 additions & 0 deletions agent/tests/test_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -263,3 +263,36 @@ def test_policy_drops_invalid_argument_values():
st.verdict_done, st.verdict = True, "INCONCLUSIVE"
a, _ = policies.enforce(Action("render", {}, "x"), st)
assert a.tool == "render_report"


def test_restarting_caller_restarts_lost_server_and_retries_once():
import asyncio

from agent.orchestrator import ConnectionLost, RestartingMcpCaller

class Dead:
async def call(self, tool, args):
raise ConnectionLost("Connection closed")

class Alive:
async def call(self, tool, args):
return {"tool": tool}

servers = [Dead(), Alive()]
c = RestartingMcpCaller(params=None) # type: ignore[arg-type]

async def start():
c.caller = servers.pop(0)
return c.caller

c._start = start # type: ignore[method-assign]
assert asyncio.run(c.call("analyze_motion", {"job_id": "j"})) == {"tool": "analyze_motion"}
assert c.restarts == 1

servers[:] = [Dead(), Dead()]
c.caller = None
try:
asyncio.run(c.call("analyze_motion", {"job_id": "j"}))
raise AssertionError("expected ConnectionLost")
except ConnectionLost:
pass
7 changes: 7 additions & 0 deletions api/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,13 @@ async def upload(job_id: str, request: Request) -> dict[str, Any]:
@app.post("/v1/jobs/{job_id}/start")
def start(job_id: str, body: StartBody) -> dict[str, Any]:
job = get_job(job_id)
if job["status"] == "created" and hasattr(ctx.storage, "uploaded_size"):
# AWS: the browser uploaded straight to S3 with the presigned URL, so the API never saw the bytes
size = ctx.storage.uploaded_size(job["upload_key"])
if size is not None:
if size > limits["max_file_bytes"]:
raise HTTPException(413, "file too large")
job.update({"status": "uploaded", "size_bytes": size})
if job["status"] not in ("uploaded", "done", "failed"):
raise HTTPException(409, f"job status is {job['status']}")
if body.sidecar is not None:
Expand Down
15 changes: 12 additions & 3 deletions api/aws_adapters.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from typing import Any

import boto3
from botocore.config import Config

from gridprint_engine.adapters import JobStoreAdapter, StorageAdapter

Expand All @@ -16,7 +17,8 @@

class S3Storage(StorageAdapter):
def __init__(self, results_bucket: str, uploads_bucket: str, region: str, expiry_s: int):
self.s3 = boto3.client("s3", region_name=region)
# SigV4 is required for presigned PUTs to a bucket with SSE-KMS default encryption
self.s3 = boto3.client("s3", region_name=region, config=Config(signature_version="s3v4"))
self.results, self.uploads, self.expiry = results_bucket, uploads_bucket, expiry_s

def _bucket(self, key: str) -> str:
Expand All @@ -36,10 +38,17 @@ def url(self, key: str) -> str:
ExpiresIn=self.expiry))

def presign_put(self, key: str) -> str:
return str(self.s3.generate_presigned_url("put_object", Params={"Bucket": self.uploads, "Key": key,
"ContentType": "video/mp4"},
# no ContentType in the signature: browsers send the file's own type (video/mp4, video/quicktime, ...)
return str(self.s3.generate_presigned_url("put_object", Params={"Bucket": self.uploads, "Key": key},
ExpiresIn=self.expiry))

def uploaded_size(self, key: str) -> int | None:
"""Size of an object uploaded directly to S3 through a presigned URL, or None if it is not there."""
try:
return int(self.s3.head_object(Bucket=self._bucket(key), Key=key)["ContentLength"])
except self.s3.exceptions.ClientError:
return None


class DynamoJobStore(JobStoreAdapter):
"""Items are stored as {pk, data(JSON string)}; tables other than Jobs/Reviews/TraceIndex go to Jobs with a
Expand Down
6 changes: 2 additions & 4 deletions bench/inputs.txt
Original file line number Diff line number Diff line change
@@ -1,4 +1,2 @@
# sha256 video sidecar gridlog-root (SYNTHETIC clips from eval/simulator/suites.py)
ab2f7e2c226048ab4b77a10156538f5f68216d69d0bc33607a2a7c24f27a8387 data/synthetic/v1/clean__s1000.mp4 data/synthetic/v1/clean__s1000.sidecar.json data/synthetic/v1/gridlog/clean__s1000
cbaac48734a453a0f3ba58f23c7bceca5cf55a67d07892a1f877291dd068376a data/synthetic/v1/camera_motion__s1060.mp4 data/synthetic/v1/camera_motion__s1060.sidecar.json data/synthetic/v1/gridlog/camera_motion__s1060
77b87f3d115f3f920e8b9b1aaabb9ee6bba662c5672021d36bb9eb98e9910257 data/synthetic/v1/moving_objects__s1070.mp4 data/synthetic/v1/moving_objects__s1070.sidecar.json data/synthetic/v1/gridlog/moving_objects__s1070
# sha256 video sidecar gridlog-root (REAL footage: public CC BY-SA 3.0 clip "Fluorescent beat effect" from Wikimedia Commons, mp4 transcode of the .ogv in docs/LICENSES.md)
b1101d4f30371132a23ddf4da29e03dc48db9731ba04b0fffb81d3afb5474db4 data/public/fluorescent_beat_effect.mp4 - data/gridlog_empty
32 changes: 22 additions & 10 deletions config/thresholds.yaml
Original file line number Diff line number Diff line change
@@ -1,16 +1,28 @@
# PROVISIONAL — development placeholders. Must be replaced by eval/calibrate_thresholds.py (R8).
provenance:
status: PROVISIONAL
generated_by: hand-written placeholder
signal:
snr_absent_db: 3.0
snr_weak_db: 8.0
snr_absent_db: 0.0
snr_weak_db: 4.0
match:
min_track_s: 20
rho1_min: 0.7
margin_min: 0.15
rho1_min: 0.8
margin_min: 0.01
continuity:
edit_significance_sigma: 6.0
edit_min_rad: 0.6
edit_significance_sigma: 10.0
edit_min_rad: 0.1
facecheck:
ratio_anomaly: 0.5
provenance:
status: PROVISIONAL
note: 'PROVISIONAL: calibrated on SYNTHETIC data only; recalibrate on the real dataset
(M9, H5)'
generated_by: eval/calibrate_thresholds.py v1.0
date: '2026-10-27T03:48:08.984059+00:00'
run_dir: results\eval\2026-10-27\synthetic-v1-fallback
calibration_clips: 22
data_sha256: ed8004b5b62e0c1b910bdb438edc8e00ddf9a837b8e5b43f44e496bbcb9de5df
target_false_contradicted_rate: 0.01
calibration_fit:
false_contradicted: 0.0
coverage: 0.5625
wrong_time: 0
edit_detection: 0.4
n_auth: 17
10 changes: 5 additions & 5 deletions docs/ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,15 +20,15 @@ The components below are the system design. All components have code in the repo
| Jobs queue | CDK in `infra/` | Amazon SQS | Decouples the API from the engine workers. Worker autoscaling uses queue depth. |
| Engine worker | `engine/`, `agent/`, `mcp_server/` | Amazon ECS on Graviton4 `c8g`, `linux/arm64`, OpenCV 5 (stock wheel); COOL AMI supported by the stack when `aws.cool_ami_id` is set | One task holds the agent orchestrator, the MCP client, and the MCP server `gridprint-engine` (stdio transport). |
| GRIDPRINT Engine | `engine/gridprint_engine/` | inside the worker | OpenCV 5 analysis stages S1–S15 (SPEC Part 8). Python package with no AWS imports. |
| GRIDPRINT Agent | `agent/` | inside the worker | Plans and adapts the analysis with engine tools. Planner: Claude on Amazon Bedrock, or the deterministic fallback planner. Never writes a verdict. |
| GRIDPRINT Agent | `agent/` | inside the worker | Plans and adapts the analysis with engine tools. Planner: an LLM on Amazon Bedrock (Amazon Nova Lite through a cross-account role, D-026), with the deterministic fallback planner for any failed model step. Never writes a verdict. |
| Reference Store | DynamoDB `GridLog` | Amazon DynamoDB | Grid frequency per zone and UTC second. Key `zone#YYYYMMDD` + epoch second. |
| GRIDPRINT Node | `node/` | ESP32 + GPS PPS at a mains socket (through a certified plug-in AC-to-AC adapter) | Measures grid frequency each UTC second. Publishes over MQTT/TLS to AWS IoT Core. |
| Ingest | `node/ingest/` | AWS Lambda (IoT topic rule) | Writes Node samples to `GridLog`. Median over nodes in one zone. |
| Auth | CDK | Amazon Cognito | User pool with the `reviewer` group. Reviewer endpoints require login. |
| Stores | CDK | Amazon DynamoDB, Amazon S3 | `Jobs`, `Reviews`, `GridLog`, `TraceIndex` tables. Upload and results buckets with SSE-KMS and lifecycle deletion. |
| Observability | CDK | Amazon CloudWatch | Structured JSON logs, metrics, dashboards (jobs, stage latency, verdict counts, agent iterations, errors, Node freshness), alarms. |
| Evaluation | `eval/` | developer machine or AWS | Simulator, dataset validator, edit generator, threshold calibration, reports into `results/`. |
| Benchmark | `bench/` | EC2 instances (configs A, B, C) | COOL vs stock OpenCV 5 on Graviton vs x86 (SPEC Part 12). |
| Benchmark | `bench/` | EC2 instances (configs A, B, C) | Stock OpenCV 5 on x86 (A) and Graviton (B), COOL on Graviton (C) (SPEC Part 12). A and B ran on 2026-10-27 (`results/bench/2026-10-27/`); C waits for the COOL subscription. |

## 3. Data flow for one job

Expand Down Expand Up @@ -81,7 +81,7 @@ The engine has no AWS imports. All access to storage and data goes through three

## 6. Agent and MCP

- The orchestrator (`agent/orchestrator.py`) runs the planner loop. The planner is `planner_bedrock.py` (Claude on Amazon Bedrock, model ID pinned in `config/default.yaml`) or `fallback_planner.py` (rules for B1–B6).
- The orchestrator (`agent/orchestrator.py`) runs the planner loop. The planner is `planner_bedrock.py` (Amazon Bedrock Converse API with tool use, model ID pinned in `config/default.yaml`) or `fallback_planner.py` (rules for B1–B6).
- `config/default.yaml` sets `agent.planner: bedrock` with `bedrock.model_id: us.amazon.nova-lite-v1:0`, called through a cross-account role (D-026). The fallback planner runs any step where the model call fails, and the trace records which planner made each decision.
- `policies.py` enforces budgets and order rules in code, not only in the prompt.
- The MCP server `gridprint-engine` runs in the same worker task as the orchestrator. Transport: stdio.
Expand All @@ -92,8 +92,8 @@ The engine has no AWS imports. All access to storage and data goes through three
- Production engine containers are `linux/arm64` and run on Graviton `c8g` instances. The API runs on Fargate `linux/arm64`.
- The `linux/amd64` engine image exists only for the x86 baseline of the COOL benchmark (SPEC Part 12, configuration A). It is not deployed in the production stack.
- Images are built by `infra/build_images.py`: the `linux/arm64` images on a Graviton EC2 build instance, from the committed tree, and pushed to Amazon ECR. The `linux/amd64` engine image (benchmark only) is built with `--amd64`. All engine images use `infra/docker/engine.Dockerfile`, the same engine code, the same pinned Python dependencies (`uv.lock`), and the same OpenCV 5 release (`opencv-python-headless==5.0.0.93`, DECISIONS D-003). Configuration C replaces stock OpenCV 5 with the COOL build on Graviton.
- The benchmark parity check compares ENF tracks and verdicts from A, B, and C.
- If COOL cannot be obtained (V5), production runs stock OpenCV 5 on Graviton, the reason is documented, and the COOL award is not claimed.
- The benchmark parity check compares ENF tracks and verdicts across configurations (A and B so far).
- COOL is not active (V5: Marketplace subscription pending, H1). Production runs stock OpenCV 5 on Graviton, and the COOL award is not claimed.

## 8. Security boundaries

Expand Down
Loading