Menu
Public repository / 1812db66bd54bca3e7125b216a5a8a07efcd53c9
AkurAI-EC2 / libexec/akurai-ec2-workspace.py
Latest change 61d0777e9306a5b3b94a6987c70a8be1e718cd11 - feat: add honest workspace lifecycle boundary by Ólafur Búi Ólafsson
#!/usr/bin/env python3
"""Durable, honest workspace-lifecycle boundary for akurai-ec2.
This helper records a fenced operation before it calls a source-owned control
surface. It never treats a local record as a Drive, Notes, Tasks, or IDP
resource: unsupported source operations stay explicitly pending. DNS is the
one currently supported workspace-scoped mutation and is delegated to the
maintained ``akurai-ec2 dns`` command behind an explicit ``--apply`` gate.
"""
from __future__ import annotations
import argparse
import copy
import datetime as dt
import fcntl
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
from typing import Any, Callable
class LifecycleError(RuntimeError):
"""A request was invalid or could not safely advance."""
ACTIONS = ("provision", "deprovision", "upgrade", "health-register", "dns-handoff")
DNS_TYPES = {"A", "AAAA", "CNAME", "TXT"}
IDEMPOTENCY_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,199}\Z")
NAME_RE = re.compile(r"[^\x00-\x1f\x7f]{1,200}\Z")
DOMAIN_RE = re.compile(r"[A-Za-z0-9](?:[A-Za-z0-9.-]{0,251}[A-Za-z0-9])?\Z")
LEASE_SECONDS = 900
def now() -> str:
return dt.datetime.now(dt.UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def expires_at() -> str:
return (dt.datetime.now(dt.UTC) + dt.timedelta(seconds=LEASE_SECONDS)).replace(
microsecond=0
).isoformat().replace("+00:00", "Z")
def parse_time(value: str) -> dt.datetime:
return dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
def state_dir() -> Path:
configured = os.environ.get("AKURAI_WORKSPACE_STATE_DIR")
if configured:
return Path(configured).expanduser()
return Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local/state")) / "akurai-ec2"
class OperationStore:
"""Small, locked, atomically-written operation journal.
The lock covers only record transitions; source calls happen after the
running state and fence are durable, so a later resume can reject stale
completion from an older execution attempt.
"""
def __init__(self) -> None:
self.directory = state_dir()
self.path = self.directory / "workspace-lifecycle.json"
self.lock_path = self.directory / "workspace-lifecycle.lock"
def _default(self) -> dict[str, Any]:
return {"schema": 1, "operations": {}}
def _load(self) -> dict[str, Any]:
if not self.path.exists():
return self._default()
try:
value = json.loads(self.path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise LifecycleError(f"workspace lifecycle journal is unreadable: {exc}") from exc
if not isinstance(value, dict) or value.get("schema") != 1 or not isinstance(
value.get("operations"), dict
):
raise LifecycleError("workspace lifecycle journal has an unsupported schema")
return value
def _save(self, state: dict[str, Any]) -> None:
self.directory.mkdir(mode=0o700, parents=True, exist_ok=True)
fd, temporary = tempfile.mkstemp(prefix=".workspace-lifecycle-", dir=self.directory)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
os.fchmod(handle.fileno(), 0o600)
json.dump(state, handle, sort_keys=True, indent=2)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, self.path)
finally:
try:
os.unlink(temporary)
except FileNotFoundError:
pass
def mutate(self, operation: Callable[[dict[str, Any]], Any]) -> Any:
self.directory.mkdir(mode=0o700, parents=True, exist_ok=True)
with self.lock_path.open("a+", encoding="utf-8") as lock:
os.chmod(self.lock_path, 0o600)
fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
state = self._load()
result = operation(state)
self._save(state)
return result
def read(self, operation: Callable[[dict[str, Any]], Any]) -> Any:
self.directory.mkdir(mode=0o700, parents=True, exist_ok=True)
with self.lock_path.open("a+", encoding="utf-8") as lock:
os.chmod(self.lock_path, 0o600)
fcntl.flock(lock.fileno(), fcntl.LOCK_SH)
return operation(self._load())
def append_audit(record: dict[str, Any], event: str, **details: Any) -> None:
audit = record.setdefault("audit", [])
audit.append({"sequence": len(audit) + 1, "at": now(), "event": event, **details})
def operation_id(action: str, idempotency_key: str, request: dict[str, Any]) -> str:
source = json.dumps(
{"action": action, "idempotency_key": idempotency_key, "request": request},
sort_keys=True,
separators=(",", ":"),
)
return "wop_" + hashlib.sha256(source.encode()).hexdigest()[:24]
def validate_name(label: str, value: str) -> str:
if not NAME_RE.fullmatch(value.strip()):
raise LifecycleError(f"{label} must contain 1 through 200 printable characters")
return value.strip()
def validate_domain(value: str) -> str:
domain = value.rstrip(".").lower()
if not DOMAIN_RE.fullmatch(domain) or ".." in domain or "." not in domain:
raise LifecycleError("domain must be a fully qualified DNS name")
return domain
def pending_steps(action: str) -> list[dict[str, str]]:
if action == "provision":
return [
{
"name": "drive_workspace",
"state": "pending",
"reason": "Drive lacks an authenticated idempotent workspace create/read/delete write contract.",
},
{
"name": "notes_workspace",
"state": "pending",
"reason": "Notes lacks an authenticated workspace tenancy write contract.",
},
{
"name": "tasks_workspace",
"state": "pending",
"reason": "Tasks lacks a source-owned workspace tenancy write contract.",
},
{
"name": "idp_tenant",
"state": "pending",
"reason": "IDP lacks an authenticated idempotent tenant create/read/delete write contract; provision-oidc-client only creates app clients.",
},
{
"name": "workspace_install",
"state": "pending",
"reason": "The EC2 deploy surface is app-scoped, not workspace-scoped; no installation target contract exists.",
},
]
if action == "deprovision":
return [
{
"name": "retention_transition",
"state": "pending",
"reason": "No source-owned cancel-at-period-end or retention write contract exists.",
},
{
"name": "source_resource_delete",
"state": "pending",
"reason": "No source-owned Drive, Notes, Tasks, or IDP deletion write contract exists.",
},
]
if action == "upgrade":
return [
{
"name": "workspace_upgrade",
"state": "pending",
"reason": "The maintained EC2 deployment contract has no workspace resource selector or installed-version readback.",
}
]
if action == "health-register":
return [
{
"name": "monitor_registration",
"state": "pending",
"reason": "The maintained Monitor MCP exposes reads only; it has no workspace health-registration write contract.",
}
]
return [{"name": "dns_handoff", "state": "pending", "reason": "Requires explicit --apply."}]
def response(record: dict[str, Any]) -> dict[str, Any]:
result = copy.deepcopy(record)
result["next_action"] = next_action(result)
return result
def next_action(record: dict[str, Any]) -> str:
if record["state"] == "approval_required":
return "Approve the irreversible deletion in a separate command with workspace approve-delete."
if record["state"] == "completed":
return "No action required; the source-owned DNS handoff is complete."
if record["state"] == "compensation_pending":
return "Inspect the recorded prior DNS resource, then resume after the source is reachable."
if record["action"] == "dns-handoff":
return "Run workspace resume <operation_id> --apply only after confirming the DNS target."
return "Implement the listed source-owned write contracts, then resume with the returned operation_id and fence."
def find_operation(state: dict[str, Any], op_id: str) -> dict[str, Any]:
record = state["operations"].get(op_id)
if record is None:
raise LifecycleError(f"workspace operation not found: {op_id}")
return record
def request_payload(args: argparse.Namespace) -> dict[str, Any]:
payload: dict[str, Any] = {
"organization": validate_name("organization", args.organization),
"workspace": validate_name("workspace", args.workspace),
}
if args.action == "dns-handoff":
if not args.domain or not args.record_type or not args.record_value:
raise LifecycleError("dns-handoff requires --domain, --record-type, and --record-value")
record_type = args.record_type.upper()
if record_type not in DNS_TYPES:
raise LifecycleError("record type must be A, AAAA, CNAME, or TXT")
if "\n" in args.record_value or not args.record_value.strip():
raise LifecycleError("record value must be one non-empty line")
if not 60 <= args.ttl <= 86400:
raise LifecycleError("DNS TTL must be from 60 through 86400 seconds")
payload.update(
{
"domain": validate_domain(args.domain),
"record_type": record_type,
"record_value": args.record_value,
"ttl": args.ttl,
"zone": validate_domain(args.zone) if args.zone else None,
"provider": args.provider,
}
)
if args.action == "deprovision":
payload["irreversible"] = bool(args.irreversible)
return payload
def create_or_get(store: OperationStore, args: argparse.Namespace) -> dict[str, Any]:
if not IDEMPOTENCY_RE.fullmatch(args.idempotency_key):
raise LifecycleError("idempotency key must be 1 through 200 URL-safe characters")
if args.apply and args.action != "dns-handoff":
raise LifecycleError("--apply is currently supported only for dns-handoff")
payload = request_payload(args)
op_id = operation_id(args.action, args.idempotency_key, payload)
def create(state: dict[str, Any]) -> dict[str, Any]:
for existing_id, existing in state["operations"].items():
if existing["idempotency_key"] != args.idempotency_key:
continue
if existing_id == op_id:
append_audit(existing, "idempotency_replayed", fence=existing["fence"])
return existing
raise LifecycleError(
f"idempotency key already owns {existing_id}; use a distinct key for a different request"
)
record: dict[str, Any] = {
"operation_id": op_id,
"action": args.action,
"idempotency_key": args.idempotency_key,
"request": payload,
"state": "approval_required"
if args.action == "deprovision" and payload["irreversible"]
else "pending",
"fence": 0,
"lease_expires_at": None,
"resource_ids": {},
"steps": pending_steps(args.action),
"audit": [],
}
if args.action == "deprovision":
record["irreversible_delete"] = {
"requested": payload["irreversible"],
"approval_state": "required" if payload["irreversible"] else "not_requested",
"approval_token_sha256": None,
}
append_audit(record, "requested", state=record["state"])
state["operations"][op_id] = record
return record
record = create_or_apply(store, create, args.apply)
return response(record)
def create_or_apply(
store: OperationStore, create: Callable[[dict[str, Any]], dict[str, Any]], apply: bool
) -> dict[str, Any]:
record = store.mutate(create)
if apply and record["action"] == "dns-handoff" and record["state"] != "completed":
record = execute_dns(store, record["operation_id"])
return record
def approve_delete(store: OperationStore, op_id: str, approval_token: str) -> dict[str, Any]:
if len(approval_token) < 16:
raise LifecycleError("approval token must contain at least 16 characters")
def approve(state: dict[str, Any]) -> dict[str, Any]:
record = find_operation(state, op_id)
if record["action"] != "deprovision" or not record["request"].get("irreversible"):
raise LifecycleError("only an irreversible deprovision operation can be approved")
digest = hashlib.sha256(approval_token.encode()).hexdigest()
deletion = record["irreversible_delete"]
if deletion["approval_state"] == "approved":
if deletion["approval_token_sha256"] != digest:
raise LifecycleError("a different irreversible-delete approval was already recorded")
append_audit(record, "approval_idempotency_replayed")
return record
deletion["approval_state"] = "approved"
deletion["approval_token_sha256"] = digest
record["state"] = "pending"
append_audit(record, "irreversible_delete_approved")
return record
return response(store.mutate(approve))
def resume(store: OperationStore, op_id: str, apply: bool) -> dict[str, Any]:
def take_fence(state: dict[str, Any]) -> dict[str, Any]:
record = find_operation(state, op_id)
if record["state"] == "completed":
append_audit(record, "resume_idempotency_replayed", fence=record["fence"])
return record
if record["state"] == "approval_required":
raise LifecycleError("irreversible deletion requires workspace approve-delete before resume")
if record["state"] == "running" and record["lease_expires_at"]:
if parse_time(record["lease_expires_at"]) > dt.datetime.now(dt.UTC):
raise LifecycleError("operation has a live execution lease; wait for it or retry after it expires")
record["fence"] += 1
record["lease_expires_at"] = None
record["state"] = "pending"
append_audit(record, "resumed", fence=record["fence"])
return record
record = store.mutate(take_fence)
if apply:
if record["action"] != "dns-handoff":
raise LifecycleError("--apply is currently supported only for dns-handoff")
record = execute_dns(store, op_id)
return response(record)
def cli_args(request: dict[str, Any], action: str) -> list[str]:
args = ["dns", action, request["domain"]]
if action == "upsert":
args.extend([request["record_type"], request["record_value"], "--ttl", str(request["ttl"])])
else:
args.extend(["--type", request["record_type"]])
if action == "delete":
args.extend(["--value", request["record_value"]])
if request.get("zone"):
args.extend(["--zone", request["zone"]])
args.extend(["--provider", request["provider"]])
if action in {"upsert", "delete"}:
args.append("--yes")
return args
def run_cli(args: list[str]) -> dict[str, Any]:
cli = os.environ.get("AKURAI_EC2_CLI", "akurai-ec2")
try:
completed = subprocess.run([cli, *args], capture_output=True, text=True, check=False)
except OSError as exc:
raise LifecycleError(f"{' '.join(args[:2])}: unable to execute {cli}: {exc}") from exc
if completed.returncode:
message = completed.stderr.strip() or completed.stdout.strip() or "akurai-ec2 command failed"
raise LifecycleError(f"{' '.join(args[:2])}: {message}")
try:
return json.loads(completed.stdout)
except json.JSONDecodeError as exc:
raise LifecycleError(f"{' '.join(args[:2])}: expected JSON output") from exc
def matching_record(payload: dict[str, Any], record_value: str | None = None) -> dict[str, Any] | None:
records = payload.get("records", [])
if not isinstance(records, list):
raise LifecycleError("DNS source returned an invalid records payload")
candidates = [
value
for value in records
if isinstance(value, dict)
and value.get("fqdn", "").rstrip(".").lower() == payload["name"].rstrip(".").lower()
and value.get("type") == payload["type"]
and (record_value is None or value.get("value") == record_value)
]
if len(candidates) > 1:
raise LifecycleError("DNS handoff refuses multiple same-type records because exact compensation is ambiguous")
return candidates[0] if candidates else None
def dns_lookup(request: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any] | None]:
payload = run_cli(cli_args(request, "get"))
query = {"name": request["domain"], "type": request["record_type"]}
return payload, matching_record({**payload, **query})
def set_running(store: OperationStore, op_id: str) -> tuple[dict[str, Any], int]:
def start(state: dict[str, Any]) -> tuple[dict[str, Any], int]:
record = find_operation(state, op_id)
if record["state"] == "completed":
return record, record["fence"]
if record["state"] == "approval_required":
raise LifecycleError("irreversible deletion requires explicit approval")
if record["state"] == "running" and record["lease_expires_at"]:
if parse_time(record["lease_expires_at"]) > dt.datetime.now(dt.UTC):
raise LifecycleError("operation already has a live execution lease")
record["fence"] += 1
record["state"] = "running"
record["lease_expires_at"] = expires_at()
record["steps"][0]["state"] = "running"
append_audit(record, "dns_handoff_started", fence=record["fence"])
return copy.deepcopy(record), record["fence"]
return store.mutate(start)
def update_for_fence(
store: OperationStore, op_id: str, fence: int, update: Callable[[dict[str, Any]], None]
) -> dict[str, Any]:
def apply(state: dict[str, Any]) -> dict[str, Any]:
record = find_operation(state, op_id)
if record["fence"] != fence:
raise LifecycleError("stale execution fence; its source result was not recorded")
update(record)
return record
return store.mutate(apply)
def compensate_dns(request: dict[str, Any], prior: dict[str, Any] | None) -> tuple[bool, str]:
try:
_, current = dns_lookup(request)
if prior:
if current and current.get("value") == prior.get("value"):
return True, "original DNS record remained in place"
restore = dict(request)
restore["record_value"] = str(prior["value"])
restore["ttl"] = int(prior.get("ttl") or request["ttl"])
run_cli(cli_args(restore, "upsert"))
return True, "restored the prior DNS record"
if current and current.get("value") == request["record_value"]:
run_cli(cli_args(request, "delete"))
return True, "removed the partially created DNS record"
return True, "no DNS compensation was needed"
except LifecycleError as exc:
return False, str(exc)
def execute_dns(store: OperationStore, op_id: str) -> dict[str, Any]:
record, fence = set_running(store, op_id)
if record["state"] == "completed":
return record
request = record["request"]
prior: dict[str, Any] | None = None
try:
_, prior = dns_lookup(request)
update_for_fence(
store,
op_id,
fence,
lambda current: current.update({"prior_dns_record": prior}),
)
run_cli(cli_args(request, "upsert"))
source, created = dns_lookup(request)
if not created or created.get("value") != request["record_value"]:
raise LifecycleError("DNS upsert returned without the requested record")
def finish(current: dict[str, Any]) -> None:
current["state"] = "completed"
current["lease_expires_at"] = None
current["steps"][0]["state"] = "completed"
current["steps"][0].pop("reason", None)
current["resource_ids"] = {
"dns": {
"provider": source.get("provider"),
"zone": source.get("zone"),
"record_id": str(created.get("id")),
"fqdn": created.get("fqdn"),
"type": created.get("type"),
}
}
append_audit(current, "dns_handoff_completed", fence=fence)
return update_for_fence(store, op_id, fence, finish)
except LifecycleError as exc:
compensated, detail = compensate_dns(request, prior)
def fail(current: dict[str, Any]) -> None:
current["lease_expires_at"] = None
current["steps"][0]["state"] = "rolled_back" if compensated else "compensation_pending"
current["steps"][0]["reason"] = str(exc)
current["state"] = "rolled_back" if compensated else "compensation_pending"
append_audit(
current,
"dns_handoff_failed",
fence=fence,
error=str(exc),
compensation=detail,
)
return update_for_fence(store, op_id, fence, fail)
def list_operations(store: OperationStore) -> list[dict[str, Any]]:
return store.read(
lambda state: [
response(record)
for _, record in sorted(state["operations"].items(), key=lambda item: item[0])
]
)
def get_operation(store: OperationStore, op_id: str) -> dict[str, Any]:
return store.read(lambda state: response(find_operation(state, op_id)))
def parser() -> argparse.ArgumentParser:
root = argparse.ArgumentParser(prog="akurai-ec2 workspace")
sub = root.add_subparsers(dest="command", required=True)
request = sub.add_parser("request", help="Record one idempotent lifecycle request")
request.add_argument("action", choices=ACTIONS)
request.add_argument("--idempotency-key", required=True)
request.add_argument("--organization", required=True)
request.add_argument("--workspace", required=True)
request.add_argument("--domain")
request.add_argument("--record-type")
request.add_argument("--record-value")
request.add_argument("--zone")
request.add_argument("--provider", choices=("ec2", "1984"), default="ec2")
request.add_argument("--ttl", type=int, default=900)
request.add_argument("--irreversible", action="store_true")
request.add_argument("--apply", action="store_true")
resume_command = sub.add_parser("resume", help="Acquire a new fence and resume an operation")
resume_command.add_argument("operation_id")
resume_command.add_argument("--apply", action="store_true")
approve = sub.add_parser("approve-delete", help="Record a separate irreversible-delete approval")
approve.add_argument("operation_id")
approve.add_argument("--approval-token", required=True)
status = sub.add_parser("status", help="Read one durable lifecycle operation")
status.add_argument("operation_id")
sub.add_parser("list", help="List durable lifecycle operations")
return root
def main() -> int:
args = parser().parse_args()
store = OperationStore()
if args.command == "request":
result: Any = create_or_get(store, args)
elif args.command == "resume":
result = resume(store, args.operation_id, args.apply)
elif args.command == "approve-delete":
result = approve_delete(store, args.operation_id, args.approval_token)
elif args.command == "status":
result = get_operation(store, args.operation_id)
else:
result = list_operations(store)
print(json.dumps(result, sort_keys=True, separators=(",", ":")))
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except LifecycleError as exc:
print(json.dumps({"ok": False, "error": str(exc)}, separators=(",", ":")), file=sys.stderr)
raise SystemExit(1)