feat(atlas-action): add Atlas Action Service (FastAPI control plane, runner, policy gate, docs)

atlas_action_service/Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+COPY requirements.txt requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY atlas_actiond atlas_actiond
+COPY config config
+
+ENV PYTHONPATH=/app
+
+CMD ["uvicorn", "atlas_actiond.main:app", "--host", "0.0.0.0", "--port", "8080"]

atlas_action_service/README.md

@@ -0,0 +1,37 @@
+# Atlas Action Service
+
+Civilization-grade internal CI/Runner platform for Atlas, designed for auditability, rollback safety, isolation, and reproducibility.
+
+## Components
+- **Control Plane**: `atlas-actiond` (FastAPI) — webhook ingestion, policy gate, queue, job spec, audit chain.
+- **Data Plane**: `runner` — isolated execution domains per job, short-lived capability tokens.
+- **Policy Gate**: guards for shipping/private-scope/L3/egress/secrets.
+- **Artifacts**: local filesystem + pluggable object storage.
+
+## Quick start (Docker)
+```bash
+cp config/example.env config/.env
+cp config/job_spec.sample.json config/job_spec.sample.json
+
+docker compose up --build
+```
+
+## API
+- `POST /webhooks/github`: ingest GitHub webhook.
+- `GET /jobs/{job_id}`: job status + artifacts.
+- `GET /healthz`: health check.
+
+## Execution Domains
+- **read**: read-only, no secrets, no network.
+- **code**: write repo, no network.
+- **network**: allowlist egress.
+- **deploy**: limited to explicit deploy policies.
+
+## Auditing
+Each job produces:
+- `job_spec.json`
+- `result.json`
+- `write_contract.json` (if any write actions)
+- `logs.txt` (structured + hash chain)
+
+All artifacts are hashed and recorded in the audit ledger.

atlas_action_service/atlas_actiond/api/jobs.py

@@ -0,0 +1,24 @@
+from pathlib import Path
+from typing import Dict
+
+from fastapi import APIRouter, HTTPException
+
+from atlas_actiond.core.config import settings
+from atlas_actiond.storage.artifacts import job_artifact_root
+
+router = APIRouter()
+
+
+@router.get("/jobs/{job_id}")
+def get_job(job_id: str) -> Dict[str, str]:
+    root = job_artifact_root(settings.atlas_action_artifact_root, job_id)
+    if not root.exists():
+        raise HTTPException(status_code=404, detail="job not found")
+
+    artifacts = {
+        "job_spec": str(root / "job_spec.json"),
+        "result": str(root / "result.json"),
+        "write_contract": str(root / "write_contract.json"),
+        "logs": str(root / "logs.txt"),
+    }
+    return {"job_id": job_id, "artifacts": artifacts}

atlas_action_service/atlas_actiond/api/webhooks.py

@@ -0,0 +1,66 @@
+import json
+import uuid
+from typing import Any, Dict
+
+from fastapi import APIRouter, Header, HTTPException, Request
+
+from atlas_actiond.core.config import settings
+from atlas_actiond.core.github import parse_github_event, verify_signature
+from atlas_actiond.core.job_spec import AtlasJobSpec, Capabilities, Commit, Event, Policy, Repository, Step
+from atlas_actiond.core.queue import JobQueue
+from atlas_actiond.policy.gate import PolicyGate
+
+router = APIRouter()
+queue = JobQueue(settings.atlas_action_redis_url)
+policy_gate = PolicyGate()
+
+
+@router.post("/webhooks/github")
+async def github_webhook(
+    request: Request,
+    x_github_event: str = Header(""),
+    x_hub_signature_256: str = Header(""),
+):
+    body = await request.body()
+    if not verify_signature(settings.atlas_action_github_webhook_secret, body, x_hub_signature_256):
+        raise HTTPException(status_code=401, detail="invalid signature")
+
+    payload = await request.json()
+    event = parse_github_event({"X-GitHub-Event": x_github_event}, payload)
+
+    job_id = str(uuid.uuid4())
+    repo = payload.get("repository", {})
+    repository = Repository(
+        owner=repo.get("owner", {}).get("login", ""),
+        name=repo.get("name", ""),
+        clone_url=repo.get("clone_url", ""),
+        default_branch=repo.get("default_branch", ""),
+    )
+    commit = Commit(sha=payload.get("after", ""), ref=payload.get("ref", ""))
+
+    job_spec = AtlasJobSpec(
+        job_id=job_id,
+        source="github",
+        repository=repository,
+        commit=commit,
+        event=Event(type=event["type"], payload=event["payload"]),
+        execution_domain="read",
+        steps=[
+            Step(name="shipping_guard", run="./scripts/check_shipping_guard.sh"),
+            Step(name="pytest", run="pytest"),
+        ],
+        policy=Policy(version=settings.atlas_action_policy_version, guards=["shipping_guard", "private_scope_guard", "l3_gate"]),
+        capabilities=Capabilities(
+            write_repo=False,
+            network_egress=False,
+            secrets_read=False,
+            allowed_egress=settings.atlas_action_allowed_egress.split(",") if settings.atlas_action_allowed_egress else [],
+        ),
+    )
+
+    gate_result = policy_gate.evaluate(job_spec, context={"event": event["type"]})
+    if not gate_result.allowed:
+        raise HTTPException(status_code=403, detail={"reasons": gate_result.reasons})
+
+    queue.enqueue("atlas.jobs", job_spec.model_dump())
+    return {"job_id": job_id, "status": "queued"}

