#!/usr/bin/env python3 """Darkoza EPC v0.2 — scenario-robust possibility planning. Deterministic, bounded, CPU-only planner. It evaluates a plan across a declared set of adverse/favourable scenarios, prices negative externalities, uses resource shadow prices, and credits only next-step options that stay feasible in every declared scenario. It never executes actions. """ 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.2" SCHEMA = "darkoza.existence-problem.v2" DIMENSIONS = ("discover", "create", "preserve", "evolve") CASES = ("low", "base", "high") EVIDENCE_STATES = {"verified", "unverified", "hypothesis"} EVIDENCE_GRADES = {"measured", "tested", "pilot", "hypothesis", "unknown"} MAX_ACTIONS = 16 MAX_SCENARIOS = 12 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 strings(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 range3(value: Any, name: str, *, minimum: float = 0, maximum: float = 1e12) -> dict[str, float]: if not isinstance(value, dict) or set(value) != set(CASES): raise ValueError(f"{name} must contain exactly low/base/high") result = {k: number(value[k], f"{name}.{k}", minimum, maximum) for k in CASES} if not result["low"] <= result["base"] <= result["high"]: raise ValueError(f"{name} must be monotonic: low <= base <= high") return result def range_by_dimension(value: Any, name: str, *, maximum: float = 10) -> dict: if not isinstance(value, dict) or set(value) - set(DIMENSIONS): raise ValueError(f"{name} may contain only {DIMENSIONS}") return {d: range3(value.get(d, {"low": 0, "base": 0, "high": 0}), f"{name}.{d}", maximum=maximum) for d in DIMENSIONS} def impact(ranged: dict, weights: dict, case: str) -> float: return sum(weights[d] * ranged[d][case] for d in DIMENSIONS) def validate(problem: dict) -> tuple[dict, list[dict], list[dict], dict]: if not isinstance(problem, dict) or problem.get("schema") != SCHEMA: raise ValueError(f"schema must equal {SCHEMA}") budget_raw = problem.get("resource_budget") if not isinstance(budget_raw, dict) or not budget_raw: raise ValueError("resource_budget must be a non-empty object") budgets = {} for key, item in budget_raw.items(): if not isinstance(key, str) or not key.strip() or not isinstance(item, dict): raise ValueError("each resource budget must be a named object") budgets[key] = { "limit": number(item.get("limit"), f"resource_budget.{key}.limit", 0, 1e12), "shadow_price": number(item.get("shadow_price", 0), f"resource_budget.{key}.shadow_price", 0, 1e9), } weights_raw = problem.get("mission_weights", {}) if not isinstance(weights_raw, dict) or set(weights_raw) - set(DIMENSIONS): raise ValueError(f"mission_weights may contain only {DIMENSIONS}") weights = {d: number(weights_raw.get(d, 1), f"mission_weights.{d}", 0, 1e6) for d in DIMENSIONS} total_weight = sum(weights.values()) if total_weight <= 0: raise ValueError("at least one mission weight must be positive") weights = {k: v / total_weight for k, v in weights.items()} current = strings(problem.get("current_capabilities", []), "current_capabilities") targets = strings(problem.get("target_capabilities", []), "target_capabilities") observations_raw = problem.get("observations", []) if not isinstance(observations_raw, list): raise ValueError("observations must be a list") observations = {} for obs in observations_raw: 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 observations: 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 observation requires a source") observations[oid] = obs scenarios_raw = problem.get("scenarios") if not isinstance(scenarios_raw, list) or not 2 <= len(scenarios_raw) <= MAX_SCENARIOS: raise ValueError(f"scenarios must contain 2..{MAX_SCENARIOS} explicit scenarios") scenarios, scenario_ids = [], set() for raw in scenarios_raw: if not isinstance(raw, dict): raise ValueError("each scenario must be an object") sid = raw.get("id") if not isinstance(sid, str) or not sid.strip() or sid in scenario_ids: raise ValueError("scenario ids must be non-empty and unique") scenario_ids.add(sid) scenario = {"id": sid} for field in ("effect_case", "externality_case", "cost_case", "risk_case", "confidence_case"): val = raw.get(field, "base") if val not in CASES: raise ValueError(f"{sid}.{field} must be one of {CASES}") scenario[field] = val scenarios.append(scenario) actions_raw = problem.get("actions") if not isinstance(actions_raw, list) or not actions_raw: raise ValueError("actions must be a non-empty list") if len(actions_raw) > MAX_ACTIONS: raise ValueError(f"at most {MAX_ACTIONS} actions are supported") actions, action_ids = [], set() for raw in actions_raw: 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 action_ids: raise ValueError("action ids must be non-empty and unique") action_ids.add(aid) for field in ("requires", "provides", "requires_observations", "conflicts_with", "preconditions", "verification", "success_metrics"): a[field] = strings(a.get(field, []), f"{aid}.{field}", required=(field == "provides")) if not isinstance(a.get("title"), str) or not a["title"].strip(): raise ValueError(f"{aid}: title is required") if not isinstance(a.get("rollback"), str) or not a["rollback"].strip(): raise ValueError(f"{aid}: rollback 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["effects"] = range_by_dimension(a.get("effects"), f"{aid}.effects") a["externalities"] = range_by_dimension(a.get("externalities", {}), f"{aid}.externalities") a["risk"] = range3(a.get("risk"), f"{aid}.risk", maximum=1) a["confidence"] = range3(a.get("confidence"), f"{aid}.confidence", maximum=1) if a.get("evidence_grade") not in EVIDENCE_GRADES: raise ValueError(f"{aid}: invalid evidence_grade") cost_raw = a.get("cost", {}) if not isinstance(cost_raw, dict) or set(cost_raw) - set(budgets): raise ValueError(f"{aid}: cost may use only declared resources") a["cost"] = { resource: range3(cost_raw.get(resource, {"low": 0, "base": 0, "high": 0}), f"{aid}.cost.{resource}") for resource in budgets } unknown_obs = set(a["requires_observations"]) - set(observations) if unknown_obs: raise ValueError(f"{aid}: unknown observation ids {sorted(unknown_obs)}") actions.append(a) constraints_raw = problem.get("constraints", {}) if not isinstance(constraints_raw, dict): raise ValueError("constraints must be an object") constraints = { "max_risk_per_action": number(constraints_raw.get("max_risk_per_action", .65), "max_risk_per_action", 0, 1), "max_total_risk": number(constraints_raw.get("max_total_risk", 1), "max_total_risk", 0, 16), "max_weighted_externality": number(constraints_raw.get("max_weighted_externality", 100), "max_weighted_externality", 0, 1e9), "max_plan_steps": int(number(constraints_raw.get("max_plan_steps", 3), "max_plan_steps", 1, 8)), "regret_weight": number(constraints_raw.get("regret_weight", .25), "regret_weight", 0, 1), "require_reversible": constraints_raw.get("require_reversible", True), "allow_approval_required": constraints_raw.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") spec = {"budgets": budgets, "weights": weights, "current": current, "targets": targets, "observations": observations} return spec, sorted(actions, key=lambda x: x["id"]), scenarios, constraints def action_cost(action: dict, scenario: dict, resource: str) -> float: return action["cost"][resource][scenario["cost_case"]] def action_risk(action: dict, scenario: dict) -> float: return action["risk"][scenario["risk_case"]] def action_harm(action: dict, scenario: dict, weights: dict) -> float: return impact(action["externalities"], weights, scenario["externality_case"]) def state_usage(selected: tuple[str, ...], by_id: dict, scenarios: list[dict], budgets: dict) -> tuple[dict, dict, dict]: usage, risks, harms = {}, {}, {} for scenario in scenarios: sid = scenario["id"] usage[sid] = { r: sum(action_cost(by_id[a], scenario, r) for a in selected) for r in budgets } risks[sid] = sum(action_risk(by_id[a], scenario) for a in selected) # Weighted harm is computed later with mission weights. harms[sid] = 0.0 return usage, risks, harms def selected_harm(selected: tuple[str, ...], by_id: dict, scenario: dict, weights: dict) -> float: return sum(action_harm(by_id[a], scenario, weights) for a in selected) def assess(action: dict, capabilities: set[str], selected: set[str], usage: dict, risks: dict, harms: dict, spec: dict, scenarios: list[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 spec["observations"][oid]["status"] != "verified"] if unverified: reasons.append("unverified_observations:" + ",".join(sorted(unverified))) 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 in selected: if action["id"] in by_id[other]["conflicts_with"]: reasons.append("selected_action_conflicts_with_this") break for scenario in scenarios: sid = scenario["id"] new_risk = risks[sid] + action_risk(action, scenario) if action_risk(action, scenario) > constraints["max_risk_per_action"]: reasons.append(f"per_action_risk_limit:{sid}") if new_risk > constraints["max_total_risk"] + 1e-12: reasons.append(f"cumulative_risk_limit:{sid}") new_harm = harms[sid] + action_harm(action, scenario, spec["weights"]) if new_harm > constraints["max_weighted_externality"] + 1e-12: reasons.append(f"externality_limit:{sid}") for resource, budget in spec["budgets"].items(): new_used = usage[sid][resource] + action_cost(action, scenario, resource) if new_used > budget["limit"] + 1e-12: reasons.append(f"resource_limit:{resource}:{sid}") return reasons def enabled_actions(actions: list[dict], capabilities: set[str], selected: set[str], usage: dict, risks: dict, harms: dict, spec: dict, scenarios: list[dict], constraints: dict, by_id: dict) -> list[dict]: return [a for a in actions if a["id"] not in selected and not assess( a, capabilities, selected, usage, risks, harms, spec, scenarios, constraints, by_id )] def option_potential(action: dict, scenario: dict, weights: dict) -> float: benefit = impact(action["effects"], weights, scenario["effect_case"]) harm = action_harm(action, scenario, weights) confidence = action["confidence"][scenario["confidence_case"]] risk = action_risk(action, scenario) return max(0.0, benefit - harm) * confidence / (1 + 2 * risk) def summarize(selected: tuple[str, ...], capabilities: set[str], usage: dict, risks: dict, selected_harms: dict, actions: list[dict], by_id: dict, base_enabled_by_scenario: dict, spec: dict, scenarios: list[dict], constraints: dict) -> dict: scenario_rows = {} enabled_by_scenario = {} new_by_scenario = {} for scenario in scenarios: sid = scenario["id"] enabled = enabled_actions(actions, capabilities, set(selected), usage, risks, selected_harms, spec, [scenario], constraints, by_id) enabled_ids = {a["id"] for a in enabled} enabled_by_scenario[sid] = enabled_ids new_by_scenario[sid] = enabled_ids - base_enabled_by_scenario[sid] stable_new = set.intersection(*(new_by_scenario[s["id"]] for s in scenarios)) if scenarios else set() union_new = set.union(*(new_by_scenario[s["id"]] for s in scenarios)) if scenarios else set() fragile_new = union_new - stable_new persistent_values = {} for aid in sorted(stable_new): per_scenario = [option_potential(by_id[aid], s, spec["weights"]) for s in scenarios] persistent_values[aid] = min(per_scenario) persistent_option_value = sum(persistent_values.values()) option_bonus = 2.0 * math.log1p(persistent_option_value) selected_actions = [by_id[x] for x in selected] target_hits = sorted(set(spec["targets"]) & capabilities) target_progress = len(target_hits) / len(spec["targets"]) if spec["targets"] else 0.0 target_bonus = 4.0 * target_progress for scenario in scenarios: sid = scenario["id"] direct = sum( impact(a["effects"], spec["weights"], scenario["effect_case"]) * a["confidence"][scenario["confidence_case"]] for a in selected_actions ) harm = selected_harms[sid] risk = risks[sid] shadow_cost = sum( usage[sid][r] * spec["budgets"][r]["shadow_price"] for r in spec["budgets"] ) fractions = [ usage[sid][r] / spec["budgets"][r]["limit"] for r in spec["budgets"] if spec["budgets"][r]["limit"] > 0 ] burden = sum(fractions) / len(fractions) if fractions else 0.0 score = (direct + option_bonus + target_bonus - harm - 4.0 * risk - shadow_cost - 1.5 * burden) scenario_rows[sid] = { "direct_value": round(direct, 6), "weighted_externality": round(harm, 6), "risk_total": round(risk, 6), "resource_shadow_cost": round(shadow_cost, 6), "resource_budget_burden": round(burden, 6), "target_bonus": round(target_bonus, 6), "persistent_option_bonus": round(option_bonus, 6), "score": round(score, 6), "resource_use": usage[sid], } return { "actions": list(selected), "capabilities_after": sorted(capabilities), "target_capabilities_reached": target_hits, "target_progress": round(target_progress, 6), "persistent_new_options": [ {"id": aid, "title": by_id[aid]["title"], "worst_scenario_option_value": round(value, 6)} for aid, value in persistent_values.items() ], "fragile_new_options": [ {"id": aid, "title": by_id[aid]["title"], "available_in_scenarios": sorted( sid for sid in new_by_scenario if aid in new_by_scenario[sid] )} for aid in sorted(fragile_new) ], "persistent_option_value": round(persistent_option_value, 6), "scenario_scores": scenario_rows, "scenario_mean_score_unweighted": round( sum(row["score"] for row in scenario_rows.values()) / len(scenario_rows), 6 ), "worst_case_score": round(min(row["score"] for row in scenario_rows.values()), 6), "best_case_score": round(max(row["score"] for row in scenario_rows.values()), 6), "worst_case_externality": round(max(row["weighted_externality"] for row in scenario_rows.values()), 6), "worst_case_risk": round(max(row["risk_total"] for row in scenario_rows.values()), 6), "resource_use_by_scenario": { sid: row["resource_use"] for sid, row in scenario_rows.items() }, } def dominates(a: dict, b: dict) -> bool: av = (a["robust_score"], a["worst_case_score"], a["persistent_option_value"], a["target_progress"], -a["max_regret"], -a["worst_case_externality"]) bv = (b["robust_score"], b["worst_case_score"], b["persistent_option_value"], b["target_progress"], -b["max_regret"], -b["worst_case_externality"]) return all(x >= y for x, y in zip(av, bv)) and any(x > y for x, y in zip(av, bv)) def compile_plan(problem: dict, *, now: datetime | None = None) -> dict: spec, actions, scenarios, constraints = validate(problem) by_id = {a["id"]: a for a in actions} start_caps = set(spec["current"]) initial_selected: tuple[str, ...] = tuple() initial_usage, initial_risks, _ = state_usage(initial_selected, by_id, scenarios, spec["budgets"]) initial_harms = {s["id"]: 0.0 for s in scenarios} base_enabled_by_scenario = {} for s in scenarios: base_enabled_by_scenario[s["id"]] = { a["id"] for a in enabled_actions(actions, start_caps, set(), initial_usage, initial_risks, initial_harms, spec, [s], constraints, by_id) } queue = deque([(tuple(), frozenset(start_caps))]) visited: set[frozenset[str]] = {frozenset()} summaries = [] truncated = False while queue: selected, frozen_caps = queue.popleft() caps = set(frozen_caps) usage, risks, _ = state_usage(selected, by_id, scenarios, spec["budgets"]) harms = {s["id"]: selected_harm(selected, by_id, s, spec["weights"]) for s in scenarios} summaries.append(summarize(selected, caps, usage, risks, harms, actions, by_id, base_enabled_by_scenario, spec, scenarios, 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, usage, risks, harms, spec, scenarios, 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"]) queue.append((new_ids, frozenset(new_caps))) if truncated: break best_by_scenario = { s["id"]: max(row["scenario_scores"][s["id"]]["score"] for row in summaries) for s in scenarios } for row in summaries: regrets = { s["id"]: round(best_by_scenario[s["id"]] - row["scenario_scores"][s["id"]]["score"], 6) for s in scenarios } worst_regret_scenario = max(regrets, key=regrets.get) row["scenario_regrets"] = regrets row["max_regret"] = regrets[worst_regret_scenario] row["max_regret_scenario"] = worst_regret_scenario row["robust_score"] = round( row["worst_case_score"] - constraints["regret_weight"] * row["max_regret"], 6 ) row["scenario_top_count"] = sum( 1 for s in scenarios if row["scenario_scores"][s["id"]]["score"] == best_by_scenario[s["id"]] ) row["scenario_rank_stability"] = round(row["scenario_top_count"] / len(scenarios), 6) summaries.sort(key=lambda x: ( x["robust_score"], x["worst_case_score"], x["scenario_mean_score_unweighted"], x["persistent_option_value"], -x["max_regret"] ), 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"], "benefit_ranges_not_measured": a["effects"], "externality_ranges_not_measured": a["externalities"], } for a in selected_actions] blocked = [] for a in actions: reasons = assess(a, start_caps, set(), initial_usage, initial_risks, initial_harms, spec, scenarios, constraints, by_id) if reasons: blocked.append({"id": a["id"], "title": a["title"], "reasons": reasons}) pareto = [candidate for candidate in summaries if not any(dominates(other, candidate) for other in summaries if other is not candidate)] pareto.sort(key=lambda x: x["robust_score"], reverse=True) scenario_winners = { s["id"]: max(summaries, key=lambda x: x["scenario_scores"][s["id"]]["score"])["actions"] for s in scenarios } timestamp = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) result = { "schema": "darkoza.existence-plan.v2", "engine": VERSION, "generated_at_utc": timestamp.isoformat().replace("+00:00", "Z"), "title": problem.get("title", "Existence possibility robust 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, "scenario_method": { "type": "unweighted_explicit_stress_scenarios", "probabilities_assumed": False, "scenario_count": len(scenarios), "robust_selection": "worst_case_score - regret_weight * maximum_regret", "regret_weight": constraints["regret_weight"], "option_credit": "only actions newly feasible in every declared scenario; value uses the minimum scenario potential", "mean_warning": "scenario_mean_score_unweighted is descriptive only, not an expected value or probability-weighted forecast", }, "search": {"states_explored": len(visited), "truncated_at_limit": truncated, "max_plan_steps": constraints["max_plan_steps"]}, "scoring_policy": { "direct_value": "sum(weighted benefit in scenario * scenario confidence)", "externality": "subtract weighted negative effects in each scenario", "resource_cost": "sum(resource consumption * declared resource shadow price)", "risk": "subtract 4 * scenario risk", "resource_burden": "subtract 1.5 * mean fraction of declared budgets consumed", "target_bonus": "4 * fraction of target capabilities reached", "option_bonus": "2 * ln(1 + persistent_option_value)", "robust_score": "worst_case_score - regret_weight * max_regret", "important": "Ranges, scenario design, confidence, risk, externalities, costs and shadow prices are author-supplied assumptions, not independently measured facts.", }, "recommended_plan": best, "scenario_winners": scenario_winners, "experiment_contracts": contracts, "pareto_alternatives": pareto[:8], "actions_blocked_from_initial_state": blocked, "interpretation": [ "No scenario probabilities are inferred; scenarios are stress cases, not forecasts.", "Persistent options are only immediate next steps enabled in all declared scenarios.", "Fragile options are shown but receive no persistent-option credit.", "A plan is not an execution authorization or proof that predicted outcomes occurred.", "Human approval, authenticated identity and a separate sandboxed 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) parser.add_argument("--output", type=Path) parser.add_argument("--now", help="fixed ISO-8601 timestamp for reproducibility") args = parser.parse_args() try: problem = json.loads(args.problem.read_text(encoding="utf-8")) fixed = datetime.fromisoformat(args.now.replace("Z", "+00:00")) if args.now else None if fixed and fixed.tzinfo is None: raise ValueError("--now must include timezone") result = compile_plan(problem, now=fixed) 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-v02: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())