build(agent): new-agents-4#58ba63 iteration

AGENTS.md

@@ -25,6 +25,7 @@ Architecture overview for the EquiCompiler MVP
   - equicompiler_algebraic_portfolio_dsl_to_/
     - __init__.py
   - core.py: DSL -> EquiIR compiler, lineage, attestations, and execution-plan synthesis
+  - verification.py: structured constraint parsing and deterministic verification reports
   - cli.py: CLI entry point with optional reference backtest
   - delta_sync.py: deterministic IR diffing and plan delta summaries
     - registry.py: in-process Graph-of-Contracts contract registry
@@ -57,4 +58,5 @@ Architectural Additions (MVP scaffolds)
 - Registry: A tiny in-process GraphRegistry for versioned adapters and contracts.
 - Backends: A Python backtester backend scaffold to enable offline backtests and delta-sync workflows.
 - Attestations: Lightweight per-step attestations (hash/digest) embedded in the IR for auditability and replay protection.
+- Verification: Deterministic constraint parsing and proof-digest generation for auditable plan validation.
 - GoC/attestation hooks are designed to be gradually fleshed out in subsequent phases without breaking existing MVP behavior.

README.md

@@ -1,11 +1,12 @@
 # EquiCompiler
 
-EquiCompiler translates a compact portfolio DSL into a canonical EquiIR graph with deterministic attestations, registry-backed contracts, delta-sync metadata, and a reference Python backtest path.
+EquiCompiler translates a compact portfolio DSL into a canonical EquiIR graph with deterministic attestations, structured constraint verification, registry-backed contracts, delta-sync metadata, and a reference Python backtest path.
 
 ## What It Does
 
 - Parses portfolio strategy declarations into a normalized IR.
 - Captures assets, objective clauses, constraints, execution policy, risk budgets, and regulatory metadata.
+- Normalizes constraints into a structured verification surface with deterministic proof digests.
 - Builds a deterministic execution plan with backend targets for Python, C++, and WebAssembly.
 - Emits digest-based attestations and a Graph-of-Contracts skeleton for auditability.
 - Provides a reference Python backtester for offline evaluation and replay.
@@ -44,6 +45,7 @@ Supported sections:
 - objectives and constraints
 - execution policy and risk budget
 - compiler metadata and lineage digest
+- structured constraint specifications and verification status
 - execution plan with registry contracts
 - attestations and GoC skeleton
 - delta-sync metadata
@@ -96,6 +98,7 @@ That script installs the package in editable mode, runs the unit tests, and buil
 ## Architecture Notes
 
 - `core.py`: DSL parsing, IR normalization, lineage, attestations, and plan synthesis
+- `verification.py`: structured constraint parsing and deterministic verification reports
 - `delta_sync.py`: deterministic IR diffing
 - `registry.py`: in-process GoC contract registry
 - `goc_registry.py`: GoC skeleton builder

equicompiler_algebraic_portfolio_dsl_to_/core.py

@@ -10,6 +10,7 @@ from typing import Any, Dict, Iterable, List, Sequence, Tuple, cast
 from .delta_sync import delta_sync
 from .goc_registry import build_minimal_goC_skeleton
 from .registry import GraphRegistry
+from .verification import build_verification_report, parse_constraint_surface
 
 COMPILER_NAME = "EquiCompiler"
 DEFAULT_VERSION = "0.2"
@@ -27,6 +28,7 @@ def parse_dsl_to_ir(dsl: str) -> Dict[str, object]:
     assets, asset_weights = _parse_assets(_first_value(sections, "assets"))
     objectives = _parse_list(_first_value(sections, "objectives"))
     constraints = _parse_kv_pairs(_first_value(sections, "constraints"), coerce=False)
+    structured_constraints = parse_constraint_surface(_first_value(sections, "constraints"))
     execution_policy = _parse_list(_first_value(sections, "execution_policy"))
     risk_budget = _parse_kv_pairs(_first_value(sections, "risk_budget"), coerce=True)
     market_microstructure = _parse_kv_pairs(_first_value(sections, "market_microstructure"), coerce=True)
