Skip to content

Fix runtime drift guards and write-path audit bridging #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lisihao wants to merge 1 commit into
base: codex/capsule-proof-adapter-runtime
Choose a base branch
from
Open

Fix runtime drift guards and write-path audit bridging #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions harness/lib/evolution_engine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,10 @@
from capability_registry import LEVEL_REVERSE, _open_db as open_capability_db # type: ignore # noqa: E402
from eval_runner import run_pack # type: ignore # noqa: E402
from failure_miner import mine as mine_failures # type: ignore # noqa: E402
try:
from runtime_bridge import record_legacy_event # type: ignore # noqa: E402
except Exception:
record_legacy_event = None # type: ignore

LEVEL_RANK = {"dead_end": 1, "basic_usable": 2, "default_usable": 3, "closed_loop": 4}
RANK_LEVEL = {v: k for k, v in LEVEL_RANK.items()}
Expand Down Expand Up @@ -178,6 +182,19 @@ def _append_event(sprint_id: str, event: str, severity: str, payload: dict[str,
sprint_events = SPRINTS_DIR / f"{sprint_id}.events.jsonl"
with sprint_events.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(obj, ensure_ascii=False) + "\n")
if record_legacy_event is not None:
try:
# Bridge sprint-scoped legacy events into session-log v2 so
# evolution telemetry cannot drift away from runtime state.
record_legacy_event(
sprint_id,
event,
"solar-evolution-engine",
{"severity": severity, **payload},
harness_dir=HARNESS_DIR,
)
except Exception:
pass


def _event_counts(event_names: set[str]) -> dict[str, int]:
Expand Down
171 changes: 122 additions & 49 deletions harness/lib/graph_scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -398,6 +398,50 @@ def _status_path_for_graph(graph: dict[str, Any], graph_path: str | Path | None
return SPRINTS_DIR / f"{sid}.status.json"


def _status_has_terminal_evidence(sid: str, status: dict[str, Any] | None = None, graph_path: str | Path | None = None) -> bool:
payload = status or {}
state = str(payload.get("status", "")).lower()
if state in {"passed", "completed", "eval_passed"}:
return True
base_dir = Path(graph_path).expanduser().parent if graph_path else SPRINTS_DIR
handoff = (base_dir / f"{sid}.handoff.md").exists() or any(base_dir.glob(f"{sid}.*-handoff.md"))
eval_exists = (
(base_dir / f"{sid}.eval.md").exists()
or (base_dir / f"{sid}.eval.json").exists()
or any(base_dir.glob(f"{sid}.*-eval.md"))
or any(base_dir.glob(f"{sid}.*-eval.json"))
)
return handoff and eval_exists


def _project_status_via_runtime(
status_path: Path,
*,
new_status: str,
actor: str,
event: str,
graph_path: str | Path | None = None,
allow_reopen: bool = False,
status_fields: dict[str, Any] | None = None,
extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
from runtime_status import transition_status # noqa: WPS433

payload = dict(extra or {})
payload["graph_sync"] = True
payload["graph_path"] = str(graph_path or "")
payload["allow_reopen"] = allow_reopen
payload["status_fields"] = dict(status_fields or {})
updated, _message = transition_status(
status_path,
new_status,
event,
actor,
extra=payload,
)
return updated


def _ensure_status_cache_exists_from_graph(
graph: dict[str, Any],
graph_path: str | Path | None,
Expand Down Expand Up @@ -439,19 +483,34 @@ def _ensure_status_cache_exists_from_graph(
"active_node": open_nodes[0] if open_nodes else None,
"open_nodes": open_nodes,
"failed_nodes": failed_nodes,
"history": [
{
"ts": now,
"event": event,
"by": actor,
"note": "created missing status cache from task_graph",
}
],
"history": [],
}
# Seed legacy cache once, then immediately bridge through transition_status
# so session-log v2 and compatibility status.json stay aligned.
tmp = status_path.with_suffix(status_path.suffix + ".tmp")
tmp.write_text(json.dumps(status, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
os.replace(tmp, status_path)
return status
return _project_status_via_runtime(
status_path,
new_status="active",
actor=actor,
event=event,
graph_path=graph_path,
status_fields={
"phase": "graph_in_progress",
"handoff_to": "builder_main",
"target_role": "builder_main",
"task_graph": str(graph_path or ""),
"graph_status_cache": True,
"graph_parent_ready": status.get("graph_parent_ready", {}),
"active_node": status.get("active_node"),
"open_nodes": status.get("open_nodes", []),
"failed_nodes": status.get("failed_nodes", []),
"stage": "graph_in_progress",
"task_graph_status": "active",
},
extra={"note": "created missing status cache from task_graph"},
)


def sync_status_cache_from_graph(
Expand Down Expand Up @@ -508,28 +567,43 @@ def sync_status_cache_from_graph(
if not isinstance(history, list):
history = []
if str(current.get("status") or "").lower() == "passed":
history.append({
"ts": now,
"event": "graph_parent_ready_revoked",
"by": actor,
"note": "task_graph no longer satisfies parent_ready_check; reopening legacy status cache",
})
current.update({
"status": "active",
"phase": "graph_in_progress",
"stage": "graph_in_progress",
"active_node": desired_active_node,
"open_nodes": open_nodes,
"failed_nodes": failed_nodes,
"graph_parent_ready": parent,
"task_graph_status": "active",
"updated_at": now,
"completed_at": None,
"history": history,
})
tmp = status_path.with_suffix(status_path.suffix + ".tmp")
tmp.write_text(json.dumps(current, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
os.replace(tmp, status_path)
if _status_has_terminal_evidence(sid, current, graph_path):
current = _project_status_via_runtime(
status_path,
new_status="passed",
actor=actor,
event="graph_parent_ready_preserved_terminal",
graph_path=graph_path,
status_fields={
"phase": str(current.get("phase") or "completed"),
"stage": str(current.get("stage") or "completed"),
"graph_parent_ready": parent,
"task_graph_status": str(current.get("task_graph_status") or "passed"),
"active_node": None,
},
extra={"note": "terminal closeout evidence preserved while parent projection refreshed"},
)
result.update({"updated": True, "status": current, "reason": "terminal_evidence_preserved"})
return result
current = _project_status_via_runtime(
status_path,
new_status="active",
actor=actor,
event="graph_parent_ready_revoked",
graph_path=graph_path,
allow_reopen=True,
status_fields={
"phase": "graph_in_progress",
"stage": "graph_in_progress",
"active_node": desired_active_node,
"open_nodes": open_nodes,
"failed_nodes": failed_nodes,
"graph_parent_ready": parent,
"task_graph_status": "active",
"completed_at": None,
},
extra={"note": "task_graph no longer satisfies parent_ready_check; reopening legacy status cache"},
)
result.update({"updated": True, "status": current, "reason": "parent_reopened"})
return result
projection_changed = any([
Expand All @@ -540,24 +614,23 @@ def sync_status_cache_from_graph(
str(current.get("task_graph_status") or "") != "active",
])
if projection_changed:
history.append({
"ts": now,
"event": "graph_parent_projection_refreshed",
"by": actor,
"note": "task_graph changed while in flight; refreshing legacy status projection",
})
current.update({
"active_node": desired_active_node,
"open_nodes": open_nodes,
"failed_nodes": failed_nodes,
"graph_parent_ready": parent,
"task_graph_status": "active",
"updated_at": now,
"history": history,
})
tmp = status_path.with_suffix(status_path.suffix + ".tmp")
tmp.write_text(json.dumps(current, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
os.replace(tmp, status_path)
current = _project_status_via_runtime(
status_path,
new_status=str(current.get("status") or "active"),
actor=actor,
event="graph_parent_projection_refreshed",
graph_path=graph_path,
status_fields={
"phase": str(current.get("phase") or "graph_in_progress"),
"stage": str(current.get("stage") or "graph_in_progress"),
"active_node": desired_active_node,
"open_nodes": open_nodes,
"failed_nodes": failed_nodes,
"graph_parent_ready": parent,
"task_graph_status": "active",
},
extra={"note": "task_graph changed while in flight; refreshing legacy status projection"},
)
result.update({"updated": True, "status": current, "reason": "parent_projection_refreshed"})
return result
result["reason"] = "parent_not_ready"
Expand Down
112 changes: 112 additions & 0 deletions harness/lib/runtime_doctor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,117 @@ def _check_status_json(sprint_id: str) -> Dict[str, Any]:
return {"ok": False, "warn": True, "message": f"corrupt: {exc}"}


def _artifact_exists_for_sprint(sprint_id: str, suffix: str) -> bool:
base = Path(SPRINTS_DIR)
if (base / f"{sprint_id}.{suffix}").exists():
return True
if suffix in {"design.md", "plan.md", "handoff.md", "eval.md", "eval.json"}:
return any(base.glob(f"{sprint_id}.*-{suffix}"))
return False


def _normalize_state_entry(entry: Any) -> str:
if isinstance(entry, dict):
return str(entry.get("status", "")).lower()
return str(entry or "").lower()


def _check_state_surface_drift(sprint_id: str) -> Dict[str, Any]:
"""Compare status / graph / state surfaces for obvious closeout drift."""
status_path = Path(SPRINTS_DIR) / f"{sprint_id}.status.json"
graph_path = Path(SPRINTS_DIR) / f"{sprint_id}.task_graph.json"
state_path = Path(SPRINTS_DIR) / f"{sprint_id}.task_dag.state.json"
issues: List[str] = []
details: Dict[str, Any] = {
"status_path": str(status_path),
"graph_path": str(graph_path),
"state_path": str(state_path),
"status": {},
"graph_parent_ready": None,
"artifact_evidence": {},
}

try:
status = json.loads(status_path.read_text(encoding="utf-8")) if status_path.exists() else {}
except Exception as exc:
return {"ok": False, "warn": True, "message": f"status corrupt: {exc}"}

status_value = str(status.get("status", "")).lower()
phase_value = str(status.get("phase", "")).lower()
terminal = status_value in {"passed", "completed", "eval_passed", "failed", "cancelled", "archived", "skipped", "superseded"}
terminal_pass_like = status_value in {"passed", "completed", "eval_passed"}
handoff_exists = _artifact_exists_for_sprint(sprint_id, "handoff.md")
eval_exists = _artifact_exists_for_sprint(sprint_id, "eval.md") or _artifact_exists_for_sprint(sprint_id, "eval.json")
details["artifact_evidence"] = {"handoff": handoff_exists, "eval": eval_exists}
details["status"] = {
"status": status_value,
"phase": phase_value,
"stage": str(status.get("stage", "")).lower(),
"task_graph_status": str(status.get("task_graph_status", "")).lower(),
}

graph_ready: Optional[bool] = None
state_closed: Optional[bool] = None

if graph_path.exists():
try:
graph = json.loads(graph_path.read_text(encoding="utf-8"))
except Exception as exc:
issues.append(f"graph_corrupt:{exc}")
graph = {}
if graph:
try:
sys.path.insert(0, os.path.dirname(__file__))
from graph_scheduler import parent_ready_check # noqa: WPS433

parent = parent_ready_check(graph)
except Exception as exc:
parent = {"ready": False, "error": str(exc)}
details["graph_parent_ready"] = parent
graph_ready = parent.get("ready") is True
if terminal_pass_like and parent.get("ready") is not True:
issues.append("terminal_status_parent_not_ready")

if state_path.exists():
try:
state = json.loads(state_path.read_text(encoding="utf-8"))
except Exception as exc:
issues.append(f"state_corrupt:{exc}")
state = {}
if state:
node_results = state.get("node_results") if isinstance(state.get("node_results"), dict) else {}
gate_results = state.get("gate_results") if isinstance(state.get("gate_results"), dict) else {}
open_state_nodes = sorted(
node_id for node_id, entry in node_results.items()
if _normalize_state_entry(entry) not in {"passed", "failed", "cancelled", "skipped", "completed", "eval_passed"}
)
open_state_gates = sorted(
gate_id for gate_id, entry in gate_results.items()
if _normalize_state_entry(entry) not in {"passed", "failed", "cancelled", "skipped", "completed", "eval_passed"}
)
details["state_open_nodes"] = open_state_nodes
details["state_open_gates"] = open_state_gates
state_closed = not open_state_nodes and not open_state_gates
if terminal_pass_like and (open_state_nodes or open_state_gates):
issues.append("terminal_status_state_not_closed")

if handoff_exists and eval_exists and not terminal:
if graph_ready is True:
if state_closed is not False:
issues.append("terminal_evidence_nonterminal_status")
elif graph_ready is None:
issues.append("terminal_evidence_nonterminal_status")

message = "no drift" if not issues else ",".join(issues[:6])
return {
"ok": not issues,
"warn": bool(issues),
"message": message,
"issues": issues,
"details": details,
}


def _check_interface_health(sprint_id: str) -> Dict[str, Any]:
"""Check runtime interface layer health: modules importable, adapters exist."""
sys.path.insert(0, os.path.dirname(__file__))
Expand Down Expand Up @@ -462,6 +573,7 @@ def doctor_sprint(
"duplicate_commands": _check_duplicate_commands(sprint_id),
"stale_activities": _check_stale_activities(sprint_id),
"status_json": _check_status_json(sprint_id),
"state_surface_drift": _check_state_surface_drift(sprint_id),
"interface_health": _check_interface_health(sprint_id),
"context_runtime": _check_context_runtime(sprint_id),
"model_call_runtime": _check_model_call_runtime(sprint_id),
Expand Down
Loading