atlas_action_service/atlas_actiond/core/config.py

@@ -0,0 +1,22 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    atlas_action_env: str = "dev"
+    atlas_action_host: str = "0.0.0.0"
+    atlas_action_port: int = 8080
+    atlas_action_secret: str = "change-me"
+    atlas_action_redis_url: str = "redis://localhost:6379/0"
+    atlas_action_artifact_root: str = "/data/artifacts"
+    atlas_action_github_app_id: str = ""
+    atlas_action_github_webhook_secret: str = "change-me"
+    atlas_action_github_token: str = ""
+    atlas_action_allowed_egress: str = "github.com,api.github.com"
+    atlas_action_policy_version: str = "2026-02-06"
+
+    class Config:
+        env_file = "config/.env"
+        env_file_encoding = "utf-8"
+
+
+settings = Settings()

atlas_action_service/atlas_actiond/core/github.py

@@ -0,0 +1,35 @@
+import hashlib
+import hmac
+import json
+from typing import Any, Dict
+
+import httpx
+
+
+def verify_signature(secret: str, body: bytes, signature: str) -> bool:
+    mac = hmac.new(secret.encode("utf-8"), msg=body, digestmod=hashlib.sha256)
+    expected = "sha256=" + mac.hexdigest()
+    return hmac.compare_digest(expected, signature)
+
+
+def parse_github_event(headers: Dict[str, str], payload: Dict[str, Any]) -> Dict[str, Any]:
+    return {
+        "type": headers.get("X-GitHub-Event", "unknown"),
+        "payload": payload,
+    }
+
+
+def post_status(token: str, repo: str, sha: str, state: str, description: str, target_url: str) -> None:
+    url = f"https://api.github.com/repos/{repo}/statuses/{sha}"
+    headers = {
+        "Authorization": f"token {token}",
+        "Accept": "application/vnd.github+json",
+        "X-GitHub-Api-Version": "2022-11-28",
+    }
+    payload = {
+        "state": state,
+        "description": description,
+        "context": "atlas-action",
+        "target_url": target_url,
+    }
+    httpx.post(url, headers=headers, json=payload, timeout=10)

atlas_action_service/atlas_actiond/core/job_spec.py

@@ -0,0 +1,58 @@
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, Field
+
+
+class Repository(BaseModel):
+    owner: str
+    name: str
+    clone_url: str
+    default_branch: Optional[str] = None
+
+
+class Commit(BaseModel):
+    sha: str
+    ref: Optional[str] = None
+
+
+class Event(BaseModel):
+    type: str
+    payload: Dict[str, Any]
+
+
+class Step(BaseModel):
+    name: str
+    run: str
+    env: Dict[str, str] = Field(default_factory=dict)
+
+
+class Policy(BaseModel):
+    version: str
+    guards: List[str]
+
+
+class Capabilities(BaseModel):
+    write_repo: bool = False
+    network_egress: bool = False
+    secrets_read: bool = False
+    allowed_egress: List[str] = Field(default_factory=list)
+
+
+class WriteContract(BaseModel):
+    diff: str
+    rollback_plan: str
+    proof: str
+    risk_score: float
+
+
+class AtlasJobSpec(BaseModel):
+    job_id: str
+    source: str
+    repository: Repository
+    commit: Commit
+    event: Event
+    execution_domain: str
+    steps: List[Step]
+    policy: Policy
+    capabilities: Capabilities
+    write_contract: Optional[WriteContract] = None