@@ -35,6 +37,7 @@ def parse_dsl_to_ir(dsl: str) -> Dict[str, object]:
     metadata = {
         "compiler": COMPILER_NAME,
         "normalized_sections": sorted(sections.keys()),
+        "constraint_count": len(structured_constraints),
         "unknown_sections": {
             key: values
             for key, values in sections.items()
@@ -62,12 +65,14 @@ def parse_dsl_to_ir(dsl: str) -> Dict[str, object]:
         "asset_weights": asset_weights,
         "objectives": objectives,
         "constraints": constraints,
+        "constraint_spec": structured_constraints,
         "execution_policy": execution_policy,
         "risk_budget": risk_budget,
         "market_microstructure": market_microstructure,
         "regulatory": regulatory,
         "metadata": metadata,
     }
+    ir["verification"] = build_verification_report(ir, structured_constraints)
     ir["plan"] = _build_execution_plan(ir)
     ir["lineage"] = _build_lineage(ir)
     return _attach_attestations(ir)
@@ -184,6 +189,7 @@ def _build_execution_plan(ir: Dict[str, object]) -> Dict[str, object]:
                 "actions": [
                     "hash_ir",
                     "attach_attestations",
+                    "prove_constraints",
                     "build_graph_of_contracts",
                 ],
             },
@@ -202,16 +208,21 @@ def _build_execution_plan(ir: Dict[str, object]) -> Dict[str, object]:
             "asset_count": len(assets),
             "objective_count": len(objectives),
             "execution_policy_count": len(execution_policy),
+            "verification_status": cast(Dict[str, object], ir.get("verification", {})).get("status", "unknown"),
         },
     }
+
+
 def _build_lineage(ir: Dict[str, object]) -> Dict[str, object]:
     payload = {
         "version": ir.get("version"),
         "assets": ir.get("assets", []),
         "objectives": ir.get("objectives", []),
         "constraints": ir.get("constraints", {}),
+        "constraint_spec": ir.get("constraint_spec", []),
         "execution_policy": ir.get("execution_policy", []),
         "risk_budget": ir.get("risk_budget", {}),
+        "verification": ir.get("verification", {}),
     }
     digest = hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()
     return {"digest": digest, "canonical": payload}

equicompiler_algebraic_portfolio_dsl_to_/delta_sync.py

@@ -31,12 +31,14 @@ def delta_sync(old_ir: Dict[str, Any], new_ir: Dict[str, Any]) -> Dict[str, Any]
         "assets_added": sorted(list(new_assets - old_assets)),
         "assets_removed": sorted(list(old_assets - new_assets)),
         "constraints_changed": {},  # naive: compute by key presence and value change
+        "constraint_spec_changed": old_ir.get("constraint_spec") != new_ir.get("constraint_spec"),
         "risk_budget_changed": {},
         "market_microstructure_changed": {},
         "regulatory_changed": {},
         "execution_policy_added": [],
         "execution_policy_removed": [],
         "objectives_changed": {},  # {<objective_key>: {"old": <old>, "new": <new>}}
+        "verification_changed": old_ir.get("verification") != new_ir.get("verification"),
         "plan_changed": old_ir.get("plan") != new_ir.get("plan"),
         "digest": _digest(new_ir),
     }

equicompiler_algebraic_portfolio_dsl_to_/verification.py

