continuous-intelligence/acdl
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 128 Wiki Activity

feat(P53): local emulating adapters (D-092) — full local E2E, no AWS

Browse Source 
The platform is now fully locally testable without cloud credentials.
The headline E2E (contract -> resolver -> adapter -> S3 state -> ECS
service -> DynamoDB outbox -> contract-ingestor Lambda) runs end-to-end
against the local emulating tier (D-092, REQ-113).

Four local emulating adapters in core/local_emulators.py:
- FlatFileOutbox: flat-file DynamoDB outbox emulator (hash-chained JSONL;
  resumable across instances; chain verification).
- LocalEcsEmulator: local ECS Fargate HTTP 200 emulator (free-port
  binding on 127.0.0.1; health check; clean destroy).
- LocalS3StateBackend: rewrites the terraform S3 backend to a local
  backend (per-stack tfstate in a temp folder).
- LocalLambdaStub: invokes the contract_ingestor handler in-process
  (patches _get_dynamodb / _get_secrets_client / urllib.urlopen;
  DynamoDB writes redirected to the FlatFileOutbox).

run_platform.sh gains a --local flag that short-circuits to the local
emulating tier (no AWS, no Checkov, no DynamoDB).

Regression gate (D-091) now covers 12 capabilities (was 10): +CAP-011
(local E2E microservice) + CAP-012 (local E2E static-assets).

Verified: 513 fast tests pass (was 502; +11 new). 2 slow local E2E
tests pass. run_regression.sh reports 12/12 Verified. run_platform.sh
--local exits 0 with LOCAL E2E OK. No AWS credentials required.

---ci---
project: acdl
phase: 53
milestone: v1.10
status: verify
requirements:
  covered: [REQ-113]
  partial: []
decisions: [D-092]
regression:
  - { capability: CAP-011, status: Verified }
  - { capability: CAP-012, status: Verified }
