phase: 10, status: plan-as-execute, persona: backend-engineer, task: T-10.4+T-10.5+T-10.7

---ci---
project: acdl
phase: 10
milestone: v1.1
status: plan-as-execute
persona: backend-engineer
task: [T-10.4, T-10.5, T-10.7]
requirements.covered: [REQ-27]
---/ci---

Wave 2: contract + resolver + outbox writer.

- T-10.4: contracts/spike.yaml - the spike contract (stack:
  l2-static-asset, environment: dev, inputs bucket_name + region). D-P10-2:
  YAML consumer surface; the resolver parses YAML -> validates against the
  JSON contract schema.
- T-10.5: acdl_platform/contract_resolver.py - resolve(contract_path) ->
  IR instance. 7 steps: load YAML, validate contract schema, look up L2 in
  registry, load composition.json, map inputs through wires, emit IR
  instance, validate IR against ir.schema.json. Verified end-to-end:
  spike.yaml -> IR instance with kind=l2, one l1-s3 resource, validates
  against ir.schema.json.
- T-10.7: acdl_platform/outbox_writer.py - write_event(event) ->
  DynamoDB put_item. SHA-256 over canonical JSON, prev_event_hash=GENESIS
  for the first event (D-P10-3), PK contractId, SK eventType#eventTs, TTL
  expire_at = now + 365d (D-044). stdlib + boto3.

Also regenerated terraform/spike/{main.tf,terraform.tf} by running the
adapter against the resolved L2 IR (the backend key is now
spike/l2-static-asset/terraform.tfstate, derived from the stack name per
D-P10-1).

acdl_platform/contract_resolver.py

@@ -0,0 +1,119 @@
+"""ACDL Contract Resolver — resolve a contract to a Target Stack IR instance.
+
+ARCHITECTURE.md §12.8: the contract declares intent in IR-typed terms;
+the resolver resolves the contract to a target stack (list of L1
+instances + inputs + relationships); the adapter compiles the target
+stack to a plan.
+
+Steps:
+  1. Load the contract (YAML -> dict).
+  2. Validate the contract against schemas/contract.schema.json.
+  3. Look up the L2 in modules-ir/registry.json.
+  4. Load the L2's composition.json (the thin-composition tree).
+  5. Map the contract's inputs through the composition's wires to the
+     child L1's inputs.
+  6. Emit an IR instance {version, stack:{name, kind:l2, depth},
+     resources:[<L1 instances with concrete inputs>], relationships:[...]}.
+  7. Validate the IR instance against schemas/ir.schema.json.
+
+CLI: contract_resolver.py <contract.yaml> <out_ir.json>
+"""
+
+import json
+import os
+import sys
+
+import yaml
+import jsonschema
+
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _load_json(path):
+    with open(path, "r") as fh:
+        return json.load(fh)
+
+
+def resolve(contract_path, repo_root=None):
+    """Resolve a contract YAML to an IR instance dict."""
+    rr = repo_root or REPO_ROOT
+
+    # 1. Load the contract YAML.
+    with open(contract_path, "r") as fh:
+        contract = yaml.safe_load(fh)
+
+    # 2. Validate the contract against the contract schema.
+    contract_schema = _load_json(os.path.join(rr, "schemas/contract.schema.json"))
+    jsonschema.validate(contract, contract_schema)
+
+    # 3. Look up the L2 in the registry.
+    stack_name = contract["stack"]
+    registry = _load_json(os.path.join(rr, "modules-ir/registry.json"))
+    if stack_name not in registry:
+        raise ValueError(f"stack {stack_name!r} not in registry")
+    versions = registry[stack_name]
+    # Pick the highest 1.x.x (spike: just take the first non-deprecated).
+    entry = next(v for v in versions.values() if not v.get("deprecated", False))
+
+    # 4. Load the L2's composition.json.
+    composition_key = entry.get("composition") or entry.get("interface")
+    composition = _load_json(os.path.join(rr, composition_key))
+
+    # 5. Map the contract's inputs through the wires to the child L1's inputs.
+    wires = composition.get("wires", {})
+    contract_inputs = contract.get("inputs", {})
+    children = composition.get("children", [])
+
+    resources = []
+    relationships = []
+    for child in children:
+        child_id = child["id"]
+        child_module = child["module"]  # e.g. l1-s3@1.0.0
+        # Map inputs via wires whose target is this child.
+        child_inputs = {}
+        for wire_name, wire in wires.items():
+            if wire.get("target") == child_id and wire_name in contract_inputs:
+                child_inputs[wire["input"]] = contract_inputs[wire_name]
+        # Load the L1 interface to get the IR type + outputs.
+        l1_name, l1_version = child_module.split("@", 1)
+        l1_entry = registry.get(l1_name, {}).get(l1_version)
+        if not l1_entry:
+            raise ValueError(f"L1 {child_module!r} not in registry")
+        l1_iface = _load_json(os.path.join(rr, l1_entry["interface"]))
+        resources.append({
+            "id": child_id,
+            "type": l1_iface["type"],
+            "module": child_module,
+            "inputs": child_inputs,
+            "outputs": l1_iface.get("outputs", {}),
+        })
+        relationships.append({"from": "root", "to": child_id, "kind": "parent"})
+
+    # 6. Emit the IR instance.
+    ir_instance = {
+        "version": "1.0.0",
+        "stack": {
+            "name": composition["name"],
+            "kind": composition["kind"],
+            "depth": composition["depth"],
+        },
+        "resources": resources,
+        "relationships": relationships,
+    }
+
+    # 7. Validate the IR instance against the IR schema.
+    ir_schema = _load_json(os.path.join(rr, "schemas/ir.schema.json"))
+    jsonschema.validate(ir_instance, ir_schema)
+
+    return ir_instance
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 3:
+        print("usage: contract_resolver.py <contract.yaml> <out_ir.json>", file=sys.stderr)
+        sys.exit(2)
+    ir = resolve(sys.argv[1])
+    with open(sys.argv[2], "w") as fh:
+        json.dump(ir, fh, indent=2)
+    print(f"resolver: emitted IR to {sys.argv[2]}", file=sys.stderr)