@@ -0,0 +1,154 @@
+"""Deterministic constraint verification helpers for EquiCompiler."""
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Any, Dict, List
+
+
+def parse_constraint_surface(raw_constraints: str) -> List[Dict[str, Any]]:
+    """Parse the raw constraint clause into structured expressions."""
+    expressions: List[Dict[str, Any]] = []
+    if not raw_constraints:
+        return expressions
+
+    for token in _split_tokens(raw_constraints):
+        if not token:
+            continue
+        operator = None
+        for candidate in ("<=", ">=", "==", "=", "<", ">"):
+            if candidate in token:
+                operator = candidate
+                left, right = token.split(candidate, 1)
+                expressions.append(
+                    {
+                        "left": left.strip(),
+                        "operator": candidate,
+                        "right": _coerce_scalar(right.strip()),
+                        "source": token.strip(),
+                    }
+                )
+                break
+        if operator is None:
+            expressions.append({"left": token.strip(), "operator": "present", "right": True, "source": token.strip()})
+    return expressions
+
+
+def build_verification_report(ir: Dict[str, Any], structured_constraints: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Produce a deterministic, auditable verification report."""
+    checks: List[Dict[str, Any]] = []
+    issues: List[str] = []
+
+    assets = list(ir.get("assets", []) or [])
+    weights = dict(ir.get("asset_weights", {}) or {})
+    objectives = list(ir.get("objectives", []) or [])
+    execution_policy = list(ir.get("execution_policy", []) or [])
+    risk_budget = dict(ir.get("risk_budget", {}) or {})
+    market_microstructure = dict(ir.get("market_microstructure", {}) or {})
+
+    if not assets:
+        issues.append("no_assets_declared")
+    else:
+        checks.append({"check": "asset_declaration", "status": "pass", "asset_count": len(assets)})
+
+    if weights:
+        total_weight = round(sum(weights.get(asset, 0.0) for asset in assets), 6)
+        checks.append({"check": "weight_sum", "status": "pass" if abs(total_weight - 1.0) <= 0.001 else "warn", "value": total_weight})
+        if abs(total_weight - 1.0) > 0.001:
+            issues.append("asset_weights_do_not_sum_to_one")
+
+    if objectives:
+        checks.append({"check": "objective_presence", "status": "pass", "objective_count": len(objectives)})
+    else:
+        issues.append("no_objectives_declared")
+
+    if len(set(execution_policy)) != len(execution_policy):
+        issues.append("duplicate_execution_policy_entries")
+        checks.append({"check": "execution_policy_uniqueness", "status": "fail", "policy_count": len(execution_policy)})
+    else:
+        checks.append({"check": "execution_policy_uniqueness", "status": "pass", "policy_count": len(execution_policy)})
+
+    for constraint in structured_constraints:
+        checks.append(_verify_constraint(constraint))
+
+    for key in ("max_var", "max_exposure"):
+        if key in risk_budget:
+            value = risk_budget[key]
+            status = "pass" if isinstance(value, (int, float)) and 0.0 <= float(value) <= 1.0 else "warn"
+            checks.append({"check": f"risk_budget.{key}", "status": status, "value": value})
+            if status != "pass":
+                issues.append(f"risk_budget_{key}_out_of_range")
+
+    for key in ("latency_budget_ms", "spread_bps", "slippage_bps"):
+        if key in market_microstructure:
+            value = market_microstructure[key]
+            status = "pass" if isinstance(value, (int, float)) and float(value) >= 0 else "warn"
+            checks.append({"check": f"market_microstructure.{key}", "status": status, "value": value})
+            if status != "pass":
+                issues.append(f"market_microstructure_{key}_invalid")
+
+    canonical = {
+        "assets": assets,
+        "asset_weights": {asset: weights.get(asset) for asset in assets if asset in weights},
+        "objectives": objectives,
+        "execution_policy": execution_policy,
+        "risk_budget": risk_budget,
+        "market_microstructure": market_microstructure,
+        "constraints": structured_constraints,
+    }
+    proof_digest = hashlib.sha256(json.dumps(canonical, sort_keys=True).encode("utf-8")).hexdigest()
+    status = "verified" if not issues else "needs_review"
+    return {
+        "status": status,
+        "issues": issues,
+        "checks": checks,
+        "proof_digest": proof_digest,
+        "canonical": canonical,
+    }
+
+
+def _verify_constraint(constraint: Dict[str, Any]) -> Dict[str, Any]:
+    left = str(constraint.get("left", "")).strip()
+    operator = str(constraint.get("operator", "")).strip()
+    right = constraint.get("right")
+    status = "pass"
+    if not left:
+        status = "fail"
+    elif operator not in {"<", "<=", ">", ">=", "=", "==", "present"}:
+        status = "fail"
+    return {
+        "check": f"constraint.{left or 'unknown'}",
+        "status": status,
+        "operator": operator,
+        "value": right,
+        "source": constraint.get("source"),
+    }
+
+
+def _split_tokens(raw: str) -> List[str]:
+    tokens: List[str] = []
+    current: List[str] = []
+    for char in raw:
+        if char in {",", ";"}:
+            token = "".join(current).strip()
+            if token:
+                tokens.append(token)
+            current = []
+            continue
+        current.append(char)
+    token = "".join(current).strip()
+    if token:
+        tokens.append(token)
+    return tokens
+
+
+def _coerce_scalar(value: str) -> Any:
+    lowered = value.lower()
+    if lowered in {"true", "false"}:
+        return lowered == "true"
+    try:
+        if "." in value or "e" in lowered:
+            return float(value)
+        return int(value)
+    except ValueError:
+        return value

tests/test_compiler_pipeline.py

@@ -1,4 +1,5 @@
 import unittest
+from typing import cast
 
 from equicompiler_algebraic_portfolio_dsl_to_.backends.python_backtester import run_backtest
 from equicompiler_algebraic_portfolio_dsl_to_.core import parse_dsl_to_ir
@@ -16,7 +17,7 @@ class TestCompilerPipeline(unittest.TestCase):
             "execution_policy: offline, delta_sync, rebalance=1d\n"
             "regulatory: venue=us, suitability=true"
         )
-        ir = parse_dsl_to_ir(dsl)
+        ir = cast(dict, parse_dsl_to_ir(dsl))
         self.assertEqual(ir["version"], "0.3")
         self.assertEqual(ir["assets"], ["AAPL", "MSFT"])
         self.assertEqual(ir["asset_weights"]["AAPL"], 0.6)
@@ -25,6 +26,19 @@ class TestCompilerPipeline(unittest.TestCase):
         self.assertIn("plan", ir)
         self.assertGreaterEqual(len(ir["plan"]["phases"]), 3)
         self.assertIn("lineage", ir)
+        self.assertIn("verification", ir)
+        self.assertEqual(ir["verification"]["status"], "verified")
+        self.assertIn("prove_constraints", ir["plan"]["phases"][1]["actions"])
+
+    def test_verification_flags_invalid_weights(self):
+        ir = cast(dict, parse_dsl_to_ir(
+            "assets: AAPL@0.9, MSFT@0.2\n"
+            "objectives: maximize_return\n"
+            "constraints: max_drawdown=0.2"
+        ))
+        self.assertEqual(ir["verification"]["status"], "needs_review")
+        self.assertIn("asset_weights_do_not_sum_to_one", ir["verification"]["issues"])
+        self.assertGreaterEqual(len(ir["constraint_spec"]), 1)
 
     def test_reference_backtest_produces_metrics(self):
         ir = parse_dsl_to_ir(
@@ -39,21 +53,22 @@ class TestCompilerPipeline(unittest.TestCase):
         self.assertIsInstance(result["objective_met"], bool)
 
     def test_delta_sync_detects_plan_changes(self):
-        base = parse_dsl_to_ir(
+        base = cast(dict, parse_dsl_to_ir(
             "assets: AAPL, MSFT\n"
             "objectives: maximize_return\n"
             "constraints: max_drawdown=0.2"
-        )
-        updated = parse_dsl_to_ir(
+        ))
+        updated = cast(dict, parse_dsl_to_ir(
             "assets: AAPL, MSFT, GOOG\n"
             "objectives: maximize_return\n"
             "constraints: max_drawdown=0.15"
-        )
+        ))
         delta = delta_sync(base, updated)
         self.assertIn("assets_added", delta)
         self.assertIn("GOOG", delta["assets_added"])
         self.assertTrue(delta["plan_changed"])
         self.assertIn("constraints_changed", delta)
+        self.assertTrue(delta["verification_changed"])
 
 
 if __name__ == "__main__":