---/ci---
This commit is contained in:
Jon Chery
2026-07-27 17:39:33 +00:00
parent 9897df04b2
commit 217653d6f4
9 changed files with 858 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/local_emulators.py
+489
View File
@@ -0,0 +1,489 @@
"""Local emulating adapters (D-092, REQ-113).
The platform must be fully locally testable without cloud credentials.
These adapters emulate the four cloud-backed interactions the platform
uses, so the headline E2E (contract submission -> service live ->
evidence event) runs end-to-end against the local tier with no AWS:
1. FlatFileOutbox - emulates the DynamoDB outbox (core/outbox_writer.py)
2. LocalEcsEmulator - emulates an ECS Fargate service returning HTTP 200
3. LocalS3StateBackend - rewrites the terraform S3 backend to a local backend
4. LocalLambdaStub - invokes the contract_ingestor handler in-process
Each adapter exposes the same interface as the live counterpart so the
caller code path is unchanged; only the I/O target swaps. Selection is
gated on the ACDL_LOCAL_TIER env var (set by run_platform.sh --local).
"""
from __future__ import annotations
import datetime
import hashlib
import http.server
import json
import os
import socket
import socketserver
import sys
import tempfile
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
ROOT = Path(__file__).resolve().parent.parent
def is_local_tier() -> bool:
"""True when the local emulating tier is active."""
return os.environ.get("ACDL_LOCAL_TIER", "") == "1"
# ---------------------------------------------------------------------------
# 1. Flat-file DynamoDB outbox emulator
# ---------------------------------------------------------------------------
@dataclass
class FlatFileOutbox:
"""Emulates the DynamoDB outbox with flat files in a temp folder.
Same write/read interface contract as core.outbox_writer.write_event:
accepts an event dict, returns the item dict (with a hash-chained
`hash` field). The item is appended to a JSONL file
`<dir>/outbox.jsonl` so the chain is reconstructable.
"""
dir: Path
_chain_tail_hash: str = "GENESIS"
@classmethod
def create(cls, dir: Optional[Path] = None) -> "FlatFileOutbox":
d = Path(dir) if dir else Path(tempfile.mkdtemp(prefix="acdl_outbox_"))
d.mkdir(parents=True, exist_ok=True)
out = cls(dir=d)
# Re-read the chain tail if the file already exists.
jl = d / "outbox.jsonl"
if jl.exists():
tail = None
for line in jl.read_text().splitlines():
if line.strip():
tail = json.loads(line)
if tail:
out._chain_tail_hash = tail["hash"]
return out
def _canonical_hash(self, event: Dict) -> str:
canonical = json.dumps(event, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
def write_event(self, event: Dict[str, Any],
outbox_table: str = "acdl-outbox-local",
region: str = "local") -> Dict[str, Any]:
"""Write an evidence event to the flat-file outbox.
Mirrors core.outbox_writer.write_event signature. Returns the
item dict (single-valued, not DynamoDB-typed) so the caller can
inspect it without unwrapping."""
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}"
prev_hash = event.get("prev_event_hash", self._chain_tail_hash)
event_hash = self._canonical_hash(event)
item = {
"contractId": contract_id,
"eventType#eventTs": sk,
"payload": event,
"prev_event_hash": prev_hash,
"hash": event_hash,
"environment": str(event.get("environment", "")),
"stack": str(event.get("stack", "")),
"score": event.get("score", 0),
"band": str(event.get("band", "")),
"expire_at": int((datetime.datetime.now(datetime.timezone.utc)
+ datetime.timedelta(days=365)).timestamp()),
}
jl = self.dir / "outbox.jsonl"
with jl.open("a") as f:
f.write(json.dumps(item, sort_keys=True) + "\n")
self._chain_tail_hash = event_hash
return item
def read_all(self) -> List[Dict[str, Any]]:
"""Read every event in the flat-file outbox (for verification)."""
jl = self.dir / "outbox.jsonl"
if not jl.exists():
return []
return [json.loads(line) for line in jl.read_text().splitlines()
if line.strip()]
def verify_chain(self) -> bool:
"""Verify the hash chain is intact (each prev_event_hash matches
the prior event's hash; the first event's prev is GENESIS)."""
events = self.read_all()
prev = "GENESIS"
for ev in events:
if ev["prev_event_hash"] != prev:
return False
# Recompute the hash and confirm it matches.
recomputed = self._canonical_hash(ev["payload"])
if recomputed != ev["hash"]:
return False
prev = ev["hash"]
return True
# ---------------------------------------------------------------------------
# 2. Local ECS Fargate emulator
# ---------------------------------------------------------------------------
@dataclass
class LocalEcsEmulator:
"""Emulates an ECS Fargate service by serving HTTP 200 from a local
shell process.
Records the service definition (so the caller can inspect what would
have been deployed) and starts a tiny HTTP server on a free port that
returns 200 OK for any path. The caller can then curl the endpoint to
confirm the service is "live" in the local tier.
"""
service_name: str
service_definition: Dict[str, Any]
_server: Optional[socketserver.TCPServer] = None
_thread: Optional[threading.Thread] = None
_port: int = 0
def deploy(self) -> Dict[str, Any]:
"""Start the local HTTP server; return the endpoint metadata."""
service_name = self.service_name # capture for the handler closure
class Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self, *a, **k):
body = json.dumps({
"service": service_name,
"status": "RUNNING",
"tier": "local-emulator",
"path": self.path,
}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *a, **k):
pass # silence
# Bind to a free port.
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("127.0.0.1", 0))
self._port = sock.getsockname()[1]
sock.close()
self._server = socketserver.TCPServer(
("127.0.0.1", self._port), Handler)
self._server.allow_reuse_address = True
self._thread = threading.Thread(
target=self._server.serve_forever, daemon=True)
self._thread.start()
return {
"service_arn": f"arn:local:ecs:us-east-1:000000000000:service/{self.service_name}",
"endpoint": f"http://127.0.0.1:{self._port}",
"status": "RUNNING",
"tier": "local-emulator",
"desired_count": self.service_definition.get("desired_count", 1),
"running_count": self.service_definition.get("desired_count", 1),
}
def health_check(self, endpoint: str, timeout_s: float = 5.0) -> Tuple[bool, int]:
"""curl the endpoint; return (ok, status_code)."""
import urllib.request
url = endpoint if endpoint.startswith("http") else f"http://{endpoint}"
t0 = time.monotonic()
while time.monotonic() - t0 < timeout_s:
try:
with urllib.request.urlopen(url, timeout=1.0) as r:
return (r.status == 200, r.status)
except Exception:
time.sleep(0.1)
return (False, 0)
def destroy(self):
"""Stop the local HTTP server."""
if self._server is not None:
self._server.shutdown()
self._server.server_close()
self._server = None
if self._thread is not None:
self._thread.join(timeout=2.0)
self._thread = None
# ---------------------------------------------------------------------------
# 3. Local S3 state backend (terraform backend rewrite)
# ---------------------------------------------------------------------------
@dataclass
class LocalS3StateBackend:
"""Replaces the terraform S3 backend with a local backend.
The adapter emits a `backend "s3" { ... }` block. In the local tier
we rewrite it to `backend "local" { path = "<temp>/terraform.tfstate" }`
so `terraform init/plan` runs without S3. The rewrite is applied to
the emitted terraform.tf file before terraform is invoked.
"""
state_dir: Path
@classmethod
def create(cls, dir: Optional[Path] = None) -> "LocalS3StateBackend":
d = Path(dir) if dir else Path(tempfile.mkdtemp(prefix="acdl_tfstate_"))
d.mkdir(parents=True, exist_ok=True)
return cls(state_dir=d)
def state_path(self, stack_name: str) -> Path:
return self.state_dir / f"{stack_name}.tfstate"
def rewrite_terraform_tf(self, tf_path: Path, stack_name: str) -> str:
"""Rewrite the backend block in a terraform.tf file to local.
Returns the new content (also written to disk)."""
import re
content = Path(tf_path).read_text()
# Replace the `backend "s3" { ... }` block with a local backend.
new_content = re.sub(
r'backend "s3" \{[^}]*\}',
f'backend "local" {{\n path = "{self.state_path(stack_name)}"\n }}',
content,
count=1,
flags=re.DOTALL,
)
Path(tf_path).write_text(new_content)
return new_content
# ---------------------------------------------------------------------------
# 4. Local Lambda stub (in-process handler invocation)
# ---------------------------------------------------------------------------
@dataclass
class LocalLambdaStub:
"""Invokes the contract_ingestor handler in-process.
Instead of calling AWS Lambda via boto3, this stub imports
core.lambda.contract_ingestor.lambda_handler and invokes it with a
synthesized Function-URL-style event. The DynamoDB write inside the
handler is redirected to a FlatFileOutbox so no AWS is required.
"""
outbox: FlatFileOutbox
def invoke(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Invoke the contract_ingestor handler in-process.
Returns the handler's response dict
({statusCode, body}). The handler's DynamoDB calls are
intercepted via the ACDL_LOCAL_TIER env var (the handler checks
_get_dynamodb(); under local tier it would need patching - we
patch the module's _get_dynamodb to return a local stub)."""
# Import the handler module (the dir is named `lambda`, a Python
# keyword, so use importlib instead of a dotted import).
import importlib
ci = importlib.import_module("core.lambda.contract_ingestor")
# Patch the handler's DynamoDB resource with a local stub that
# writes to the flat-file outbox. The handler uses _get_dynamodb()
# which returns a boto3 resource; we replace it with a minimal
# object exposing .Table(name) with .put_item(Item=...).
original_get = ci._get_dynamodb
class _LocalTable:
def __init__(self, name, outbox):
self.name = name
self.outbox = outbox
def put_item(self, *, TableName=None, Item=None, **kwargs):
# The handler calls put_item(TableName=..., Item=...).
# DynamoDB-typed items ({'S': ...}, {'N': ...}) are
# flattened for the flat-file outbox.
Item = Item or {}
flat = {}
for k, v in Item.items():
if isinstance(v, dict):
if "S" in v:
flat[k] = v["S"]
elif "N" in v:
flat[k] = v["N"]
else:
flat[k] = v
else:
flat[k] = v
self.outbox.write_event({
"contractId": flat.get("contractId", "local"),
"eventType": f"LAMBDA_{self.name}",
"ts": datetime.datetime.now(datetime.timezone.utc)
.strftime("%Y-%m-%dT%H:%M:%SZ"),
"environment": flat.get("environment", "local"),
"stack": self.name,
"score": 0,
"band": "local",
"prev_event_hash": "GENESIS",
})
return {}
class _LocalDynamoResource:
def __init__(self, outbox):
self.outbox = outbox
def Table(self, name):
return _LocalTable(name, self.outbox)
class _LocalSecretsClient:
def get_secret_value(self, SecretId):
return {"SecretString": json.dumps({"token": "local-stub"})}
ci._get_dynamodb = lambda: _LocalDynamoResource(self.outbox)
ci._get_secrets_client = lambda: _LocalSecretsClient()
# Stub the urllib GitHub API call so report_error doesn't hit the network.
original_urlopen = None
try:
import urllib.request
original_urlopen = urllib.request.urlopen
class _FakeResponse:
def __init__(self, body=b"{}", status=200):
self._body = body
self.status = status
def read(self):
return self._body
def __enter__(self):
return self
def __exit__(self, *a):
return False
def _fake_urlopen(url, *a, **k):
return _FakeResponse(
json.dumps([{"number": 1, "title": "stub"}]).encode())
urllib.request.urlopen = _fake_urlopen
except Exception:
pass
try:
event = {
"body": json.dumps(payload),
"requestContext": {
"httpContext": {"authorizer": {"iam": {"userId": "local-stub"}}}
},
}
result = ci.lambda_handler(event, None)
finally:
ci._get_dynamodb = original_get
if original_urlopen is not None:
import urllib.request
urllib.request.urlopen = original_urlopen
return result
# ---------------------------------------------------------------------------
# Convenience: run the headline E2E against the local tier
# ---------------------------------------------------------------------------
def run_local_e2e(contract_path: str, repo_root: Optional[Path] = None) -> Dict[str, Any]:
"""Run the headline E2E against the local emulating tier.
Steps:
1. Resolve the contract -> Target Stack.
2. Adapter compiles the stack -> terraform files (structure validated).
3. LocalS3StateBackend rewrites the backend to local.
4. LocalEcsEmulator deploys a synthetic HTTP 200 service (if the
stack has an ECS service) and confirms health.
5. FlatFileOutbox writes a CONFIDENCE_COMPUTED event; chain verified.
6. LocalLambdaStub invokes the contract_ingestor handler in-process.
Returns a dict of results. Raises AssertionError on any failure.
"""
root = Path(repo_root) if repo_root else ROOT
os.chdir(str(root))
sys.path.insert(0, str(root))
from core.contract_resolver import resolve
import adapters.terraform.adapter as adapter
stack = resolve(contract_path, str(root))
stack_name = stack["stack"]["name"]
work = Path(tempfile.mkdtemp(prefix="acdl_local_e2e_"))
tf_dir = work / "tf"
tf_dir.mkdir(exist_ok=True)
adapter.adapt(stack, str(tf_dir))
# 3. Local S3 state backend rewrite.
backend = LocalS3StateBackend.create(dir=work / "tfstate")
tf_tf = tf_dir / "terraform.tf"
backend.rewrite_terraform_tf(tf_tf, stack_name)
assert "backend \"local\"" in tf_tf.read_text(), "backend not rewritten"
# 4. Local ECS emulator (only if the stack has an ECS service).
ecs_result = None
has_ecs = any(r["type"] == "aws:ecs:service" for r in stack["resources"])
if has_ecs:
ecs = LocalEcsEmulator(
service_name=stack_name,
service_definition={"desired_count": 1},
)
deploy_meta = ecs.deploy()
ok, status = ecs.health_check(deploy_meta["endpoint"])
assert ok, f"ECS emulator health check failed: status={status}"
ecs_result = deploy_meta
ecs.destroy()
# 5. Flat-file outbox: write a CONFIDENCE_COMPUTED event + verify chain.
outbox = FlatFileOutbox.create(dir=work / "outbox")
event = {
"contractId": "local-e2e-test",
"eventType": "CONFIDENCE_COMPUTED",
"ts": datetime.datetime.now(datetime.timezone.utc)
.strftime("%Y-%m-%dT%H:%M:%SZ"),
"environment": "dev",
"stack": stack_name,
"score": 0.9,
"band": "pass",
"prev_event_hash": "GENESIS",
}
item = outbox.write_event(event)
assert item["hash"], "outbox item missing hash"
assert outbox.verify_chain(), "outbox hash chain broken"
# 6. Local Lambda stub: invoke the contract_ingestor handler.
lambda_stub = LocalLambdaStub(outbox=outbox)
lambda_result = lambda_stub.invoke({
"action": "submit_contract",
"consumerRepo": "local-test/consumer",
"contractId": "local-e2e-test",
"contract": {"module": stack_name, "environment": "dev"},
"environment": "dev",
})
assert lambda_result["statusCode"] == 200, (
f"lambda stub returned {lambda_result['statusCode']}: {lambda_result.get('body')}")
return {
"stack_name": stack_name,
"tier": "local-emulator",
"tf_dir": str(tf_dir),
"backend": "local",
"ecs": ecs_result,
"outbox_dir": str(outbox.dir),
"outbox_events": len(outbox.read_all()),
"outbox_chain_verified": True,
"lambda_status": lambda_result["statusCode"],
}
if __name__ == "__main__":
contract = sys.argv[1] if len(sys.argv) > 1 else "contracts/microservice.yaml"
os.environ["ACDL_LOCAL_TIER"] = "1"
result = run_local_e2e(contract)
print(json.dumps(result, indent=2))
core/regression_verify.py
+25
View File
@@ -266,6 +266,27 @@ def _check_run_ci_check_only() -> Tuple[Status, str]:
)
def _check_local_e2e_microservice() -> Tuple[Status, str]:
"""CAP-011: headline E2E runs against the local emulating tier (D-092).
The local tier emulates ECS, the DynamoDB outbox, S3 state, and the
contract-ingestor Lambda in-process. No AWS credentials required.
This is the local-tier half of the headline E2E; the live-AWS half
lands in Phase 54 (D-093)."""
return _check_subprocess(
["python3", "core/local_emulators.py", "contracts/microservice.yaml"],
timeout=60,
)
def _check_local_e2e_static_assets() -> Tuple[Status, str]:
"""CAP-012: local E2E on the static-assets stack (no ECS service)."""
return _check_subprocess(
["python3", "core/local_emulators.py", "contracts/static-assets.yaml"],
timeout=60,
)
# Registry: ordered, each entry is (capability_id, name, tier, check_fn).
# Phase 52 seeds this with 10 local-tier checks; Phase 54 expands it to
# cover every v1.1->v1.8 advertised capability and adds the live-AWS tier
@@ -291,6 +312,10 @@ CAPABILITY_REGISTRY: List[Tuple[str, str, str, Callable[[], Tuple[Status, str]]]
_check_pytest_offline),
("CAP-010", "run_ci.sh reproduces CI pipeline locally", "local",
_check_run_ci_check_only),
("CAP-011", "headline E2E runs against the local emulating tier (microservice)", "local",
_check_local_e2e_microservice),
("CAP-012", "local E2E on the static-assets stack (no ECS)", "local",
_check_local_e2e_static_assets),
]