acdl_platform/outbox_writer.py

@@ -0,0 +1,71 @@
+"""ACDL Outbox Writer — write an evidence event to the DynamoDB outbox.
+
+ARCHITECTURE.md §9: DynamoDB outbox, RPO=0 (synchronous write before
+ack). The event is hash-chained (SHA-256 over canonical JSON); the first
+event has prev_event_hash="GENESIS". D-P10-3: the spike writes ONE
+CONFIDENCE_COMPUTED event.
+
+The outbox table (Phase 08): acdl-outbox, PAY_PER_REQUEST, PK contractId,
+SK eventType#eventTs, TTL expire_at = now + 365d (D-044).
+
+CLI: outbox_writer.py <event.json>  (uses AWS creds from env)
+"""
+
+import datetime
+import hashlib
+import json
+import os
+import sys
+
+import boto3
+
+
+OUTBOX_TABLE = "acdl-outbox"
+REGION = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
+
+
+def _canonical_hash(event):
+    """SHA-256 over canonical JSON (sort_keys, compact separators)."""
+    canonical = json.dumps(event, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+def write_event(event, outbox_table=OUTBOX_TABLE, region=REGION):
+    """Write an evidence event to the DynamoDB outbox. Returns the item dict."""
+    contract_id = event["contractId"]
+    event_type = event.get("eventType", "CONFIDENCE_COMPUTED")
+    event_ts = event.get("ts") or datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+    sk = f"{event_type}#{event_ts}"
+
+    # Chain: first event = GENESIS (D-P10-3 spike writes one event).
+    prev_hash = event.get("prev_event_hash", "GENESIS")
+    event_hash = _canonical_hash(event)
+
+    item = {
+        "contractId": {"S": contract_id},
+        "eventType#eventTs": {"S": sk},
+        "payload": {"S": json.dumps(event, sort_keys=True)},
+        "prev_event_hash": {"S": prev_hash},
+        "hash": {"S": event_hash},
+        "environment": {"S": str(event.get("environment", ""))},
+        "stack": {"S": str(event.get("stack", ""))},
+        "score": {"N": str(event.get("score", 0))},
+        "band": {"S": str(event.get("band", ""))},
+        "expire_at": {"N": str(int((datetime.datetime.now(datetime.timezone.utc) +
+                                     datetime.timedelta(days=365)).timestamp()))},
+    }
+
+    session = boto3.Session(region_name=region)
+    dyn = session.client("dynamodb")
+    dyn.put_item(TableName=outbox_table, Item=item)
+    return item
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2:
+        print("usage: outbox_writer.py <event.json>", file=sys.stderr)
+        sys.exit(2)
+    with open(sys.argv[1], "r") as fh:
+        event = json.load(fh)
+    item = write_event(event)
+    print(json.dumps({k: list(v.values())[0] for k, v in item.items()}, indent=2))

contracts/spike.yaml

@@ -0,0 +1,5 @@
+stack: l2-static-asset
+environment: dev
+inputs:
+  bucket_name: acdl-spike-bucket
+  region: us-east-1

terraform/spike/terraform.tf

@@ -8,7 +8,7 @@ terraform {
   }
   backend "s3" {
     bucket = "acdl-tfstate-581513795199-us-east-1"
-    key    = "spike/l1-s3/terraform.tfstate"
+    key    = "spike/l2-static-asset/terraform.tfstate"
     region = "us-east-1"
   }
 }