mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-19 18:14:54 +00:00
edd8c47254
Preserve healthy registry operations beside malformed neighbours, exact continuations and frozen resources, Project addressing and child wait facts. Retain one refused evolution transaction and refresh its admitted Git basis. Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>
1596 lines
72 KiB
Python
1596 lines
72 KiB
Python
"""Evolution campaign state and transaction lifecycle.
|
|
|
|
Owns the campaign file (``state/evolution_campaign.json``), the campaign/
|
|
transaction lifecycle transitions, and the owner-facing cycle reporting.
|
|
``supervisor.queue`` imports this module top-level; anything here that needs
|
|
queue state (``DRIVE_ROOT``, locks) must import the queue lazily at call time
|
|
to avoid a module-load cycle.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import uuid
|
|
from typing import Any, Dict, Optional
|
|
|
|
from ouroboros.consciousness_authority import consciousness_origin_metadata
|
|
from ouroboros.cost_projection import honest_cost_pair_amount
|
|
from ouroboros.evolution_fingerprint import canonical_objective_fingerprint
|
|
from ouroboros.outcomes import normalize_outcome_axes
|
|
from ouroboros.utils import atomic_write_json, read_json_dict, utc_now_iso
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
EVOLUTION_CAMPAIGN_FILE = pathlib.Path("state") / "evolution_campaign.json"
|
|
EVOLUTION_CAMPAIGN_CAS_TIMEOUT_SEC = 0.05
|
|
|
|
|
|
def current_evolution_boot_generation() -> str:
|
|
"""Return the existing custody generation used by boot reconciliation."""
|
|
try:
|
|
from ouroboros.process_custody import current_custody_session_id
|
|
|
|
return str(current_custody_session_id() or "")
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _evolution_campaign_path() -> pathlib.Path:
|
|
from supervisor import queue
|
|
|
|
return pathlib.Path(queue.DRIVE_ROOT) / EVOLUTION_CAMPAIGN_FILE
|
|
|
|
|
|
def _read_evolution_campaign() -> Dict[str, Any]:
|
|
data = read_json_dict(_evolution_campaign_path()) or {}
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
# #1307: a received Stop outlives a failed state write. The live latch keeps it in this
|
|
# process; the campaign's ``stop_intent`` keeps it durably, beside (never instead of) the
|
|
# state flag and whether or not cancellation finished. Only an owner-sourced successful
|
|
# ``start_evolution_campaign`` clears them.
|
|
_STOP_LATCH = {"stopped": False}
|
|
_OWNER_START_SOURCES = frozenset({"owner", "owner_chat"})
|
|
|
|
|
|
def record_evolution_stop_intent(source: str, reason: str = "") -> bool:
|
|
"""Latch the Stop, then record it in the EXISTING campaign record (no invented
|
|
terminal status). False: the durable intent could not be written (the caller
|
|
discloses it). An absent campaign gets only the intent, never a fake status."""
|
|
from supervisor import state
|
|
|
|
_STOP_LATCH["stopped"] = True
|
|
lock_fd = state.acquire_file_lock(state.STATE_LOCK_PATH, timeout_sec=1.0)
|
|
if lock_fd is None:
|
|
return False
|
|
try:
|
|
campaign = _read_evolution_campaign()
|
|
campaign["stop_intent"] = {"at": utc_now_iso(), "source": str(source or "owner"), "reason": str(reason or "")}
|
|
return _write_evolution_campaign(campaign, _state_lock_held=True)
|
|
except OSError:
|
|
return False
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
|
|
def evolution_stop_reason(campaign: Optional[Dict[str, Any]] = None) -> str:
|
|
"""Why no new evolution may be admitted regardless of state.json: a received
|
|
Stop (latch or durable intent) or an unconsumed Panic flag. "" when none."""
|
|
from supervisor import queue
|
|
from supervisor.state_initialization import confirm_absent
|
|
|
|
if _STOP_LATCH["stopped"]:
|
|
return "stop_latched"
|
|
if isinstance((campaign if campaign is not None else _read_evolution_campaign()).get("stop_intent"), dict):
|
|
return "stop_intent"
|
|
path = pathlib.Path(queue.DRIVE_ROOT) / "state" / "panic_stop.flag"
|
|
try:
|
|
try:
|
|
os.lstat(path)
|
|
except FileNotFoundError:
|
|
confirm_absent(path)
|
|
return ""
|
|
return "panic_flag"
|
|
except OSError:
|
|
return "panic_flag_unknown"
|
|
|
|
|
|
def disable_evolution_projection() -> None:
|
|
"""Clear the scheduler projection without changing campaign history."""
|
|
from supervisor import state
|
|
|
|
try:
|
|
state.update_state(lambda live: live.update({
|
|
"evolution_mode_enabled": False,
|
|
"post_task_autostop": False,
|
|
}), confirm=("evolution_mode_enabled", "post_task_autostop"))
|
|
except state.StateUnavailable:
|
|
log.warning("evolution projection disable not persisted: state unavailable", exc_info=True)
|
|
|
|
|
|
def disable_evolution_authority(
|
|
action: str, *, campaign_id: str = "", task_id: str = "",
|
|
) -> None:
|
|
"""Clear an invalid scheduler projection and record the typed refusal."""
|
|
from supervisor import queue, state
|
|
|
|
disable_evolution_projection()
|
|
event = {
|
|
"ts": utc_now_iso(),
|
|
"type": "evolution_authority_missing",
|
|
"action": str(action or ""),
|
|
"campaign_id": str(campaign_id or ""),
|
|
}
|
|
if task_id:
|
|
event["task_id"] = str(task_id)
|
|
state.append_jsonl(pathlib.Path(queue.DRIVE_ROOT) / "logs" / "events.jsonl", event)
|
|
|
|
|
|
def deliver_pending_owner_report(notify: Any = None) -> None:
|
|
"""Deliver and clear a worker-staged owner report from the server process."""
|
|
try:
|
|
campaign = _read_evolution_campaign()
|
|
report = campaign.get("pending_owner_report")
|
|
if not isinstance(report, dict):
|
|
return
|
|
(notify or notify_owner_cycle_outcome)(campaign, report)
|
|
clear_pending_owner_report(report)
|
|
except Exception:
|
|
log.debug("failed to deliver pending owner report", exc_info=True)
|
|
|
|
|
|
def _deliver_pending_owner_report() -> None:
|
|
"""Queue-compatibility wrapper for the server-process owner report."""
|
|
from supervisor import queue as q
|
|
|
|
q.deliver_pending_owner_report(q.notify_owner_cycle_outcome)
|
|
|
|
|
|
def enqueue_evolution_task_if_needed() -> None:
|
|
"""Queue evolution only when idle, enabled, within budget, and not failure-paused."""
|
|
from supervisor import queue as q
|
|
|
|
q._deliver_pending_owner_report()
|
|
if q.PENDING or q.RUNNING:
|
|
return
|
|
from supervisor.state import StateUnavailable, control_is, update_state
|
|
|
|
st = q.load_state()
|
|
# Autonomy needs KNOWN current controls (#1307): an unknown flag is not "enabled".
|
|
# The owner chat is only the notice address, so its display value serves.
|
|
if not (control_is(st, "evolution_mode_enabled", True) and control_is(st, "evolution_owner_stopped", False)):
|
|
return
|
|
owner_chat_id = st.get("owner_chat_id")
|
|
if not owner_chat_id:
|
|
return
|
|
campaign = q._read_evolution_campaign()
|
|
if evolution_stop_reason(campaign):
|
|
return
|
|
has_authority = all(str(campaign.get(key) or "").strip() for key in ("id", "source"))
|
|
if campaign.get("status") != "active" or not has_authority:
|
|
q.disable_evolution_authority("bare_flag_disabled", campaign_id=str(campaign.get("id") or ""))
|
|
q.send_with_budget(
|
|
int(owner_chat_id),
|
|
"🧬 Evolution stayed off: the enable flag had no active campaign authority. Use /evolve start to begin a fresh campaign.",
|
|
role="system", system_type="evolution_notice")
|
|
return
|
|
active_tx = campaign.get("active_transaction") if isinstance(campaign.get("active_transaction"), dict) else {}
|
|
if active_tx and (
|
|
str(active_tx.get("commit_sha") or "").strip()
|
|
or str(active_tx.get("dispatch_status") or "") == "reaping"
|
|
):
|
|
return
|
|
|
|
# Defensive net: light mode must never run evolution even if the flag was
|
|
# left enabled (e.g. carried across a restart into light mode). Disable and
|
|
# pause once; entry points already refuse new starts up front.
|
|
block = q.evolution_block_reason()
|
|
if block:
|
|
q.pause_evolution_campaign("blocked in light runtime mode")
|
|
q.disable_evolution_projection()
|
|
q.send_with_budget(int(owner_chat_id), block, role="system", system_type="evolution_notice")
|
|
return
|
|
|
|
consecutive_failures = int(st.get("evolution_consecutive_failures") or 0)
|
|
if consecutive_failures >= 3:
|
|
q.pause_evolution_campaign("paused after consecutive failures")
|
|
q.disable_evolution_projection()
|
|
q.send_with_budget(
|
|
int(owner_chat_id),
|
|
f"🧬⚠️ Evolution paused: {consecutive_failures} consecutive failures. "
|
|
f"Use /evolve start to resume after investigating the issue.",
|
|
role="system", system_type="evolution_notice")
|
|
return
|
|
|
|
# Objective repeats count non-consecutive no-ops by the transaction's canonical
|
|
# fingerprint, clearing only on absorption. The failure streak cannot catch
|
|
# these loops because every non-failing cycle resets it (events.py).
|
|
objective_repeat_counts = campaign.get("objective_repeat_counts") or {}
|
|
active_objective_fp = canonical_objective_fingerprint(
|
|
str(campaign.get("objective") or ""),
|
|
)
|
|
objective_repeats = (
|
|
int(objective_repeat_counts.get(active_objective_fp, 0))
|
|
if active_objective_fp else 0
|
|
)
|
|
if objective_repeats >= q.OBJECTIVE_REPEAT_CAP:
|
|
q.pause_evolution_campaign(
|
|
"paused: objective re-proposed without ever absorbing",
|
|
)
|
|
q.disable_evolution_projection()
|
|
q.send_with_budget(
|
|
int(owner_chat_id),
|
|
f"🧬⚠️ Evolution paused: the current objective ran {objective_repeats} reviewed "
|
|
f"cycles WITHOUT ever being absorbed — it keeps getting re-proposed and never lands "
|
|
f"(a self-maintenance loop, not progress). A plain resume won't help; use "
|
|
f"/evolve start with a DIFFERENT objective.",
|
|
role="system", system_type="evolution_notice")
|
|
return
|
|
|
|
try: # on the supervisor loop: ride the snapshot, decide the reserve refusal on the exact read
|
|
remaining = q.budget_remaining(st, strict=True, allow_stale=True, refuse_below=q.EVOLUTION_BUDGET_RESERVE)
|
|
except Exception:
|
|
log.error("Evolution scheduling deferred: cost accounting unavailable", exc_info=True)
|
|
q.append_jsonl(q.DRIVE_ROOT / "logs" / "events.jsonl", {
|
|
"ts": utc_now_iso(), "type": "evolution_accounting_unavailable",
|
|
"action": "dispatch_deferred", "owner_visible": True,
|
|
})
|
|
return
|
|
if remaining < q.EVOLUTION_BUDGET_RESERVE:
|
|
q.pause_evolution_campaign("budget reserve reached")
|
|
q.disable_evolution_projection()
|
|
q.send_with_budget(
|
|
int(owner_chat_id),
|
|
f"💸 Evolution stopped: ${remaining:.2f} remaining "
|
|
f"(reserve ${q.EVOLUTION_BUDGET_RESERVE:.0f} for conversations).",
|
|
role="system", system_type="evolution_notice")
|
|
return
|
|
cycle = int(st.get("evolution_cycle") or 0) + 1
|
|
tid = uuid.uuid4().hex[:8]
|
|
transaction = q.begin_evolution_transaction(tid, cycle=cycle, campaign=campaign)
|
|
if not transaction:
|
|
q.disable_evolution_authority("transaction_attach_failed", campaign_id=str(campaign.get("id") or ""), task_id=tid)
|
|
q.send_with_budget(
|
|
int(owner_chat_id),
|
|
"🧬 Evolution stayed off: the campaign changed before its next task could be attached. Start it again when ready.",
|
|
role="system", system_type="evolution_notice")
|
|
return
|
|
tid = transaction["task_id"] # only a positively refused attempt retains its identity
|
|
task = {
|
|
"id": tid, "type": "evolution",
|
|
"chat_id": int(owner_chat_id),
|
|
"text": q.build_evolution_task_text(cycle),
|
|
"metadata": {"evolution_transaction": transaction, **consciousness_origin_metadata(campaign)},
|
|
}
|
|
q.attach_task_contract(task)
|
|
def prepare_admission(row: Dict[str, Any]) -> bool:
|
|
refreshed = q.begin_evolution_transaction(tid, cycle=cycle, campaign=campaign, transaction=transaction)
|
|
if refreshed:
|
|
row["metadata"]["evolution_transaction"] = refreshed
|
|
return bool(refreshed)
|
|
|
|
admitted = q.enqueue_with_admission_receipt(task, prepare_admission=prepare_admission)
|
|
if isinstance(admitted, dict) and admitted.get("_admission_blocked"):
|
|
# Preserve positive no-handoff evidence, never infer it from an idle queue.
|
|
if admitted.get("_admission_never_admitted") is True:
|
|
refusal = {"reason": admitted["_admission_blocked"], "source": campaign["source"]}
|
|
update_evolution_transaction(tid, admission_refused=refusal)
|
|
reason = str(admitted.get("_admission_blocked") or "admission_fence")
|
|
detail = str(admitted.get("_admission_detail") or admitted.get("_worker_pool_disabled_reason") or "")
|
|
if not reason.startswith("consciousness_"):
|
|
if transaction.get("admission_refused", {}).get("reason") != reason:
|
|
log.warning("evolution cycle %s was not admitted (%s); retrying on the next pass", tid, reason)
|
|
return
|
|
q.pause_evolution_campaign(f"admission_refused:{reason}")
|
|
q.disable_evolution_projection()
|
|
q.send_with_budget(
|
|
int(owner_chat_id),
|
|
f"🧬 Evolution paused: its next cycle was not admitted ({reason}{': ' + detail if detail else ''}). "
|
|
"/evolve start resumes it.",
|
|
role="system", system_type="evolution_notice")
|
|
return
|
|
|
|
def _record_cycle(live: Dict[str, Any]) -> None:
|
|
live["evolution_cycle"] = cycle
|
|
live["last_evolution_task_at"] = utc_now_iso()
|
|
|
|
try:
|
|
update_state(_record_cycle)
|
|
except StateUnavailable:
|
|
log.warning("evolution cycle %s admitted; its state counter was not persisted", tid, exc_info=True)
|
|
|
|
|
|
def _write_evolution_campaign(
|
|
data: Dict[str, Any], *, expected_campaign_id: Optional[str] = None,
|
|
expected_active_transaction: Optional[Dict[str, Any]] = None,
|
|
_state_lock_held: bool = False, _lock_timeout_sec: float = 4.0,
|
|
) -> bool:
|
|
path = _evolution_campaign_path()
|
|
from supervisor import state
|
|
|
|
state.assert_test_data_path(path)
|
|
lock_fd = None
|
|
if not _state_lock_held:
|
|
lock_fd = state.acquire_file_lock(
|
|
state.STATE_LOCK_PATH, timeout_sec=float(_lock_timeout_sec),
|
|
)
|
|
if lock_fd is None:
|
|
return False
|
|
try:
|
|
current = read_json_dict(path)
|
|
if current is None and path.is_file(): # present but unreadable is NOT "no campaign"
|
|
log.warning("Refusing evolution campaign write: %s exists but is unreadable", path)
|
|
return False
|
|
current = current or {}
|
|
current_id = str(current.get("id") or "")
|
|
data_id = str(data.get("id") or "")
|
|
expected_id = data_id if expected_campaign_id is None else str(expected_campaign_id or "")
|
|
if current_id and current_id != expected_id:
|
|
log.warning(
|
|
"Refusing stale evolution campaign write: current=%s expected=%s",
|
|
current_id, expected_id,
|
|
)
|
|
return False
|
|
if (
|
|
current_id == data_id
|
|
and current.get("status") not in {"active", "paused"}
|
|
and data.get("status") in {"active", "paused"}
|
|
):
|
|
log.warning("Refusing stale resurrection of terminal evolution campaign %s", data_id)
|
|
return False
|
|
if expected_active_transaction is not None:
|
|
current_tx = current.get("active_transaction", {})
|
|
if current_tx != expected_active_transaction:
|
|
log.warning("Refusing stale evolution transaction write for campaign %s", data_id)
|
|
return False
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
atomic_write_json(path, data, trailing_newline=True)
|
|
return True
|
|
finally:
|
|
if lock_fd is not None:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
|
|
def evolution_block_reason() -> str:
|
|
"""Refusal message when evolution may not run in the current runtime mode.
|
|
|
|
Evolution campaigns are self-modification work, so they require runtime
|
|
mode ``advanced`` or ``pro``. In ``light`` (conversation-only) mode they are
|
|
hard-blocked before any campaign state, queue entry, or expensive round.
|
|
Returns ``""`` when evolution is allowed.
|
|
"""
|
|
from ouroboros.config import get_runtime_mode
|
|
|
|
if get_runtime_mode() == "light":
|
|
return (
|
|
"🧬 Evolution campaigns are self-modification work and require runtime "
|
|
"mode 'advanced' or 'pro'. The runtime is in 'light' mode "
|
|
"(self-modification is disabled), so no campaign was started. Switch "
|
|
"the runtime mode in Settings to evolve."
|
|
)
|
|
return ""
|
|
|
|
|
|
def start_evolution_campaign(objective: str = "", *, source: str = "owner", origin: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Start or resume the active evolution campaign; ``origin`` = the consciousness origin keys of the starting turn, kept for its cycle tasks."""
|
|
from supervisor import state
|
|
|
|
state.assert_test_data_path(state.STATE_PATH)
|
|
lock_fd = state.acquire_file_lock(state.STATE_LOCK_PATH)
|
|
if lock_fd is None:
|
|
return {}
|
|
try:
|
|
campaign = _read_evolution_campaign()
|
|
owner_start = str(source or "") in _OWNER_START_SOURCES
|
|
if not owner_start and evolution_stop_reason(campaign):
|
|
return {} # a received Stop yields only to the owner's own start (#1307)
|
|
prior_campaign_id = str(campaign.get("id") or "")
|
|
now = utc_now_iso()
|
|
objective = str(objective or "").strip()
|
|
if campaign.get("status") not in {"active", "paused"}:
|
|
campaign = {
|
|
"schema_version": 1,
|
|
"id": uuid.uuid4().hex[:8],
|
|
"status": "active",
|
|
"objective": objective or "Autonomously improve Ouroboros by acting on the highest-value backlog or process-memory signal.",
|
|
"source": str(source or ""), **consciousness_origin_metadata(origin),
|
|
"started_at": now,
|
|
"updated_at": now,
|
|
"cycles_done": 0,
|
|
"absorbed_cycles_done": 0,
|
|
"objective_repeat_counts": {}, # BUG3: fp -> non-absorbing-cycle count
|
|
"dropped_objective_fps": [], # BUG3 Layer B: attempted-and-dropped objective fps
|
|
"budget_spent_usd": 0.0,
|
|
"last_task_id": "",
|
|
"progress_notes": "",
|
|
"completed_at": "",
|
|
"completion_reason": "",
|
|
}
|
|
else:
|
|
if objective:
|
|
campaign["objective"] = objective
|
|
if not str(campaign.get("source") or "").strip() and source:
|
|
campaign["source"] = str(source)
|
|
if (str(source or "") in ("owner", "owner_chat") and campaign.get("status") == "paused"
|
|
and consciousness_origin_metadata(campaign)):
|
|
# The owner's explicit start ADOPTS a paused consciousness campaign: its cycles are
|
|
# the owner's work now, no longer bound by the consciousness allowance (BIBLE P0).
|
|
for key in ("initiator", "usage_category", "consciousness_autonomy"):
|
|
campaign.pop(key, None)
|
|
campaign["adopted_by_owner_at"] = now
|
|
campaign.update({k: v for k, v in consciousness_origin_metadata(origin).items() if not campaign.get(k)})
|
|
campaign["status"] = "active"
|
|
campaign.pop("pause_reason", None)
|
|
campaign["updated_at"] = now
|
|
generation = current_evolution_boot_generation()
|
|
if generation:
|
|
campaign["last_boot_reconcile_gen"] = generation
|
|
campaign.pop("stop_intent", None) # cleared ONLY by this successful authorized start
|
|
written = _write_evolution_campaign(
|
|
campaign,
|
|
expected_campaign_id=prior_campaign_id,
|
|
_state_lock_held=True,
|
|
)
|
|
if written and owner_start:
|
|
_STOP_LATCH["stopped"] = False
|
|
return campaign if written else {}
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
|
|
def pause_evolution_campaign(reason: str = "") -> Dict[str, Any]:
|
|
"""Pause the active evolution campaign without deleting its state.
|
|
|
|
RESUMABLE: a later ``/evolve start`` resumes the SAME campaign in place
|
|
(start_evolution_campaign treats ``paused`` as resumable). Used by the system
|
|
breakers (light mode, consecutive failures, objective-repeat cap, budget reserve,
|
|
restart-blocked) — NOT by an owner stop. For an owner stop use
|
|
``complete_evolution_campaign`` (terminal).
|
|
"""
|
|
from supervisor import state
|
|
|
|
state.assert_test_data_path(state.STATE_PATH)
|
|
lock_fd = state.acquire_file_lock(state.STATE_LOCK_PATH)
|
|
if lock_fd is None:
|
|
return {}
|
|
try:
|
|
campaign = _read_evolution_campaign()
|
|
if campaign:
|
|
campaign["status"] = "paused"
|
|
campaign["updated_at"] = utc_now_iso()
|
|
campaign["pause_reason"] = str(reason or "")
|
|
if not _write_evolution_campaign(campaign, _state_lock_held=True):
|
|
return {}
|
|
return campaign
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
|
|
def complete_evolution_campaign(
|
|
reason: str = "", *, status: str = "stopped", cleanup_worktree: bool = True
|
|
) -> Dict[str, Any]:
|
|
"""Close rather than pause: a later owner start creates a fresh campaign.
|
|
|
|
Archive/remove the active transaction and backlog claim so boot cannot absorb
|
|
a stopped cycle. The owner's sticky evolution_owner_stopped flag independently
|
|
prevents autonomous re-arm. Never raises. Panic passes cleanup_worktree=False:
|
|
no git work may delay its hard exit; boot custody owns later recovery.
|
|
"""
|
|
try:
|
|
from supervisor import state
|
|
|
|
snapshot = _read_evolution_campaign()
|
|
snapshot_tx = snapshot.get("active_transaction")
|
|
cleanup_updates: Dict[str, Any] = {}
|
|
if cleanup_worktree and isinstance(snapshot_tx, dict):
|
|
try:
|
|
_cleanup_worktree_after_cycle(snapshot_tx, str(snapshot_tx.get("task_id") or ""))
|
|
cleanup_updates = {
|
|
key: value for key, value in snapshot_tx.items()
|
|
if key.startswith("cleanup_") or key == "recovery_hint"
|
|
}
|
|
except Exception:
|
|
pass
|
|
state.assert_test_data_path(state.STATE_PATH)
|
|
lock_fd = state.acquire_file_lock(
|
|
state.STATE_LOCK_PATH, timeout_sec=0.001 if not cleanup_worktree else 4.0,
|
|
)
|
|
if lock_fd is None:
|
|
return {}
|
|
try:
|
|
campaign = _read_evolution_campaign()
|
|
if not campaign:
|
|
return campaign or {}
|
|
now = utc_now_iso()
|
|
tx = campaign.get("active_transaction")
|
|
if isinstance(tx, dict):
|
|
if (
|
|
cleanup_updates
|
|
and isinstance(snapshot_tx, dict)
|
|
and str(tx.get("transaction_id") or "")
|
|
== str(snapshot_tx.get("transaction_id") or "")
|
|
):
|
|
tx.update(cleanup_updates)
|
|
tx = {**tx, "cycle_outcome": tx.get("cycle_outcome") or "owner_stopped"}
|
|
append_unique_transaction(campaign, tx)
|
|
campaign.pop("active_transaction", None)
|
|
campaign.pop("post_task_backlog_id", None)
|
|
campaign.pop("pause_reason", None)
|
|
campaign["status"] = str(status or "stopped")
|
|
campaign["updated_at"] = now
|
|
campaign["completed_at"] = now
|
|
campaign["completion_reason"] = str(reason or "")
|
|
return campaign if _write_evolution_campaign(
|
|
campaign, _state_lock_held=True,
|
|
) else {}
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
except Exception:
|
|
log.debug("complete_evolution_campaign failed", exc_info=True)
|
|
return {}
|
|
|
|
|
|
def begin_evolution_transaction(task_id: str, *, cycle: int, campaign: Dict[str, Any],
|
|
transaction: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Attach a cycle, or refresh its base under Q immediately before its receipt.
|
|
|
|
Clear positive-refusal proof BEFORE admission; interruptions keep orphan custody.
|
|
"""
|
|
previous = campaign.get("active_transaction")
|
|
previous = previous if isinstance(previous, dict) else {}
|
|
refusal = previous.get("admission_refused")
|
|
objective_fp = canonical_objective_fingerprint(str(campaign.get("objective") or ""))
|
|
if (transaction is None and isinstance(refusal, dict) and refusal.get("source") == campaign.get("source")
|
|
and previous.get("objective_fp") == objective_fp and not previous.get("commit_sha")):
|
|
return dict(previous)
|
|
expected_transaction = transaction
|
|
try:
|
|
from supervisor import git_ops
|
|
|
|
rc_head, head, _ = git_ops.git_capture(["git", "rev-parse", "HEAD"])
|
|
rc_branch, branch, _ = git_ops.git_capture(["git", "rev-parse", "--abbrev-ref", "HEAD"])
|
|
base_head = head.strip() if rc_head == 0 else ""
|
|
base_branch = branch.strip() if rc_branch == 0 else ""
|
|
except Exception:
|
|
base_head = ""
|
|
base_branch = ""
|
|
transaction = {
|
|
"schema_version": 2,
|
|
"transaction_id": uuid.uuid4().hex[:12],
|
|
"campaign_id": str((campaign or {}).get("id") or ""),
|
|
"task_id": str(task_id or ""),
|
|
"cycle": int(cycle or 0),
|
|
"objective_fp": objective_fp, # cycle-start truth, not a later campaign objective
|
|
"created_at": utc_now_iso(), "updated_at": utc_now_iso(),
|
|
"base_head": base_head, "base_branch": base_branch,
|
|
"preflight_status": "pending", "advisory_status": "pending", "triad_scope_status": "pending",
|
|
"commit_sha": "", "push_status": "pending",
|
|
"restart_decision": "", "restart_required": False, "restart_verified": False, "restart_verified_at": "",
|
|
"rescue_ref": "", "rescue_path": "", "recovery_hint": "",
|
|
}
|
|
if expected_transaction is not None:
|
|
transaction = {**expected_transaction, **{key: transaction[key] for key in
|
|
("base_head", "base_branch", "objective_fp", "cycle")}}
|
|
transaction.pop("admission_refused", None)
|
|
from supervisor import state
|
|
|
|
state.assert_test_data_path(state.STATE_PATH)
|
|
lock_fd = state.acquire_file_lock(state.STATE_LOCK_PATH)
|
|
if lock_fd is None:
|
|
return {}
|
|
try:
|
|
current = _read_evolution_campaign()
|
|
live_state = state.read_state().projection()
|
|
existing_tx = current.get("active_transaction")
|
|
existing_tx = existing_tx if isinstance(existing_tx, dict) else {}
|
|
if (
|
|
not state.control_is(live_state, "evolution_owner_stopped", False)
|
|
or not state.control_is(live_state, "evolution_mode_enabled", True)
|
|
or evolution_stop_reason(current)
|
|
or current.get("status") != "active"
|
|
or any(current.get(key) != campaign.get(key) for key in ("id", "source", "objective"))
|
|
or (expected_transaction is not None and existing_tx != expected_transaction)
|
|
or bool(str(existing_tx.get("commit_sha") or "").strip())
|
|
):
|
|
return {}
|
|
if expected_transaction is not None and existing_tx == transaction:
|
|
return transaction
|
|
if existing_tx and expected_transaction is None:
|
|
existing_tx.update({
|
|
"cycle_outcome": "abandoned",
|
|
"abandoned_reason": "dispatch_not_persisted",
|
|
"updated_at": utc_now_iso(),
|
|
})
|
|
append_unique_transaction(current, existing_tx)
|
|
current["active_transaction"] = transaction
|
|
current["updated_at"] = utc_now_iso()
|
|
if not _write_evolution_campaign(current, _state_lock_held=True):
|
|
return {}
|
|
stored = _read_evolution_campaign()
|
|
stored_tx = stored.get("active_transaction")
|
|
if stored_tx != transaction:
|
|
return {}
|
|
return transaction
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
|
|
def evolution_commit_receipt_error(
|
|
tx: Dict[str, Any], *, campaign_id: str, transaction_id: str,
|
|
task_id: str, commit_sha: str,
|
|
) -> str:
|
|
if str(tx.get("commit_sha") or "") != commit_sha:
|
|
return "commit_receipt_mismatch"
|
|
receipt = tx.get("commit_receipt")
|
|
if not isinstance(receipt, dict) or not bool(receipt.get("ok")):
|
|
return "commit_receipt_missing"
|
|
expected = {
|
|
"campaign_id": campaign_id,
|
|
"transaction_id": transaction_id,
|
|
"task_id": task_id,
|
|
"commit_sha": commit_sha,
|
|
}
|
|
if any(str(receipt.get(key) or "") != value for key, value in expected.items()):
|
|
return "commit_receipt_mismatch"
|
|
return ""
|
|
|
|
|
|
def _evolution_claim_error(
|
|
campaign: Dict[str, Any], live_state: Dict[str, Any], *,
|
|
campaign_id: str, transaction_id: str, task_id: str, commit_sha: str = "",
|
|
require_uncommitted: bool = False,
|
|
) -> str:
|
|
from supervisor.state import control_value
|
|
|
|
if not campaign_id or not transaction_id or not task_id:
|
|
return "claim_identity_missing"
|
|
stopped_known, stopped = control_value(live_state, "evolution_owner_stopped")
|
|
if (stopped_known and stopped) or evolution_stop_reason(campaign):
|
|
return "owner_stopped"
|
|
enabled_known, enabled = control_value(live_state, "evolution_mode_enabled")
|
|
if not stopped_known or (not enabled_known and not commit_sha):
|
|
return "state_controls_unknown" # an unknown control never reads as permission (#1307)
|
|
if not enabled and not commit_sha:
|
|
return "evolution_disabled"
|
|
if campaign.get("status") != "active":
|
|
return "campaign_not_active"
|
|
if str(campaign.get("id") or "") != campaign_id:
|
|
return "campaign_mismatch"
|
|
tx = campaign.get("active_transaction")
|
|
if not isinstance(tx, dict):
|
|
return "transaction_missing"
|
|
if str(tx.get("transaction_id") or "") != transaction_id:
|
|
return "transaction_mismatch"
|
|
if str(tx.get("task_id") or "") != task_id:
|
|
return "task_mismatch"
|
|
if require_uncommitted and str(tx.get("commit_sha") or "").strip():
|
|
return "transaction_already_committed"
|
|
if commit_sha:
|
|
return evolution_commit_receipt_error(
|
|
tx,
|
|
campaign_id=campaign_id,
|
|
transaction_id=transaction_id,
|
|
task_id=task_id,
|
|
commit_sha=commit_sha,
|
|
)
|
|
return ""
|
|
|
|
|
|
def check_evolution_authority(
|
|
campaign_id: str, transaction_id: str, task_id: str, *, commit_sha: str = "",
|
|
require_uncommitted: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Check the exact campaign claim under the existing state lock."""
|
|
from supervisor import state
|
|
|
|
state.assert_test_data_path(state.STATE_PATH)
|
|
lock_fd = state.acquire_file_lock(state.STATE_LOCK_PATH)
|
|
if lock_fd is None:
|
|
return {"ok": False, "reason": "state_lock_unavailable"}
|
|
try:
|
|
campaign = _read_evolution_campaign()
|
|
live_state = state.read_state().projection()
|
|
reason = _evolution_claim_error(
|
|
campaign, live_state,
|
|
campaign_id=str(campaign_id or ""),
|
|
transaction_id=str(transaction_id or ""),
|
|
task_id=str(task_id or ""),
|
|
commit_sha=str(commit_sha or ""),
|
|
require_uncommitted=bool(require_uncommitted),
|
|
)
|
|
return {
|
|
"ok": not reason,
|
|
"reason": reason,
|
|
"campaign_id": str(campaign_id or ""),
|
|
"transaction_id": str(transaction_id or ""),
|
|
"task_id": str(task_id or ""),
|
|
"commit_sha": str(commit_sha or ""),
|
|
}
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
|
|
def _build_commit_receipt(
|
|
campaign_id: str, transaction_id: str, task_id: str, commit_sha: str,
|
|
*, reason: str, recorded_at: str,
|
|
) -> Dict[str, Any]:
|
|
"""The exact receipt shape ``evolution_commit_receipt_error`` validates.
|
|
|
|
One constructor so the receipt written by the tool path and the one
|
|
re-derived by boot recovery are the same durable fact, differing only in
|
|
``reason`` (which records HOW the SHA was attributed).
|
|
"""
|
|
return {
|
|
"ok": True,
|
|
"reason": reason,
|
|
"campaign_id": str(campaign_id),
|
|
"transaction_id": str(transaction_id),
|
|
"task_id": str(task_id),
|
|
"commit_sha": str(commit_sha),
|
|
"recorded_at": recorded_at,
|
|
}
|
|
|
|
|
|
def adopt_evolution_commit_intent(
|
|
campaign: Dict[str, Any], tx: Dict[str, Any], head_sha: str = "",
|
|
) -> str:
|
|
"""Finish an interrupted commit receipt from the pre-commit intent.
|
|
|
|
``git commit`` and its SHA receipt are two writes; a crash between them used
|
|
to leave a reviewed commit on HEAD that no boot path could attribute (the
|
|
markerless reconcile short-circuits on an empty ``commit_sha``). The intent
|
|
written before the commit carries the exact reviewed tree and parents, so
|
|
attribution here is structural rather than a guess: the commit at HEAD is
|
|
adopted only when its tree AND its full parent list are identical to that
|
|
reviewed material. A failed commit, a contained orphan (the branch is rewound
|
|
to the parent) or any later HEAD movement fails the match and recovery stays
|
|
fail-closed. The caller owns persistence, so the recovered SHA and its receipt
|
|
land in the SAME write as the decision that consumed them.
|
|
"""
|
|
intent = tx.get("commit_intent") if isinstance(tx, dict) else None
|
|
if not isinstance(intent, dict):
|
|
return ""
|
|
tree_sha = str(intent.get("tree_sha") or "")
|
|
parents = [str(value) for value in (intent.get("parents") or [])]
|
|
if not tree_sha or not parents:
|
|
return ""
|
|
head = str(head_sha or "").strip() or "HEAD"
|
|
try:
|
|
from supervisor import git_ops
|
|
|
|
rc_tree, actual_tree, _ = git_ops.git_capture(["git", "rev-parse", f"{head}^{{tree}}"])
|
|
rc_parents, parent_line, _ = git_ops.git_capture(
|
|
["git", "rev-list", "--parents", "-n", "1", head]
|
|
)
|
|
except Exception:
|
|
return ""
|
|
fields = parent_line.strip().split() if rc_parents == 0 else []
|
|
if rc_tree != 0 or not fields or actual_tree.strip() != tree_sha or fields[1:] != parents:
|
|
return ""
|
|
commit_sha, now = fields[0], utc_now_iso()
|
|
tx.update({
|
|
"commit_sha": commit_sha,
|
|
"commit_receipt": _build_commit_receipt(
|
|
str(campaign.get("id") or ""), str(tx.get("transaction_id") or ""),
|
|
str(tx.get("task_id") or ""), commit_sha,
|
|
reason="recovered_from_commit_intent", recorded_at=now,
|
|
),
|
|
"commit_receipt_recovered": True,
|
|
"restart_required": True,
|
|
"restart_verified": False,
|
|
"updated_at": now,
|
|
})
|
|
campaign["active_transaction"] = tx
|
|
return commit_sha
|
|
|
|
|
|
def record_evolution_commit(
|
|
campaign_id: str, transaction_id: str, task_id: str, commit_sha: str,
|
|
*, triad_scope_status: str = "unknown", author_disposition: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""CAS the exact reviewed local commit into its still-authorized transaction."""
|
|
from supervisor import state
|
|
|
|
commit_sha = str(commit_sha or "").strip()
|
|
if not commit_sha:
|
|
return {"ok": False, "reason": "commit_sha_missing"}
|
|
state.assert_test_data_path(state.STATE_PATH)
|
|
lock_fd = state.acquire_file_lock(state.STATE_LOCK_PATH)
|
|
if lock_fd is None:
|
|
return {"ok": False, "reason": "state_lock_unavailable", "commit_sha": commit_sha}
|
|
try:
|
|
live_state = state.read_state().projection()
|
|
from ouroboros.utils import update_json_locked
|
|
|
|
receipt: Dict[str, Any] = {}
|
|
refused = {"reason": "campaign_write_refused"}
|
|
|
|
def _mutate(campaign: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
reason = _evolution_claim_error(
|
|
campaign, live_state,
|
|
campaign_id=str(campaign_id or ""),
|
|
transaction_id=str(transaction_id or ""),
|
|
task_id=str(task_id or ""),
|
|
)
|
|
tx = campaign.get("active_transaction")
|
|
prior_sha = str((tx or {}).get("commit_sha") or "") if isinstance(tx, dict) else ""
|
|
if not reason and prior_sha and prior_sha != commit_sha:
|
|
reason = "commit_receipt_conflict"
|
|
if reason:
|
|
refused["reason"] = reason
|
|
return None
|
|
now = utc_now_iso()
|
|
receipt.update(_build_commit_receipt(
|
|
campaign_id, transaction_id, task_id, commit_sha,
|
|
reason="recorded", recorded_at=now,
|
|
))
|
|
tx.update({
|
|
"preflight_status": "passed",
|
|
"advisory_status": "fresh_or_bypassed",
|
|
"triad_scope_status": str(triad_scope_status or "unknown"),
|
|
**({"author_disposition": dict(author_disposition)} if author_disposition else {}),
|
|
"commit_sha": commit_sha,
|
|
"commit_receipt": dict(receipt),
|
|
"restart_required": True,
|
|
"restart_verified": False,
|
|
"updated_at": now,
|
|
})
|
|
campaign["active_transaction"] = tx
|
|
campaign["updated_at"] = now
|
|
return campaign
|
|
|
|
update_json_locked(
|
|
_evolution_campaign_path(), _mutate,
|
|
timeout_sec=EVOLUTION_CAMPAIGN_CAS_TIMEOUT_SEC,
|
|
strict_existing_dict=True,
|
|
)
|
|
if not receipt:
|
|
return {"ok": False, "reason": refused["reason"], "commit_sha": commit_sha}
|
|
return receipt
|
|
except Exception as exc:
|
|
log.warning("Failed to record exact evolution commit receipt", exc_info=True)
|
|
return {"ok": False, "reason": f"campaign_write_failed:{type(exc).__name__}", "commit_sha": commit_sha}
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
|
|
def link_evolution_rescue(drive_root: pathlib.Path, rescue_info: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Attach rescue pointers without racing an exact commit receipt."""
|
|
from supervisor import state
|
|
|
|
root = pathlib.Path(drive_root)
|
|
path = root / EVOLUTION_CAMPAIGN_FILE
|
|
lock_path = root / "locks" / "state.lock"
|
|
state.assert_test_data_path(path)
|
|
lock_fd = state.acquire_file_lock(lock_path)
|
|
if lock_fd is None:
|
|
return {}
|
|
try:
|
|
from ouroboros.utils import update_json_locked
|
|
|
|
linked: Dict[str, Any] = {}
|
|
|
|
def _mutate(campaign: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
if campaign.get("status") not in {"active", "paused"}:
|
|
return None
|
|
tx = campaign.get("active_transaction")
|
|
if not isinstance(tx, dict):
|
|
return None
|
|
tx["rescue_ref"] = str(rescue_info.get("rescue_ref") or "")
|
|
tx["dirty_snapshot_ref"] = str(rescue_info.get("rescue_ref") or "")
|
|
tx["rescue_path"] = str(rescue_info.get("path") or "")
|
|
tx["restart_decision"] = "rescue_snapshot_created"
|
|
tx["recovery_hint"] = (
|
|
f"Recover with {tx['rescue_ref']} or inspect {tx['rescue_path']}"
|
|
if tx.get("rescue_ref") or tx.get("rescue_path")
|
|
else "Rescue attempted; inspect supervisor logs."
|
|
)
|
|
tx["updated_at"] = utc_now_iso()
|
|
campaign["active_transaction"] = tx
|
|
campaign["updated_at"] = utc_now_iso()
|
|
linked.update(tx)
|
|
return campaign
|
|
|
|
update_json_locked(
|
|
path, _mutate,
|
|
timeout_sec=EVOLUTION_CAMPAIGN_CAS_TIMEOUT_SEC,
|
|
strict_existing_dict=True,
|
|
)
|
|
return linked
|
|
finally:
|
|
state.release_file_lock(lock_path, lock_fd)
|
|
|
|
|
|
def _bump_objective_repeat_count(campaign: Dict[str, Any], tx: Dict[str, Any]) -> None:
|
|
"""BUG3: count one non-absorbing cycle against its objective fingerprint.
|
|
|
|
Cumulative PER-FINGERPRINT (not a consecutive streak), so a blocked objective that is
|
|
re-proposed NON-consecutively (interleaved with other no_op work) still accumulates toward
|
|
the pause gate. ``setdefault`` tolerates campaigns persisted before this field existed; a
|
|
transaction without an ``objective_fp`` (e.g. a tx-less idle cycle) is skipped, never
|
|
bucketed under the empty key.
|
|
"""
|
|
fp = str((tx or {}).get("objective_fp") or "")
|
|
if not fp:
|
|
return
|
|
counts = campaign.setdefault("objective_repeat_counts", {})
|
|
counts[fp] = int(counts.get(fp, 0) or 0) + 1
|
|
# Layer B: also mark this objective attempted-and-dropped so the chooser (Layer A) can be
|
|
# told not to re-propose it. This is a campaign-local signal, NOT a backlog status flip:
|
|
# the backlog item stays "open" (the work is genuinely unsolved), we only stop FEEDING it
|
|
# back to the evolution objective chooser.
|
|
dropped = campaign.setdefault("dropped_objective_fps", [])
|
|
if fp not in dropped:
|
|
dropped.append(fp)
|
|
|
|
|
|
def _clear_objective_repeat_count(campaign: Dict[str, Any], tx: Dict[str, Any]) -> None:
|
|
"""BUG3: a genuine absorb clears ONLY this objective's repeat tally and dropped flag.
|
|
|
|
Called at every site that sets ``cycle_outcome == "absorbed"`` (task-done here, plus the
|
|
two durable boot/restart-verify absorb sites in agent_startup_checks). Keyed on the same
|
|
SSOT fingerprint as the bump so success on the looping objective resets exactly its bucket
|
|
and un-drops it (it landed, so it is no longer do-not-re-propose).
|
|
"""
|
|
fp = str((tx or {}).get("objective_fp") or "")
|
|
if not fp:
|
|
return
|
|
counts = campaign.setdefault("objective_repeat_counts", {})
|
|
counts.pop(fp, None)
|
|
dropped = campaign.setdefault("dropped_objective_fps", [])
|
|
if fp in dropped:
|
|
dropped.remove(fp)
|
|
|
|
|
|
def update_evolution_transaction(task_id: str, **updates: Any) -> bool:
|
|
"""Best-effort update of the active/lightweight evolution transaction."""
|
|
from supervisor import state
|
|
|
|
state.assert_test_data_path(state.STATE_PATH)
|
|
lock_fd = state.acquire_file_lock(state.STATE_LOCK_PATH)
|
|
if lock_fd is None:
|
|
return False
|
|
try:
|
|
campaign = _read_evolution_campaign()
|
|
tx = campaign.get("active_transaction")
|
|
if not isinstance(tx, dict) or str(tx.get("task_id") or "") != str(task_id or ""):
|
|
return False
|
|
updates = {key: value for key, value in updates.items() if value is not None}
|
|
if all(tx.get(key) == value for key, value in updates.items()):
|
|
return True
|
|
tx.update(updates)
|
|
tx["updated_at"] = utc_now_iso()
|
|
campaign["active_transaction"] = tx
|
|
campaign["updated_at"] = utc_now_iso()
|
|
return _write_evolution_campaign(campaign, _state_lock_held=True)
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
|
|
def _cleanup_worktree_after_cycle(tx: Dict[str, Any], task_id: str) -> None:
|
|
"""Restore a no-op/abandoned admitted cycle's base, preserving dirty/ahead work
|
|
in recorded stash/local refs. Never reset a positively unadmitted cycle or
|
|
another live writer; unknown base, live tests and the cleanup kill-switch skip
|
|
cleanup with a reason. Never raises (OUROBOROS_EVOLUTION_CYCLE_CLEANUP=false).
|
|
"""
|
|
if str(os.environ.get("OUROBOROS_EVOLUTION_CYCLE_CLEANUP", "true") or "true").lower() in {"0", "false", "no", "off"}:
|
|
tx["cleanup_status"] = "disabled"
|
|
return
|
|
if tx.get("admission_refused") and not tx.get("commit_sha"):
|
|
tx["cleanup_status"] = "skipped_never_admitted"
|
|
return
|
|
base_head = str(tx.get("base_head") or "").strip()
|
|
if not base_head:
|
|
tx["cleanup_status"] = "skipped_no_base"
|
|
return
|
|
update_lock_fh = None
|
|
release_update_lock = None
|
|
try:
|
|
from supervisor import git_ops, queue
|
|
from supervisor.update_merge import acquire_update_lock, release_update_lock
|
|
from supervisor.workers import repo_writer_admission_closed
|
|
|
|
try:
|
|
update_lock_fh = acquire_update_lock()
|
|
except RuntimeError:
|
|
tx["cleanup_status"] = "skipped_managed_update_lock_busy"
|
|
return
|
|
admission_reason = repo_writer_admission_closed()
|
|
if admission_reason:
|
|
tx["cleanup_status"] = "skipped_repo_writer_admission_closed"
|
|
tx["cleanup_deferred_reason"] = admission_reason
|
|
return
|
|
|
|
# Same protection class as git_ops._guard_live_repo_destructive_git, but
|
|
# covering the stash too: a unit test that never re-pointed
|
|
# git_ops.REPO_DIR must not stash/reset the LIVE repo's working tree.
|
|
if os.environ.get("OUROBOROS_ALLOW_LIVE_REPO_TESTS") != "1":
|
|
import sys as _sys
|
|
try:
|
|
live_repo = git_ops.REPO_DIR.resolve(strict=False) == (
|
|
pathlib.Path.home() / "Ouroboros" / "repo"
|
|
).resolve(strict=False)
|
|
except OSError:
|
|
live_repo = False
|
|
if live_repo and ("PYTEST_CURRENT_TEST" in os.environ or "pytest" in _sys.modules):
|
|
tx["cleanup_status"] = "skipped_live_repo_test_guard"
|
|
return
|
|
|
|
# Lock-free RUNNING snapshot is safe because this runs on the SAME
|
|
# single supervisor thread that assigns tasks (dispatch_event ->
|
|
# assign_tasks are sequential); only cancel paths mutate RUNNING from
|
|
# HTTP threads, which can only shrink the set.
|
|
running_others = [tid for tid in list(queue.RUNNING.keys()) if str(tid) != str(task_id)]
|
|
if running_others:
|
|
# The live worktree is shared: a reset would destroy concurrent
|
|
# tasks' work. Leave state for the boot reconcile / next cycle.
|
|
tx["cleanup_status"] = "skipped_other_tasks_running"
|
|
return
|
|
|
|
rc_status, status_out, _ = git_ops.git_capture(["git", "status", "--porcelain"])
|
|
rc_head, head_out, _ = git_ops.git_capture(["git", "rev-parse", "HEAD"])
|
|
if rc_status != 0 or rc_head != 0:
|
|
tx["cleanup_status"] = "skipped_git_unavailable"
|
|
return
|
|
dirty = bool(status_out.strip())
|
|
head = head_out.strip()
|
|
if not dirty and head == base_head:
|
|
tx["cleanup_status"] = "already_clean"
|
|
return
|
|
|
|
if dirty:
|
|
stash_label = f"evolution-cycle-cleanup-{tx.get('transaction_id') or task_id}"
|
|
rc_stash, _, stash_err = git_ops.git_capture(
|
|
["git", "stash", "push", "--include-untracked", "-m", stash_label]
|
|
)
|
|
if rc_stash != 0:
|
|
from ouroboros.utils import truncate_review_artifact
|
|
|
|
# Refuse to reset over unsaved changes (P1: no silent loss).
|
|
tx["cleanup_status"] = "skipped_stash_failed"
|
|
tx["recovery_hint"] = (
|
|
"worktree dirty and stash failed: "
|
|
+ truncate_review_artifact(str(stash_err or "").strip(), 400)
|
|
)
|
|
return
|
|
tx["cleanup_stash"] = stash_label
|
|
|
|
if head != base_head:
|
|
preserved, ref_name = git_ops.preserve_local_ref_branch("HEAD", prefix="evolution-leftover")
|
|
if not preserved:
|
|
tx["cleanup_status"] = "skipped_preserve_failed"
|
|
tx["recovery_hint"] = (
|
|
"HEAD ahead of base_head and preserve-branch failed; left as-is"
|
|
+ (f" (dirty files already saved in stash {tx['cleanup_stash']})." if tx.get("cleanup_stash") else ".")
|
|
)
|
|
return
|
|
tx["cleanup_preserved_ref"] = ref_name
|
|
rc_reset, _, reset_err = git_ops.git_capture(["git", "reset", "--hard", base_head])
|
|
if rc_reset != 0:
|
|
from ouroboros.utils import truncate_review_artifact
|
|
|
|
tx["cleanup_status"] = "reset_failed"
|
|
tx["recovery_hint"] = (
|
|
f"git reset --hard {base_head[:12]} failed: "
|
|
+ truncate_review_artifact(str(reset_err or "").strip(), 400)
|
|
)
|
|
return
|
|
tx["cleanup_status"] = "reset_to_base"
|
|
else:
|
|
tx["cleanup_status"] = "stashed_dirty" # HEAD already at base; only the stash happened
|
|
log.info(
|
|
"Evolution cycle %s cleanup: worktree restored to base %s (stash=%s, preserved=%s)",
|
|
tx.get("transaction_id") or task_id, base_head[:12],
|
|
tx.get("cleanup_stash") or "-", tx.get("cleanup_preserved_ref") or "-",
|
|
)
|
|
except Exception:
|
|
tx["cleanup_status"] = "error"
|
|
log.debug("Evolution cycle worktree cleanup failed", exc_info=True)
|
|
finally:
|
|
if update_lock_fh is not None and release_update_lock is not None:
|
|
release_update_lock(update_lock_fh)
|
|
|
|
|
|
def _persist_evolution_cleanup(campaign_id: str, tx: Dict[str, Any]) -> bool:
|
|
"""Patch cleanup evidence into the exact terminal transaction under the state lock."""
|
|
from supervisor import state
|
|
|
|
lock_fd = state.acquire_file_lock(state.STATE_LOCK_PATH)
|
|
if lock_fd is None:
|
|
return False
|
|
try:
|
|
campaign = _read_evolution_campaign()
|
|
if str(campaign.get("id") or "") != str(campaign_id or ""):
|
|
return False
|
|
tx_id = str(tx.get("transaction_id") or "")
|
|
cleanup = {
|
|
key: tx[key]
|
|
for key in (
|
|
"cleanup_status", "cleanup_stash", "cleanup_preserved_ref",
|
|
"recovery_hint",
|
|
)
|
|
if key in tx
|
|
}
|
|
updated = False
|
|
for item in list(campaign.get("transaction_history") or []):
|
|
if isinstance(item, dict) and str(item.get("transaction_id") or "") == tx_id:
|
|
item.update(cleanup)
|
|
updated = True
|
|
for row in list(campaign.get("history") or []):
|
|
item = row.get("transaction") if isinstance(row, dict) else None
|
|
if isinstance(item, dict) and str(item.get("transaction_id") or "") == tx_id:
|
|
item.update(cleanup)
|
|
updated = True
|
|
if not updated:
|
|
return False
|
|
campaign["updated_at"] = utc_now_iso()
|
|
return _write_evolution_campaign(
|
|
campaign, expected_campaign_id=campaign_id, _state_lock_held=True,
|
|
)
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
|
|
def _resume_evolution_terminal_effects(
|
|
campaign_id: str, task_id: str, tx: Dict[str, Any],
|
|
) -> Dict[str, Any]:
|
|
"""Finish idempotent effects that follow the durable terminal transition."""
|
|
from supervisor import queue
|
|
|
|
resumed = dict(tx or {})
|
|
if resumed.get("cleanup_status") == "pending":
|
|
_cleanup_worktree_after_cycle(resumed, str(task_id or ""))
|
|
try:
|
|
if not _persist_evolution_cleanup(campaign_id, resumed):
|
|
log.warning("Failed to persist evolution cleanup evidence for %s", task_id)
|
|
except Exception:
|
|
log.warning("Failed to persist evolution cleanup evidence for %s", task_id, exc_info=True)
|
|
if (
|
|
resumed.get("cycle_outcome") == "waiting_for_restart"
|
|
and resumed.get("restart_required")
|
|
and not resumed.get("restart_verified")
|
|
):
|
|
try:
|
|
request_evolution_restart(queue.DRIVE_ROOT, resumed, log=log)
|
|
except Exception:
|
|
log.warning("Failed to resume evolution restart request for %s", task_id, exc_info=True)
|
|
# Abandon/absorb reports are staged in the same campaign write as the
|
|
# transition. Delivery clears only the exact report after a successful send.
|
|
deliver_pending_owner_report()
|
|
return resumed
|
|
|
|
|
|
def update_evolution_campaign_after_task(
|
|
task_id: str,
|
|
*,
|
|
cost_usd: Optional[float],
|
|
cost_accounting_status: str = "available",
|
|
outcome_axes: Dict[str, Any],
|
|
rounds: int,
|
|
transaction: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Record an evolution cycle outcome in the active campaign file."""
|
|
from supervisor import state
|
|
|
|
state.assert_test_data_path(state.STATE_PATH)
|
|
lock_fd = state.acquire_file_lock(state.STATE_LOCK_PATH)
|
|
if lock_fd is None:
|
|
return {
|
|
"accepted": True, "persisted": False, "replay": False,
|
|
"reason": "state_lock_unavailable", "transaction": {},
|
|
}
|
|
replay_found = False
|
|
replay_tx: Dict[str, Any] = {}
|
|
campaign_id = ""
|
|
tx: Dict[str, Any] = {}
|
|
try:
|
|
# The complete read/check/mutate/write is protected by the same state lock
|
|
# used by pause, owner stop, boot reconciliation, and transaction updates.
|
|
campaign = _read_evolution_campaign()
|
|
if campaign.get("status") not in {"active", "paused"}:
|
|
return {
|
|
"accepted": False, "persisted": False, "replay": False,
|
|
"reason": "campaign_not_active", "transaction": {},
|
|
}
|
|
metadata_tx = transaction if isinstance(transaction, dict) else {}
|
|
active_tx = (
|
|
campaign.get("active_transaction")
|
|
if isinstance(campaign.get("active_transaction"), dict) else {}
|
|
)
|
|
campaign_id = str(campaign.get("id") or "")
|
|
|
|
def _matches(candidate: Dict[str, Any]) -> bool:
|
|
return bool(
|
|
campaign_id
|
|
and str(candidate.get("campaign_id") or "") == campaign_id
|
|
and str(candidate.get("transaction_id") or "")
|
|
and str(candidate.get("task_id") or "") == str(task_id or "")
|
|
)
|
|
|
|
metadata_tx_id = (
|
|
str(metadata_tx.get("transaction_id") or "") if _matches(metadata_tx) else ""
|
|
)
|
|
if metadata_tx_id:
|
|
for existing in list(campaign.get("history") or []):
|
|
if (
|
|
not isinstance(existing, dict)
|
|
or str(existing.get("task_id") or "") != str(task_id or "")
|
|
):
|
|
continue
|
|
existing_tx = existing.get("transaction")
|
|
if (
|
|
isinstance(existing_tx, dict)
|
|
and str(existing_tx.get("campaign_id") or "") == campaign_id
|
|
and str(existing_tx.get("transaction_id") or "") == metadata_tx_id
|
|
):
|
|
replay_found = True
|
|
replay_tx = dict(existing_tx)
|
|
break
|
|
|
|
if not replay_found and not active_tx and not metadata_tx:
|
|
# Preserve idempotency for old history-only records, but never let a
|
|
# new metadata-less terminal mutate whichever campaign happens to be active.
|
|
for existing in list(campaign.get("history") or []):
|
|
if (
|
|
isinstance(existing, dict)
|
|
and str(existing.get("task_id") or "") == str(task_id or "")
|
|
):
|
|
replay_found = True
|
|
replay_tx = dict(existing.get("transaction") or {})
|
|
break
|
|
if not replay_found:
|
|
return {
|
|
"accepted": False, "persisted": False, "replay": False,
|
|
"reason": "transaction_missing", "transaction": {},
|
|
}
|
|
|
|
if not replay_found:
|
|
if (
|
|
not _matches(active_tx)
|
|
or not _matches(metadata_tx)
|
|
or str(active_tx.get("transaction_id") or "")
|
|
!= str(metadata_tx.get("transaction_id") or "")
|
|
):
|
|
return {
|
|
"accepted": False, "persisted": False, "replay": False,
|
|
"reason": "transaction_mismatch", "transaction": {},
|
|
}
|
|
tx = {**metadata_tx, **active_tx}
|
|
axes = normalize_outcome_axes({"outcome_axes": outcome_axes or {}})
|
|
tx["outcome_axes"] = axes
|
|
tx["updated_at"] = utc_now_iso()
|
|
history = list(campaign.get("history") or [])
|
|
cost_available = cost_accounting_status == "available" and cost_usd is not None
|
|
row = {
|
|
"task_id": str(task_id or ""),
|
|
"ts": utc_now_iso(),
|
|
# ABI-3 (fix-round-3): the honest cost name — this row reaches
|
|
# /api/state through the evolution snapshot. Stored legacy rows
|
|
# (cost_usd) keep resolving deprecated-wins at the readers and
|
|
# at the /api/state projection boundary.
|
|
"accounted_upper_bound_usd": float(cost_usd) if cost_available else None,
|
|
"cost_accounting_status": "available" if cost_available else "unavailable",
|
|
"outcome_axes": axes,
|
|
"rounds": int(rounds or 0),
|
|
"transaction": tx,
|
|
}
|
|
history.append(row)
|
|
campaign["history"] = history[-50:]
|
|
has_commit = bool(str(tx.get("commit_sha") or "").strip())
|
|
if not has_commit:
|
|
# A crash between the reviewed commit and its SHA receipt can reach
|
|
# task-done before boot recovery: adopt the commit the intent proves,
|
|
# so the cycle is classified as commit-bearing instead of ``no_op``.
|
|
has_commit = bool(adopt_evolution_commit_intent(campaign, tx))
|
|
restart_verified = bool(tx.get("restart_verified"))
|
|
has_rescue = bool(str(tx.get("rescue_ref") or "").strip())
|
|
if has_commit and restart_verified:
|
|
tx["cycle_outcome"] = "absorbed"
|
|
campaign["absorbed_cycles_done"] = int(
|
|
campaign.get("absorbed_cycles_done") or 0
|
|
) + 1
|
|
append_unique_transaction(campaign, tx)
|
|
campaign.pop("active_transaction", None)
|
|
_clear_objective_repeat_count(campaign, tx)
|
|
elif has_rescue:
|
|
tx["cycle_outcome"] = "abandoned"
|
|
tx["abandoned_reason"] = "rescue_ref_present"
|
|
tx["cleanup_status"] = "pending"
|
|
append_unique_transaction(campaign, tx)
|
|
campaign.pop("active_transaction", None)
|
|
campaign.pop("post_task_backlog_id", None)
|
|
_bump_objective_repeat_count(campaign, tx)
|
|
elif not has_commit:
|
|
tx["cycle_outcome"] = "no_op"
|
|
tx["restart_required"] = False
|
|
tx["recovery_hint"] = ""
|
|
tx["cleanup_status"] = "pending"
|
|
append_unique_transaction(campaign, tx)
|
|
campaign.pop("active_transaction", None)
|
|
campaign.pop("post_task_backlog_id", None)
|
|
_bump_objective_repeat_count(campaign, tx)
|
|
else:
|
|
tx["cycle_outcome"] = "waiting_for_restart"
|
|
tx["recovery_hint"] = tx.get("recovery_hint") or (
|
|
"Task ended without a reviewed commit plus restart verification; active "
|
|
"transaction retained until restart verifies, repo state is recovered, or it is superseded."
|
|
)
|
|
tx["restart_required"] = True
|
|
if not tx.get("restart_decision"):
|
|
tx["restart_decision"] = "supervisor_auto_requested"
|
|
campaign["active_transaction"] = tx
|
|
if tx.get("cycle_outcome") in {"absorbed", "abandoned"}:
|
|
campaign["pending_owner_report"] = dict(tx)
|
|
campaign["last_task_id"] = str(task_id or "")
|
|
campaign["cycles_done"] = int(campaign.get("cycles_done") or 0) + 1
|
|
execution_status = str((axes.get("execution") or {}).get("status") or "unknown")
|
|
objective_status = str((axes.get("objective") or {}).get("status") or "not_evaluated")
|
|
cost_note = f"${float(cost_usd):.4f}" if cost_available else "unavailable"
|
|
campaign["progress_notes"] = (
|
|
f"Last cycle {task_id}: execution={execution_status}, objective={objective_status}, "
|
|
f"rounds={int(rounds or 0)}, cost={cost_note}."
|
|
)
|
|
if cost_available:
|
|
campaign["budget_spent_usd"] = round(
|
|
float(campaign.get("budget_spent_usd") or 0.0) + float(cost_usd), 6,
|
|
)
|
|
else:
|
|
campaign["cost_accounting_status"] = "unavailable"
|
|
campaign["updated_at"] = utc_now_iso()
|
|
try:
|
|
persisted = _write_evolution_campaign(
|
|
campaign,
|
|
expected_campaign_id=campaign_id,
|
|
_state_lock_held=True,
|
|
)
|
|
except Exception:
|
|
log.warning("Failed to persist evolution terminal for %s", task_id, exc_info=True)
|
|
return {
|
|
"accepted": True, "persisted": False, "replay": False,
|
|
"reason": "campaign_write_failed", "transaction": {},
|
|
}
|
|
if not persisted:
|
|
return {
|
|
"accepted": True, "persisted": False, "replay": False,
|
|
"reason": "campaign_write_refused", "transaction": {},
|
|
}
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
if replay_found:
|
|
replay_tx = _resume_evolution_terminal_effects(campaign_id, str(task_id or ""), replay_tx)
|
|
return {
|
|
"accepted": True, "persisted": True, "replay": True,
|
|
"reason": "duplicate_terminal", "transaction": replay_tx,
|
|
}
|
|
tx = _resume_evolution_terminal_effects(campaign_id, str(task_id or ""), tx)
|
|
return {
|
|
"accepted": True, "persisted": True, "replay": False,
|
|
"reason": "", "transaction": tx,
|
|
}
|
|
|
|
|
|
def build_evolution_task_text(cycle: int) -> str:
|
|
"""Build the next evolution-campaign task prompt."""
|
|
from ouroboros.config import get_evolution_persistent_objective
|
|
|
|
campaign = _read_evolution_campaign()
|
|
if campaign.get("status") != "active":
|
|
return f"EVOLUTION #{cycle}"
|
|
parts = [
|
|
f"EVOLUTION CAMPAIGN {campaign.get('id') or 'active'} — CYCLE #{cycle}",
|
|
"",
|
|
"## Objective",
|
|
str(campaign.get("objective") or "Autonomously improve Ouroboros."),
|
|
]
|
|
steer = get_evolution_persistent_objective()
|
|
if steer:
|
|
parts.extend([
|
|
"",
|
|
"## Owner Standing Steer (optional bias — does NOT override the Objective above)",
|
|
steer,
|
|
])
|
|
progress = str(campaign.get("progress_notes") or "").strip()
|
|
if progress:
|
|
parts.extend(["", "## Progress So Far", progress])
|
|
history = list(campaign.get("history") or [])[-3:]
|
|
if history:
|
|
parts.extend(["", "## Recent Campaign Cycles"])
|
|
for row in history:
|
|
axes = normalize_outcome_axes(row)
|
|
execution_status = str((axes.get("execution") or {}).get("status") or "unknown")
|
|
objective_status = str((axes.get("objective") or {}).get("status") or "not_evaluated")
|
|
# ABI-3 read tolerance: a stored legacy row's spelling resolves
|
|
# deprecated-wins; new rows carry the honest name only.
|
|
_, row_amount = honest_cost_pair_amount(
|
|
row, "accounted_upper_bound_usd", "cost_usd",
|
|
)
|
|
row_cost = (
|
|
f"${row_amount:.4f}"
|
|
if row.get("cost_accounting_status") != "unavailable"
|
|
and row_amount is not None else "unavailable"
|
|
)
|
|
parts.append(
|
|
f"- {row.get('task_id')}: execution={execution_status}, objective={objective_status}; "
|
|
f"rounds={row.get('rounds', 0)}; cost={row_cost}"
|
|
)
|
|
# Fix B (C10.2): surface the durable improvement backlog and recent solve-capability
|
|
# as optional CONTEXT, never a directive. Ouroboros decides what (if anything) to act
|
|
# on — an evolution cycle is NOT obligated to draw from the backlog or repeat past
|
|
# patterns. Injecting them is LLM-first steering, not a hardcoded work order.
|
|
try:
|
|
from ouroboros.evolution_checkpoints import build_solve_capability_digest
|
|
from ouroboros.improvement_backlog import format_backlog_digest
|
|
from ouroboros.utils import truncate_review_artifact
|
|
from supervisor import queue
|
|
|
|
_digest_root = pathlib.Path(queue.DRIVE_ROOT)
|
|
_backlog_digest = format_backlog_digest(_digest_root, limit=8, max_chars=3000)
|
|
if _backlog_digest:
|
|
parts.extend([
|
|
"",
|
|
"## Improvement Backlog (context only — NOT a work order)",
|
|
"Standing nominations from past cycles. Weigh them if useful, but you are "
|
|
"free to pursue the Objective however you judge best; you need not pick "
|
|
"from this list.",
|
|
"",
|
|
_backlog_digest,
|
|
])
|
|
_capability_digest = truncate_review_artifact(build_solve_capability_digest(_digest_root), 2000)
|
|
if _capability_digest:
|
|
parts.extend([
|
|
"",
|
|
"## Recent Solve-Capability (context only)",
|
|
_capability_digest,
|
|
])
|
|
except Exception:
|
|
log.debug("evolution task digest injection failed", exc_info=True)
|
|
parts.extend([
|
|
"",
|
|
"## Execution Contract",
|
|
"- Work as a normal Ouroboros self-improvement task.",
|
|
"- Use standard tests and the normal advisory + triad + scope review flow before committing code.",
|
|
"- Land at most ONE reviewed self-modification commit in this cycle. Fold reviewer fixes into that commit before committing; do not churn follow-up commits.",
|
|
"- After a reviewed commit lands, call request_restart once and stop. Restart verification is the absorption boundary for the cycle.",
|
|
"- An honest no-op is a legitimate outcome when the objective is unsafe, already solved, too broad, or needs owner input; do not commit just to make a cycle non-empty.",
|
|
"- If the best next step is memory/identity/backlog rather than code, update those durable artifacts with provenance, but do not treat that as an absorbed self-evolution cycle.",
|
|
"- A true absorbed self-evolution cycle requires one reviewed self-modification commit followed by successful restart verification before the next campaign cycle.",
|
|
"- The review enforcement mode (advisory vs blocking) is the owner's setting. Do NOT hardcode review findings to always block (or always pass) regardless of that mode: forcing per-finding blocks under an owner-chosen advisory mode is forbidden self-modification (BIBLE P3), not a hardening. If advisory pass-through of a critical finding feels wrong, surface it to the owner — never patch the enforcement gate to override their choice.",
|
|
"- If the objective is complete or needs owner input, say so clearly in the final result.",
|
|
])
|
|
return "\n".join(parts)
|
|
|
|
|
|
def notify_owner_cycle_outcome(campaign: Dict[str, Any], tx: Dict[str, Any]) -> None:
|
|
"""WS-13.5 (e5=ux_absorb_report): owner-facing chat note for a finished
|
|
self-evolution cycle. Absorbed -> short what/why; abandoned -> honest
|
|
warning; no_op / waiting -> quiet (the lifecycle event already records it).
|
|
No web/UI edits; chat only, budget-gated. Lazy imports avoid an import
|
|
cycle with supervisor.state / supervisor.message_bus.
|
|
"""
|
|
outcome = str(tx.get("cycle_outcome") or "")
|
|
if outcome not in ("absorbed", "abandoned"):
|
|
return # no_op / waiting_for_restart: event-only, stay quiet
|
|
from supervisor.state import load_state
|
|
from supervisor.message_bus import send_with_budget
|
|
owner_chat_id = int(load_state().get("owner_chat_id") or 0)
|
|
if not owner_chat_id:
|
|
return
|
|
objective = str(campaign.get("objective") or "").strip()
|
|
obj_short = (objective[:160] + "…") if len(objective) > 160 else objective
|
|
if outcome == "absorbed":
|
|
commit_sha = str(tx.get("commit_sha") or "").strip()[:12]
|
|
msg = (
|
|
f"🧬 Evolution cycle absorbed (commit {commit_sha}).\n"
|
|
f"Objective: {obj_short or 'autonomous self-improvement'}\n"
|
|
"The reviewed self-modification is now live (restart verified). Reply if you want it reverted."
|
|
)
|
|
else:
|
|
reason = str(tx.get("abandoned_reason") or "unspecified")
|
|
msg = (
|
|
f"⚠️ Evolution cycle abandoned (reason: {reason}).\n"
|
|
f"Objective: {obj_short or 'autonomous self-improvement'}\n"
|
|
"No change was absorbed; the transaction was rolled back/closed to unblock the next cycle."
|
|
)
|
|
send_with_budget(owner_chat_id, msg, role="system", system_type="evolution_cycle_outcome")
|
|
|
|
|
|
def clear_pending_owner_report(expected: Dict[str, Any]) -> bool:
|
|
"""Clear only the report that was sent, without overwriting newer campaign state."""
|
|
from ouroboros.utils import update_json_locked
|
|
from supervisor import state
|
|
|
|
path = _evolution_campaign_path()
|
|
state.assert_test_data_path(path)
|
|
lock_fd = state.acquire_file_lock(state.STATE_LOCK_PATH)
|
|
if lock_fd is None:
|
|
return False
|
|
cleared = {"ok": False}
|
|
|
|
def _mutate(campaign: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
if campaign.get("pending_owner_report") != expected:
|
|
return None
|
|
campaign.pop("pending_owner_report", None)
|
|
campaign["updated_at"] = utc_now_iso()
|
|
cleared["ok"] = True
|
|
return campaign
|
|
|
|
try:
|
|
update_json_locked(path, _mutate, strict_existing_dict=True)
|
|
return bool(cleared["ok"])
|
|
finally:
|
|
state.release_file_lock(state.STATE_LOCK_PATH, lock_fd)
|
|
|
|
|
|
def append_unique_transaction(campaign: Dict[str, Any], tx: Dict[str, Any]) -> None:
|
|
tx_history = list(campaign.get("transaction_history") or [])
|
|
tx_id = str(tx.get("transaction_id") or "")
|
|
if tx_id and any(isinstance(item, dict) and str(item.get("transaction_id") or "") == tx_id for item in tx_history):
|
|
campaign["transaction_history"] = tx_history[-50:]
|
|
return
|
|
tx_history.append(dict(tx))
|
|
campaign["transaction_history"] = tx_history[-50:]
|
|
|
|
|
|
def write_pending_restart_marker(
|
|
drive_root: pathlib.Path, *, expected_sha: str, expected_branch: str, reason: str,
|
|
evolution_claim: Optional[Dict[str, Any]] = None,
|
|
) -> pathlib.Path:
|
|
"""One schema of ``state/pending_restart_verify.json`` for both writers (the
|
|
supervisor's evolution restart, the agent's ``restart`` tool); ``verify_restart``
|
|
reads it at boot. The claim key exists only for an exact evolution claim — an
|
|
empty one would read as a claim mismatch."""
|
|
path = pathlib.Path(drive_root) / "state" / "pending_restart_verify.json"
|
|
atomic_write_json(path, {
|
|
"ts": utc_now_iso(),
|
|
"expected_sha": str(expected_sha or "").strip(),
|
|
"expected_branch": str(expected_branch or "").strip(),
|
|
"reason": str(reason or "").strip(),
|
|
**({"evolution_claim": dict(evolution_claim)} if evolution_claim else {}),
|
|
}, trailing_newline=True)
|
|
return path
|
|
|
|
|
|
def request_evolution_restart(drive_root: pathlib.Path, tx: Dict[str, Any], log: Any = None) -> None:
|
|
"""Write the exact restart-verify claim, then ask the server to restart.
|
|
``OUROBOROS_EVOLUTION_AUTO_RESTART`` off skips ONLY the restart: the marker is
|
|
what lets the next boot — the owner's manual one included — attribute the cycle
|
|
by exact claim rather than the weaker markerless reconcile (W4-F3)."""
|
|
commit_sha = str(tx.get("commit_sha") or "").strip()
|
|
if not commit_sha:
|
|
return
|
|
claim = {key: str(tx.get(key) or "") for key in ("campaign_id", "transaction_id", "task_id")}
|
|
claim["commit_sha"] = commit_sha
|
|
authority = check_evolution_authority(**claim)
|
|
if not authority.get("ok"):
|
|
if log is not None:
|
|
log.warning("Automatic evolution restart cancelled: exact authority changed (%s)",
|
|
authority.get("reason") or "unknown")
|
|
return
|
|
try:
|
|
existing = read_json_dict(pathlib.Path(drive_root) / "state" / "pending_restart_verify.json") or {}
|
|
restart_reason = (
|
|
str(existing.get("reason") or "").strip() if existing.get("evolution_claim") == claim else ""
|
|
) or "supervisor_auto_evolution_restart"
|
|
write_pending_restart_marker(
|
|
drive_root, expected_sha=commit_sha, expected_branch=str(tx.get("base_branch") or ""),
|
|
reason=restart_reason, evolution_claim=claim,
|
|
)
|
|
auto_restart = str(os.environ.get("OUROBOROS_EVOLUTION_AUTO_RESTART", "true") or "true").lower()
|
|
if auto_restart in {"0", "false", "no", "off"}:
|
|
if log is not None:
|
|
log.info("Automatic evolution restart is off; the restart-verify marker for %s awaits "
|
|
"a manual restart", commit_sha[:12])
|
|
return
|
|
from supervisor import workers
|
|
|
|
workers.get_event_q().put({
|
|
"type": "restart_request", "reason": restart_reason,
|
|
"evolution_restart": True, "ts": utc_now_iso(),
|
|
})
|
|
except Exception:
|
|
if log is not None:
|
|
log.debug("Failed to request automatic evolution restart", exc_info=True)
|