atlas_action_service/atlas_actiond/core/logging.py

@@ -0,0 +1,51 @@
+import hashlib
+import json
+import time
+from dataclasses import dataclass
+from typing import Any, Dict, Iterable
+
+
+@dataclass
+class AuditLogEntry:
+    timestamp: float
+    level: str
+    message: str
+    data: Dict[str, Any]
+    previous_hash: str
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "timestamp": self.timestamp,
+            "level": self.level,
+            "message": self.message,
+            "data": self.data,
+            "previous_hash": self.previous_hash,
+        }
+
+    def hash(self) -> str:
+        payload = json.dumps(self.to_dict(), sort_keys=True).encode("utf-8")
+        return hashlib.sha256(payload).hexdigest()
+
+
+class AuditLogger:
+    def __init__(self) -> None:
+        self._previous_hash = ""
+        self._entries: list[AuditLogEntry] = []
+
+    def log(self, level: str, message: str, **data: Any) -> AuditLogEntry:
+        entry = AuditLogEntry(
+            timestamp=time.time(),
+            level=level,
+            message=message,
+            data=data,
+            previous_hash=self._previous_hash,
+        )
+        self._previous_hash = entry.hash()
+        self._entries.append(entry)
+        return entry
+
+    def export(self) -> Iterable[Dict[str, Any]]:
+        for entry in self._entries:
+            export_entry = entry.to_dict()
+            export_entry["hash"] = entry.hash()
+            yield export_entry

atlas_action_service/atlas_actiond/core/queue.py

@@ -0,0 +1,24 @@
+import json
+from typing import Any, Dict
+
+from redis import Redis
+
+
+class JobQueue:
+    def __init__(self, redis_url: str) -> None:
+        self._redis = Redis.from_url(redis_url, decode_responses=True)
+
+    def enqueue(self, stream: str, payload: Dict[str, Any]) -> str:
+        return self._redis.xadd(stream, {"payload": json.dumps(payload)})
+
+    def read(self, stream: str, group: str, consumer: str, count: int = 1):
+        return self._redis.xreadgroup(group, consumer, {stream: ">"}, count=count, block=5000)
+
+    def ensure_group(self, stream: str, group: str) -> None:
+        try:
+            self._redis.xgroup_create(stream, group, id="0", mkstream=True)
+        except Exception:
+            pass
+
+    def ack(self, stream: str, group: str, message_id: str) -> None:
+        self._redis.xack(stream, group, message_id)

atlas_action_service/atlas_actiond/main.py

@@ -0,0 +1,14 @@
+from fastapi import FastAPI
+
+from atlas_actiond.api.jobs import router as jobs_router
+from atlas_actiond.api.webhooks import router as webhooks_router
+
+app = FastAPI(title="Atlas Action Service")
+
+app.include_router(webhooks_router)
+app.include_router(jobs_router)
+
+
+@app.get("/healthz")
+def health_check():
+    return {"status": "ok"}

atlas_action_service/atlas_actiond/policy/gate.py

@@ -0,0 +1,31 @@
+from typing import Dict, List
+
+from atlas_actiond.core.job_spec import AtlasJobSpec
+from atlas_actiond.policy.guards import (
+    EgressAllowlistGuard,
+    GuardResult,
+    L3Gate,
+    PolicyGuard,
+    PrivateScopeGuard,
+    ShippingGuard,
+    WriteContractGuard,
+)
+
+
+class PolicyGate:
+    def __init__(self) -> None:
+        self._guards: List[PolicyGuard] = [
+            ShippingGuard(),
+            PrivateScopeGuard(),
+            L3Gate(),
+            EgressAllowlistGuard(),
+            WriteContractGuard(),
+        ]
+
+    def evaluate(self, job: AtlasJobSpec, context: Dict[str, str]) -> GuardResult:
+        failures: List[str] = []
+        for guard in self._guards:
+            result = guard.evaluate(job, context)
+            if not result.allowed:
+                failures.extend([f"{guard.name}: {reason}" for reason in result.reasons])
+        return GuardResult(allowed=not failures, reasons=failures)