Repository navigation
Integrate genesis-alpha-mcp and drp-daf-allocator into Ouroboros cycle with Alpha telemetry - #32
Conversation
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code review is billed via overage credits. To resume reviews, an organization admin can raise the monthly limit at claude.ai/admin-settings/claude-code.
Once credits are available, push a new commit or reopen this pull request to trigger a review.
|
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| @@ -0,0 +1,8 @@ | |||
| FROM python:3.12-slim | |||
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request integrates two production MCP endpoints, 'drp-daf-allocator' and 'genesis-alpha-mcp', which share an Ouroboros state cycle, emit structured telemetry, and implement gated YOLO bypass mechanisms. The review feedback highlights several critical robustness and reliability improvements, including fixing database connection leaks in SQLite, resolving potential directory creation failures and ID collisions in the Node allocator, and adding defensive guards against null or malformed inputs in both the YOLO bypass and subgraph isomorphism functions.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| def init_db(path: Path = DB_PATH) -> Path: | ||
| path.parent.mkdir(parents=True, exist_ok=True) | ||
| with sqlite3.connect(path) as con: | ||
| con.execute(""" | ||
| CREATE TABLE IF NOT EXISTS alpha_traces ( | ||
| id INTEGER PRIMARY KEY AUTOINCREMENT, | ||
| evt_id TEXT UNIQUE NOT NULL, | ||
| schema_version TEXT NOT NULL, | ||
| record_type TEXT NOT NULL, | ||
| tags TEXT NOT NULL, | ||
| connections TEXT NOT NULL, | ||
| status TEXT NOT NULL, | ||
| payload TEXT NOT NULL, | ||
| yolo_override INTEGER NOT NULL DEFAULT 0, | ||
| thermodynamic_yield REAL NOT NULL, | ||
| crystal_score REAL NOT NULL, | ||
| ouroboros_step TEXT, | ||
| created_at TEXT NOT NULL DEFAULT (datetime('now')) | ||
| )""") | ||
| con.execute("CREATE INDEX IF NOT EXISTS idx_alpha_traces_step ON alpha_traces(ouroboros_step)") | ||
| con.execute("CREATE INDEX IF NOT EXISTS idx_alpha_traces_yolo ON alpha_traces(yolo_override)") | ||
| return path |
There was a problem hiding this comment.
Using sqlite3.connect(path) as con: as a context manager only manages transactions (committing on success, rolling back on exception). It does not close the connection. In a long-running MCP server, this will leak database connections and file descriptors, eventually leading to "Too many open files" errors or database locks.
To fix this, wrap the connection in a try...finally block to ensure con.close() is always called.
def init_db(path: Path = DB_PATH) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
con = sqlite3.connect(path)
try:
with con:
con.execute("""
CREATE TABLE IF NOT EXISTS alpha_traces (
id INTEGER PRIMARY KEY AUTOINCREMENT,
evt_id TEXT UNIQUE NOT NULL,
schema_version TEXT NOT NULL,
record_type TEXT NOT NULL,
tags TEXT NOT NULL,
connections TEXT NOT NULL,
status TEXT NOT NULL,
payload TEXT NOT NULL,
yolo_override INTEGER NOT NULL DEFAULT 0,
thermodynamic_yield REAL NOT NULL,
crystal_score REAL NOT NULL,
ouroboros_step TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)""")
con.execute("CREATE INDEX IF NOT EXISTS idx_alpha_traces_step ON alpha_traces(ouroboros_step)")
con.execute("CREATE INDEX IF NOT EXISTS idx_alpha_traces_yolo ON alpha_traces(yolo_override)")
finally:
con.close()
return path| def log_alpha_trace(record: dict[str, Any], db_path: Path = DB_PATH) -> dict[str, Any]: | ||
| init_db(db_path) | ||
| payload = record.get("payload", {}) | ||
| with sqlite3.connect(db_path) as con: | ||
| con.execute("""INSERT OR IGNORE INTO alpha_traces | ||
| (evt_id,schema_version,record_type,tags,connections,status,payload,yolo_override,thermodynamic_yield,crystal_score,ouroboros_step) | ||
| VALUES (?,?,?,?,?,?,?,?,?,?,?)""", ( | ||
| record["evt_id"], record["schema_version"], record["record_type"], canonical(record.get("tags", [])), | ||
| canonical(record.get("connections", {})), record["status"], canonical(payload), int(bool(payload.get("yolo_override"))), | ||
| float(payload.get("thermodynamic_yield", 0.0)), float(payload.get("crystal_score", 0.0)), payload.get("ouroboros_step") | ||
| )) | ||
| return {"ok": True, "evt_id": record["evt_id"], "db_path": str(db_path)} |
There was a problem hiding this comment.
Similarly, sqlite3.connect(db_path) in log_alpha_trace is used as a context manager but never closed. This will leak connection handles on every trace log. Additionally, accessing record["evt_id"], record["schema_version"], etc., directly using [] can raise a KeyError if the client passes malformed input to this MCP tool.
Using .get() with safe defaults and ensuring the connection is closed in a finally block makes this tool robust and leak-free.
| def log_alpha_trace(record: dict[str, Any], db_path: Path = DB_PATH) -> dict[str, Any]: | |
| init_db(db_path) | |
| payload = record.get("payload", {}) | |
| with sqlite3.connect(db_path) as con: | |
| con.execute("""INSERT OR IGNORE INTO alpha_traces | |
| (evt_id,schema_version,record_type,tags,connections,status,payload,yolo_override,thermodynamic_yield,crystal_score,ouroboros_step) | |
| VALUES (?,?,?,?,?,?,?,?,?,?,?)""", ( | |
| record["evt_id"], record["schema_version"], record["record_type"], canonical(record.get("tags", [])), | |
| canonical(record.get("connections", {})), record["status"], canonical(payload), int(bool(payload.get("yolo_override"))), | |
| float(payload.get("thermodynamic_yield", 0.0)), float(payload.get("crystal_score", 0.0)), payload.get("ouroboros_step") | |
| )) | |
| return {"ok": True, "evt_id": record["evt_id"], "db_path": str(db_path)} | |
| def log_alpha_trace(record: dict[str, Any], db_path: Path = DB_PATH) -> dict[str, Any]: | |
| init_db(db_path) | |
| payload = record.get("payload") or {} | |
| evt_id = record.get("evt_id") | |
| if not evt_id: | |
| raise ValueError("Record must contain 'evt_id'") | |
| con = sqlite3.connect(db_path) | |
| try: | |
| with con: | |
| con.execute("""INSERT OR IGNORE INTO alpha_traces | |
| (evt_id,schema_version,record_type,tags,connections,status,payload,yolo_override,thermodynamic_yield,crystal_score,ouroboros_step) | |
| VALUES (?,?,?,?,?,?,?,?,?,?,?)""", ( | |
| evt_id, record.get("schema_version", "1.0"), record.get("record_type", "unknown"), canonical(record.get("tags", [])), | |
| canonical(record.get("connections", {})), record.get("status", "unknown"), canonical(payload), int(bool(payload.get("yolo_override"))), | |
| float(payload.get("thermodynamic_yield", 0.0)), float(payload.get("crystal_score", 0.0)), payload.get("ouroboros_step") | |
| )) | |
| finally: | |
| con.close() | |
| return {"ok": True, "evt_id": evt_id, "db_path": str(db_path)} |
| function sha256(v){ return crypto.createHash('sha256').update(canonical(v)).digest('hex'); } | ||
| function thermo(payload){ const bits=Math.max(Buffer.byteLength(canonical(payload))*8,1); const useful=Object.keys(payload.connections||{}).length+(payload.tags||[]).length+1; return Number(Math.min(.99,useful/(useful+bits/4096)).toFixed(6)); } | ||
| function evt(record_type,status,payload,tags=[],connections={}){ const rec={evt_id:'evt_'+crypto.createHash('sha256').update(`${Date.now()}:${record_type}:${canonical(payload)}`).digest('hex').slice(0,24),schema_version:'1.0',record_type,tags:['mcp','drp-daf-allocator',...tags],connections,status,payload:{thermodynamic_yield:thermo(payload),crystal_score:.97,...payload}}; process.stderr.write('evt-'+canonical(rec)+'\n'); logAlpha(rec); return rec; } | ||
| function logAlpha(rec){ if(!process.env.ALPHA_TRACE_DB) return; const line=canonical(rec)+'\n'; fs.mkdirSync(process.env.ALPHA_TRACE_DIR || 'backend/data',{recursive:true}); fs.appendFileSync(process.env.ALPHA_TRACE_JSONL || 'backend/data/alpha_traces.drp.jsonl', line); } |
There was a problem hiding this comment.
If ALPHA_TRACE_JSONL is set to a custom path (e.g., /app/data/traces.jsonl) but ALPHA_TRACE_DIR is not set, fs.mkdirSync will default to creating backend/data (which is unused), and fs.appendFileSync will fail with ENOENT if /app/data does not exist.
We should dynamically extract the directory of the JSONL file and create it to ensure robust directory creation.
| function logAlpha(rec){ if(!process.env.ALPHA_TRACE_DB) return; const line=canonical(rec)+'\n'; fs.mkdirSync(process.env.ALPHA_TRACE_DIR || 'backend/data',{recursive:true}); fs.appendFileSync(process.env.ALPHA_TRACE_JSONL || 'backend/data/alpha_traces.drp.jsonl', line); } | |
| function logAlpha(rec){ if(!process.env.ALPHA_TRACE_DB) return; const line=canonical(rec)+'\\n'; const jsonlPath=process.env.ALPHA_TRACE_JSONL || 'backend/data/alpha_traces.drp.jsonl'; const dir=process.env.ALPHA_TRACE_DIR || (jsonlPath.includes('/') ? jsonlPath.substring(0, jsonlPath.lastIndexOf('/')) : ''); if(dir) fs.mkdirSync(dir, {recursive:true}); fs.appendFileSync(jsonlPath, line); } |
| function sortValue(v){ if(Array.isArray(v)) return v.map(sortValue); if(v && typeof v==='object') return Object.fromEntries(Object.keys(v).sort().map(k=>[k,sortValue(v[k])])); return v; } | ||
| function sha256(v){ return crypto.createHash('sha256').update(canonical(v)).digest('hex'); } | ||
| function thermo(payload){ const bits=Math.max(Buffer.byteLength(canonical(payload))*8,1); const useful=Object.keys(payload.connections||{}).length+(payload.tags||[]).length+1; return Number(Math.min(.99,useful/(useful+bits/4096)).toFixed(6)); } | ||
| function evt(record_type,status,payload,tags=[],connections={}){ const rec={evt_id:'evt_'+crypto.createHash('sha256').update(`${Date.now()}:${record_type}:${canonical(payload)}`).digest('hex').slice(0,24),schema_version:'1.0',record_type,tags:['mcp','drp-daf-allocator',...tags],connections,status,payload:{thermodynamic_yield:thermo(payload),crystal_score:.97,...payload}}; process.stderr.write('evt-'+canonical(rec)+'\n'); logAlpha(rec); return rec; } |
There was a problem hiding this comment.
Using Date.now() (millisecond resolution) to generate evt_id introduces a risk of ID collisions if multiple events are logged in the same millisecond with the same payload. Since the SQLite database uses evt_id TEXT UNIQUE NOT NULL with INSERT OR IGNORE, duplicate IDs will cause subsequent events to be silently dropped.
Using process.hrtime.bigint() (nanosecond resolution) instead of Date.now() provides high-resolution timing and matches the nanosecond precision used in the Python server (time.time_ns()).
| function evt(record_type,status,payload,tags=[],connections={}){ const rec={evt_id:'evt_'+crypto.createHash('sha256').update(`${Date.now()}:${record_type}:${canonical(payload)}`).digest('hex').slice(0,24),schema_version:'1.0',record_type,tags:['mcp','drp-daf-allocator',...tags],connections,status,payload:{thermodynamic_yield:thermo(payload),crystal_score:.97,...payload}}; process.stderr.write('evt-'+canonical(rec)+'\n'); logAlpha(rec); return rec; } | |
| function evt(record_type,status,payload,tags=[],connections={}){ const rec={evt_id:'evt_'+crypto.createHash('sha256').update(`${process.hrtime.bigint()}:${record_type}:${canonical(payload)}`).digest('hex').slice(0,24),schema_version:'1.0',record_type,tags:['mcp','drp-daf-allocator',...tags],connections,status,payload:{thermodynamic_yield:thermo(payload),crystal_score:.97,...payload}}; process.stderr.write('evt-'+canonical(rec)+'\\n'); logAlpha(rec); return rec; } |
| def trigger_yolo_swarm(headers: dict[str, str], payload: dict[str, Any] | None = None) -> dict[str, Any]: | ||
| allowed = os.environ.get("YOLO_MODE", "").lower() == "true" and bool(headers.get("x-yolo-override")) and bool(headers.get("x-alpha-token")) | ||
| status = "accepted" if allowed else "denied" | ||
| rec = evt("yolo_bypass", status, {"tool": "trigger_yolo_swarm", "yolo_override": allowed, "ouroboros_step": "synthesize_gaps", "headers_present": sorted(headers)}, ["yolo", "alpha-gated"]) | ||
| log_alpha_trace(rec) | ||
| if not allowed: | ||
| raise PermissionError("YOLO bypass requires YOLO_MODE=true plus x-yolo-override and x-alpha-token headers") | ||
| return {"ok": True, "mode": "yolo_swarm", "trace": rec, "payload": payload or {}} |
There was a problem hiding this comment.
If headers is omitted or passed as None by the client, calling headers.get will raise an AttributeError. Adding a default fallback headers = headers or {} ensures defensive handling of nullable references.
| def trigger_yolo_swarm(headers: dict[str, str], payload: dict[str, Any] | None = None) -> dict[str, Any]: | |
| allowed = os.environ.get("YOLO_MODE", "").lower() == "true" and bool(headers.get("x-yolo-override")) and bool(headers.get("x-alpha-token")) | |
| status = "accepted" if allowed else "denied" | |
| rec = evt("yolo_bypass", status, {"tool": "trigger_yolo_swarm", "yolo_override": allowed, "ouroboros_step": "synthesize_gaps", "headers_present": sorted(headers)}, ["yolo", "alpha-gated"]) | |
| log_alpha_trace(rec) | |
| if not allowed: | |
| raise PermissionError("YOLO bypass requires YOLO_MODE=true plus x-yolo-override and x-alpha-token headers") | |
| return {"ok": True, "mode": "yolo_swarm", "trace": rec, "payload": payload or {}} | |
| def trigger_yolo_swarm(headers: dict[str, str] | None = None, payload: dict[str, Any] | None = None) -> dict[str, Any]: | |
| headers = headers or {} | |
| allowed = os.environ.get("YOLO_MODE", "").lower() == "true" and bool(headers.get("x-yolo-override")) and bool(headers.get("x-alpha-token")) | |
| status = "accepted" if allowed else "denied" | |
| rec = evt("yolo_bypass", status, {"tool": "trigger_yolo_swarm", "yolo_override": allowed, "ouroboros_step": "synthesize_gaps", "headers_present": sorted(headers)}, ["yolo", "alpha-gated"]) | |
| log_alpha_trace(rec) | |
| if not allowed: | |
| raise PermissionError("YOLO bypass requires YOLO_MODE=true plus x-yolo-override and x-alpha-token headers") | |
| return {"ok": True, "mode": "yolo_swarm", "trace": rec, "payload": payload or {}} |
| def is_subgraph_isomorphic(candidate, target): | ||
| c_edges = {tuple(e) for e in candidate.get("edges", [])} | ||
| t_edges = {tuple(e) for e in target.get("edges", [])} | ||
| c_nodes = set(candidate.get("nodes", [])) | ||
| t_nodes = set(target.get("nodes", [])) | ||
| return c_nodes.issubset(t_nodes) and c_edges.issubset(t_edges) |
There was a problem hiding this comment.
If candidate or target is None or missing the expected keys, this function will raise an AttributeError or TypeError. Adding defensive guards ensures safe handling of nullable references and malformed inputs.
| def is_subgraph_isomorphic(candidate, target): | |
| c_edges = {tuple(e) for e in candidate.get("edges", [])} | |
| t_edges = {tuple(e) for e in target.get("edges", [])} | |
| c_nodes = set(candidate.get("nodes", [])) | |
| t_nodes = set(target.get("nodes", [])) | |
| return c_nodes.issubset(t_nodes) and c_edges.issubset(t_edges) | |
| def is_subgraph_isomorphic(candidate, target): | |
| if not candidate or not target: | |
| return False | |
| c_edges = {tuple(e) for e in candidate.get("edges", []) if e} | |
| t_edges = {tuple(e) for e in target.get("edges", []) if e} | |
| c_nodes = set(candidate.get("nodes", []) or []) | |
| t_nodes = set(target.get("nodes", []) or []) | |
| return c_nodes.issubset(t_nodes) and c_edges.issubset(t_edges) |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5aaea352ea
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "Codex (@codex) review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "Codex (@codex) address that feedback".
Motivation
fetch_workspace_state,feed_to_k_graph) produce identical structural invariants (order-independent SHA-256 state hashes and DiGraph isomorphism) and emit consistentevt-records.Description
mcp-registry.jsonand two MCP schemas:mcp-servers/drp-daf-allocator/drp-daf-allocator.json(Node stdio) andmcp-servers/genesis-alpha-mcp/mcp.json(podman-python) to register both endpoints.mcp-servers/drp-daf-allocator/scripts/drp-daf-allocator.mjsand light Python helpers inmcp-servers/drp-daf-allocator/src/that provide canonical hashing, structural k-graph mapping, proposal synthesis, ledger updates, cycle closure, andevt-emission to stderr plus optional delegation to Alpha trace JSONL.mcp-servers/genesis-alpha-mcp/alpha_mcp_server/main.pywith canonical JSON+SHA-256 hashing, Landauer-inspiredthermodynamic_yield,evt-records on stderr,log_alpha_traceto persist records toalpha_traces.db, stricttrigger_yolo_swarmgating requiringYOLO_MODE=trueplusx-yolo-overrideandx-alpha-token, and a ProofOfThoughthealthcheckhook.Dockerfile,alpha_mcp_init.sh) with exact volume semantics./backend/data:/app/data, documentation (docs/MCP_ALPHA_DRP_ARCHITECTURE.md), a Conductor-friendly launchGates entry for DRP, and a test suitetests/mcp/test_mcp_ouroboros.pycovering invariants, telemetry persistence, YOLO gating, DRP cycle closure, and registry discovery.Testing
python -m pytest tests/mcp/test_mcp_ouroboros.pywhich passed (5 tests, all passed).node mcp-servers/drp-daf-allocator/scripts/drp-daf-allocator.mjs) and validated expectedevt-stderr emission and tool responses, which succeeded.ALPHA_TRACE_DB=backend/data/alpha_traces.db python3 mcp-servers/genesis-alpha-mcp/alpha_mcp_server/main.py healthcheck) and confirmed DB initialization and trace insertion succeeded.Codex Task