#!/usr/bin/env python3 """Darkoza Existence Possibility Compiler (EPC). A deterministic, bounded planner that values both achieved outcomes and the future options unlocked by a reversible experiment. It plans only: it never executes tools, spends money, changes files, or grants authority. """ from __future__ import annotations import argparse from collections import deque from datetime import datetime, timezone import hashlib import json import math from pathlib import Path import sys from typing import Any VERSION = "darkoza-epc-0.1" SCHEMA = "darkoza.existence-problem.v1" DIMENSIONS = ("discover", "create", "preserve", "evolve") EVIDENCE_STATES = {"verified", "unverified", "hypothesis"} EVIDENCE_GRADES = {"measured", "tested", "pilot", "hypothesis", "unknown"} MAX_ACTIONS = 16 MAX_SEARCH_STATES = 50000 def canonical(value: Any) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode("utf-8") def digest(value: Any) -> str: return hashlib.sha256(canonical(value)).hexdigest() def number(value: Any, name: str, low: float, high: float) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError(f"{name} must be numeric") if not math.isfinite(value) or value < low or value > high: raise ValueError(f"{name} must be between {low} and {high}") return float(value) def string_list(value: Any, name: str, *, required: bool = False) -> list[str]: if not isinstance(value, list) or (required and not value): raise ValueError(f"{name} must be a {'non-empty ' if required else ''}list") if any(not isinstance(x, str) or not x.strip() for x in value): raise ValueError(f"{name} entries must be non-empty strings") if len(set(value)) != len(value): raise ValueError(f"{name} entries must be unique") return value def validate(problem: dict[str, Any]) -> tuple[dict, list[dict], dict, dict]: if not isinstance(problem, dict) or problem.get("schema") != SCHEMA: raise ValueError(f"schema must equal {SCHEMA}") resources = problem.get("resource_budget") if not isinstance(resources, dict) or not resources: raise ValueError("resource_budget must be a non-empty object") for key, val in resources.items(): if not isinstance(key, str) or not key.strip(): raise ValueError("resource names must be non-empty strings") number(val, f"resource_budget.{key}", 0, 1e12) weights = problem.get("mission_weights", {}) if not isinstance(weights, dict) or set(weights) - set(DIMENSIONS): raise ValueError("mission_weights may contain only discover/create/preserve/evolve") normalized = {d: number(weights.get(d, 1), f"mission_weights.{d}", 0, 1e6) for d in DIMENSIONS} total = sum(normalized.values()) if total <= 0: raise ValueError("at least one mission weight must be positive") normalized = {k: v / total for k, v in normalized.items()} current = string_list(problem.get("current_capabilities", []), "current_capabilities") targets = string_list(problem.get("target_capabilities", []), "target_capabilities") observations = problem.get("observations", []) if not isinstance(observations, list): raise ValueError("observations must be a list") obs_by_id = {} for obs in observations: if not isinstance(obs, dict): raise ValueError("each observation must be an object") oid = obs.get("id") if not isinstance(oid, str) or not oid.strip() or oid in obs_by_id: raise ValueError("observation ids must be non-empty and unique") if obs.get("status") not in EVIDENCE_STATES: raise ValueError(f"{oid}: invalid observation status") if obs["status"] == "verified" and not ( isinstance(obs.get("source"), str) and obs["source"].strip() ): raise ValueError(f"{oid}: verified observations require a source") obs_by_id[oid] = obs actions = problem.get("actions") if not isinstance(actions, list) or not actions: raise ValueError("actions must be a non-empty list") if len(actions) > MAX_ACTIONS: raise ValueError(f"at most {MAX_ACTIONS} actions are supported in v0.1") seen = set() clean = [] for raw in actions: if not isinstance(raw, dict): raise ValueError("each action must be an object") a = dict(raw) aid = a.get("id") if not isinstance(aid, str) or not aid.strip() or aid in seen: raise ValueError("action ids must be non-empty and unique") seen.add(aid) for key in ("requires", "provides", "requires_observations", "conflicts_with", "preconditions", "verification", "success_metrics"): a[key] = string_list(a.get(key, []), f"{aid}.{key}", required=(key == "provides")) if not isinstance(a.get("title"), str) or not a["title"].strip(): raise ValueError(f"{aid}: title is required") for key in ("rollback",): if not isinstance(a.get(key), str) or not a[key].strip(): raise ValueError(f"{aid}: {key} is required") if type(a.get("reversible")) is not bool or type(a.get("approval_required", False)) is not bool: raise ValueError(f"{aid}: reversible and approval_required must be boolean") a["approval_required"] = a.get("approval_required", False) a["confidence"] = number(a.get("confidence"), f"{aid}.confidence", 0, 1) a["risk"] = number(a.get("risk"), f"{aid}.risk", 0, 1) if a.get("evidence_grade") not in EVIDENCE_GRADES: raise ValueError(f"{aid}: invalid evidence_grade") effects = a.get("effects") if not isinstance(effects, dict) or set(effects) - set(DIMENSIONS): raise ValueError(f"{aid}: effects may contain only {DIMENSIONS}") a["effects"] = {d: number(effects.get(d, 0), f"{aid}.effects.{d}", 0, 10) for d in DIMENSIONS} costs = a.get("cost", {}) if not isinstance(costs, dict) or set(costs) - set(resources): raise ValueError(f"{aid}: cost must use declared resource_budget keys") a["cost"] = {k: number(costs.get(k, 0), f"{aid}.cost.{k}", 0, 1e12) for k in resources} unknown_obs = set(a["requires_observations"]) - set(obs_by_id) if unknown_obs: raise ValueError(f"{aid}: unknown observation ids: {sorted(unknown_obs)}") clean.append(a) constraints = problem.get("constraints", {}) if not isinstance(constraints, dict): raise ValueError("constraints must be an object") constraints = { "max_risk_per_action": number(constraints.get("max_risk_per_action", 0.65), "max_risk_per_action", 0, 1), "max_total_risk": number(constraints.get("max_total_risk", 1.0), "max_total_risk", 0, 16), "max_plan_steps": int(number(constraints.get("max_plan_steps", 3), "max_plan_steps", 1, 8)), "require_reversible": constraints.get("require_reversible", True), "allow_approval_required": constraints.get("allow_approval_required", False), } if type(constraints["require_reversible"]) is not bool or type(constraints["allow_approval_required"]) is not bool: raise ValueError("reversibility and approval constraints must be boolean") return ({"resources": resources, "weights": normalized, "current": current, "targets": targets, "observations": obs_by_id}, sorted(clean, key=lambda x: x["id"]), constraints, problem) def impact(action: dict, weights: dict) -> float: return sum(weights[d] * action["effects"][d] for d in DIMENSIONS) def assess(action: dict, capabilities: set[str], selected: set[str], used: dict[str, float], risk_used: float, resources: dict, observations: dict, constraints: dict, by_id: dict) -> list[str]: reasons = [] missing = sorted(set(action["requires"]) - capabilities) if missing: reasons.append("missing_capabilities:" + ",".join(missing)) unverified = [oid for oid in action["requires_observations"] if observations[oid]["status"] != "verified"] if unverified: reasons.append("unverified_observations:" + ",".join(sorted(unverified))) if action["risk"] > constraints["max_risk_per_action"]: reasons.append("per_action_risk_limit") if risk_used + action["risk"] > constraints["max_total_risk"] + 1e-12: reasons.append("cumulative_risk_limit") if constraints["require_reversible"] and not action["reversible"]: reasons.append("reversibility_required") if (action["approval_required"] or not action["reversible"]) and not constraints["allow_approval_required"]: reasons.append("approval_not_permitted") if any(x in selected for x in action["conflicts_with"]): reasons.append("conflicts_with_selected_action") for other_id in selected: if action["id"] in by_id[other_id]["conflicts_with"]: reasons.append("selected_action_conflicts_with_this") break for key, budget in resources.items(): if used[key] + action["cost"][key] > budget + 1e-12: reasons.append("resource_limit:" + key) return reasons def enabled_actions(actions: list[dict], capabilities: set[str], selected: set[str], used: dict, risk_used: float, resources: dict, observations: dict, constraints: dict, by_id: dict) -> list[dict]: result = [] for a in actions: if a["id"] in selected: continue if not assess(a, capabilities, selected, used, risk_used, resources, observations, constraints, by_id): result.append(a) return result def summarize(selected_ids: tuple[str, ...], caps: set[str], used: dict[str, float], risk_used: float, actions: list[dict], by_id: dict, base_enabled: set[str], spec: dict, constraints: dict) -> dict: chosen = [by_id[x] for x in selected_ids] direct = sum(impact(a, spec["weights"]) * a["confidence"] for a in chosen) uncertainty = sum(impact(a, spec["weights"]) * (1 - a["confidence"]) for a in chosen) irreversible = sum(not a["reversible"] for a in chosen) approvals = sum(a["approval_required"] or not a["reversible"] for a in chosen) used_fraction = [used[k] / v for k, v in spec["resources"].items() if v > 0] resource_burden = (sum(used_fraction) / len(used_fraction)) if used_fraction else 0.0 resource_penalty = resource_burden * 1.5 post_enabled = enabled_actions(actions, caps, set(selected_ids), used, risk_used, spec["resources"], spec["observations"], constraints, by_id) newly_unlocked = [a for a in post_enabled if a["id"] not in base_enabled] option_value = sum(impact(a, spec["weights"]) * a["confidence"] / (1 + 2 * a["risk"]) for a in newly_unlocked) option_bonus = 2.0 * math.log1p(option_value) target_hits = sorted(set(spec["targets"]) & caps) target_progress = (len(target_hits) / len(spec["targets"])) if spec["targets"] else 0.0 target_bonus = 4.0 * target_progress risk_penalty = 4.0 * risk_used uncertainty_penalty = 1.5 * uncertainty reversibility_penalty = 3.0 * irreversible approval_penalty = 1.0 * approvals score = (direct + option_bonus + target_bonus - risk_penalty - uncertainty_penalty - resource_penalty - reversibility_penalty - approval_penalty) return { "actions": list(selected_ids), "capabilities_after": sorted(caps), "target_capabilities_reached": target_hits, "target_progress": round(target_progress, 4), "direct_value": round(direct, 4), "newly_unlocked_options": [ {"id": a["id"], "title": a["title"], "weighted_potential": round(impact(a, spec["weights"]), 4), "minimum_next_step": True} for a in newly_unlocked ], "new_option_value": round(option_value, 4), "risk_total": round(risk_used, 4), "resource_use": {k: {"used": round(v, 6), "budget": spec["resources"][k]} for k, v in used.items()}, "resource_burden": round(resource_burden, 6), "score_components": { "option_bonus": round(option_bonus, 4), "risk_penalty": round(risk_penalty, 4), "uncertainty_penalty": round(uncertainty_penalty, 4), "resource_penalty": round(resource_penalty, 4), "irreversibility_penalty": round(reversibility_penalty, 4), "approval_penalty": round(approval_penalty, 4), "target_bonus": round(target_bonus, 4), }, "score": round(score, 4), "requires_human_approval": bool(approvals), } def dominates(a: dict, b: dict) -> bool: a_vals = (a["direct_value"], a["new_option_value"], a["target_progress"], -a["risk_total"], -a["resource_burden"]) b_vals = (b["direct_value"], b["new_option_value"], b["target_progress"], -b["risk_total"], -b["resource_burden"]) return all(x >= y for x, y in zip(a_vals, b_vals)) and any(x > y for x, y in zip(a_vals, b_vals)) def compile_plan(problem: dict, *, now: datetime | None = None) -> dict: spec, actions, constraints, original = validate(problem) by_id = {a["id"]: a for a in actions} start_caps = set(spec["current"]) initial_used = {k: 0.0 for k in spec["resources"]} initial_enabled = enabled_actions(actions, start_caps, set(), initial_used, 0.0, spec["resources"], spec["observations"], constraints, by_id) base_enabled = {a["id"] for a in initial_enabled} queue = deque([(tuple(), frozenset(start_caps), initial_used, 0.0)]) visited: set[frozenset[str]] = {frozenset()} summaries = [] truncated = False while queue: selected, frozen_caps, used, risk_used = queue.popleft() caps = set(frozen_caps) summaries.append(summarize(selected, caps, used, risk_used, actions, by_id, base_enabled, spec, constraints)) if len(selected) >= constraints["max_plan_steps"]: continue selected_set = set(selected) for action in actions: if action["id"] in selected_set: continue if assess(action, caps, selected_set, used, risk_used, spec["resources"], spec["observations"], constraints, by_id): continue new_ids = selected + (action["id"],) key = frozenset(new_ids) if key in visited: continue if len(visited) >= MAX_SEARCH_STATES: truncated = True break visited.add(key) new_caps = caps | set(action["provides"]) new_used = {k: used[k] + action["cost"][k] for k in used} queue.append((new_ids, frozenset(new_caps), new_used, risk_used + action["risk"])) if truncated: break summaries.sort(key=lambda x: (x["score"], x["target_progress"], x["direct_value"], x["new_option_value"], -x["risk_total"]), reverse=True) best = summaries[0] selected_actions = [by_id[x] for x in best["actions"]] contracts = [{ "action_id": a["id"], "title": a["title"], "status": "proposed_not_executed", "preconditions": a["preconditions"], "required_observations": a["requires_observations"], "success_metrics": a["success_metrics"], "verification": a["verification"], "rollback": a["rollback"], "reversible": a["reversible"], "approval_required": a["approval_required"] or not a["reversible"], "evidence_grade": a["evidence_grade"], "predicted_effects_not_measured": a["effects"], } for a in selected_actions] blocked = [] for a in actions: reasons = assess(a, start_caps, set(), initial_used, 0.0, spec["resources"], spec["observations"], constraints, by_id) if reasons: blocked.append({"id": a["id"], "title": a["title"], "reasons": reasons}) pareto = [] for candidate in summaries: if any(dominates(candidate, other) for other in summaries if other is not candidate): continue pareto.append(candidate) pareto.sort(key=lambda x: x["score"], reverse=True) timestamp = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) result = { "schema": "darkoza.existence-plan.v1", "engine": VERSION, "generated_at_utc": timestamp.isoformat().replace("+00:00", "Z"), "title": original.get("title", "Existence possibility plan"), "input_digest_sha256": digest(problem), "status": ("target_reached" if best["target_progress"] == 1.0 and spec["targets"] else "feasible_plan" if best["actions"] else "no_safe_feasible_action"), "decision": "PLAN_ONLY_NO_EXECUTION", "execution_performed": False, "search": {"states_explored": len(visited), "truncated_at_limit": truncated, "max_plan_steps": constraints["max_plan_steps"]}, "scoring_policy": { "mission_weights": spec["weights"], "direct_value": "sum(weighted_effects * declared_confidence)", "option_value": "new immediate next-step proposals unlocked by the plan, weighted by confidence and risk", "option_bonus": "2 * ln(1 + newly_unlocked_option_value)", "risk_penalty": "4 * sum(action risk)", "uncertainty_penalty": "1.5 * sum(weighted_effect * (1 - declared confidence))", "resource_penalty": "1.5 * mean fraction of declared budgets consumed", "target_bonus": "4 * fraction of target capabilities reached", "important": "All effects, confidence, risk and costs are input estimates, not independently measured facts.", }, "recommended_plan": best, "experiment_contracts": contracts, "pareto_alternatives": pareto[:8], "actions_blocked_from_initial_state": blocked, "interpretation": [ "Newly unlocked options are immediate next steps only; they are not selected or executed.", "Deeper opportunity reach, unlisted alternatives and external constraints are not inferred.", "A feasible plan is not an execution authorization or evidence that predicted effects occurred.", "Human approval, identity checks, and a separate executor remain external requirements.", ], } result["decision_digest_sha256"] = digest(result) return result def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("problem", type=Path, help="JSON problem definition") parser.add_argument("--output", type=Path, help="write result JSON to this path") parser.add_argument("--now", help="fixed ISO-8601 timestamp for reproducible output") args = parser.parse_args() try: problem = json.loads(args.problem.read_text(encoding="utf-8")) fixed_now = datetime.fromisoformat(args.now.replace("Z", "+00:00")) if args.now else None if fixed_now and fixed_now.tzinfo is None: raise ValueError("--now must include a timezone") result = compile_plan(problem, now=fixed_now) rendered = json.dumps(result, indent=2, ensure_ascii=False, allow_nan=False) + "\n" if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(rendered, encoding="utf-8") else: sys.stdout.write(rendered) except (OSError, json.JSONDecodeError, ValueError, TypeError) as exc: print(f"existence-possibility-compiler: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())