diff --git a/research/experiments/STUDY_CONTEXT.md b/research/experiments/STUDY_CONTEXT.md index 24c998f..13b8b1a 100644 --- a/research/experiments/STUDY_CONTEXT.md +++ b/research/experiments/STUDY_CONTEXT.md @@ -132,12 +132,16 @@ against the original scientific validator after optional null placeholders are restored. A later executor error preserves the completed model response and its physical inference timing. -A `context_page_read` ledger event records the exact locally constructed page -and its source cursor before the parent episode checkpoint. It does not prove -physical model delivery. The request record and actual CLI trace identify the -page or reference representation sent; the verified transport outcome records -whether that request completed. Native AIDE prompt selection must preserve its -ordinary saved Journal, action receipts and RNG state. +A `context_page_read` ledger event records the source segment and exact cursor +before the parent episode checkpoint. The complete delivered page remains in +the canonical action history; the event points to it instead of copying its +contents. This follows the same checkpoint-plus-reference pattern used by +LangGraph's persistence layer and avoids multiplying large evidence pages in +append-only audit logs. The event does not prove physical model delivery. The +request record and actual CLI trace identify the page or reference +representation sent; the verified transport outcome records whether that +request completed. Native AIDE prompt selection must preserve its ordinary +saved Journal, action receipts and RNG state. ## Regression coverage diff --git a/research/experiments/aide_runtime.py b/research/experiments/aide_runtime.py index 4969e78..b0c3965 100644 --- a/research/experiments/aide_runtime.py +++ b/research/experiments/aide_runtime.py @@ -855,10 +855,14 @@ def _restore_node(record, journal, upstream): def _save(runtime, work_id, state, *, event_type=None, event_payload=None): + from .study_runtime import _compact_action_history_frontiers + stored_state = dict(state) + if isinstance(stored_state.get("action_history"), list): + stored_state["action_history"] = _compact_action_history_frontiers(stored_state["action_history"]) with runtime.ledger.connect() as db: db.execute("BEGIN IMMEDIATE") db.execute("INSERT INTO forest_study_aide_sessions VALUES(?,?,?,?) ON CONFLICT(campaign_id,work_id) DO UPDATE SET state_json=excluded.state_json,updated_at=excluded.updated_at", - (runtime.campaign_id, work_id, _json(state), time.time())) + (runtime.campaign_id, work_id, _json(stored_state), time.time())) if event_type is not None: db.execute("INSERT INTO forest_study_events VALUES(?,?,?,?,?,?)", (str(uuid.uuid4()), runtime.campaign_id, work_id, event_type, @@ -923,6 +927,18 @@ def run_aide_episode(runtime, episode_item: Mapping[str, Any], *, client, row = db.execute("SELECT state_json FROM forest_study_aide_sessions WHERE campaign_id=? AND work_id=?", (runtime.campaign_id, work_id)).fetchone() state = json.loads(row[0]) if row else None + if state is not None: + from .study_runtime import _compact_action_history_frontiers, _expand_action_history_frontiers + state["action_history"] = _expand_action_history_frontiers(state.get("action_history", [])) + compacted = dict(state) + compacted["action_history"] = _compact_action_history_frontiers(state["action_history"]) + compacted_json = _json(compacted) + if len(compacted_json) < len(row[0]): + # Migrate legacy full snapshots losslessly on recovery. Event + # records and the reconstructed in-memory state remain untouched. + with runtime.ledger.connect() as db: + db.execute("UPDATE forest_study_aide_sessions SET state_json=?,updated_at=? WHERE campaign_id=? AND work_id=?", + (compacted_json, time.time(), runtime.campaign_id, work_id)) if state is not None and state.get("finished"): return dict(state["final_outcome"]) _restore_pending_task_context(state, payload) diff --git a/research/experiments/end_to_end_runtime.py b/research/experiments/end_to_end_runtime.py index f88fb97..f804651 100644 --- a/research/experiments/end_to_end_runtime.py +++ b/research/experiments/end_to_end_runtime.py @@ -13,6 +13,8 @@ from typing import Any, Mapping, Sequence import uuid +from .study_runtime import _decode_study_result_json + SCHEMA = """ CREATE TABLE IF NOT EXISTS forest_study_end_to_end ( campaign_id TEXT NOT NULL, work_id TEXT NOT NULL, state_json TEXT NOT NULL, @@ -159,7 +161,7 @@ def _episode_history(runtime, work_id): (runtime.campaign_id, work_id)).fetchone() calls = db.execute("SELECT request_id,status FROM forest_study_model_calls WHERE campaign_id=? AND work_id=? ORDER BY recorded_at,request_id", (runtime.campaign_id, work_id)).fetchall() - result = json.loads(row["result_json"] or "{}") if row else {} + result = _decode_study_result_json(row["result_json"]) if row else {} return {"last_actual_decision": json.loads(row["decision_json"]) if row and row["decision_json"] else None, "actual_action_history": result.get("action_history", []), "model_response_records": [dict(call) for call in calls]} @@ -267,7 +269,7 @@ def _checked_scores(runtime, state, observations): if row is None or row["state"] != "completed": pending.append(binding) continue - result = json.loads(row["result_json"] or "{}") + result = _decode_study_result_json(row["result_json"]) observation = next(value for value in observations if value["receipt_path"] == binding["reference_receipt_path"]) actual = result.get("metrics") or result.get("validation_metrics") or {} if result.get("status") != "SUCCEEDED" or not isinstance(actual, Mapping): @@ -318,7 +320,7 @@ def build_end_to_end_episode_request(runtime, payload: Mapping[str, Any]) -> dic known_rows = _work_rows(runtime, [value["work_id"] for value in state["bindings"].values()]) select_values = [(binding, known_rows.get(binding["work_id"])) for binding in state["bindings"].values() if binding.get("analysis_operation") == "end_to_end_select"] - completed_selects = [(binding, json.loads(row["result_json"] or "{}")) for binding, row in select_values + completed_selects = [(binding, _decode_study_result_json(row["result_json"])) for binding, row in select_values if row and row["state"] == "completed"] recorded_selection_ids = {value["selection_work_id"] for value in state["comparison_records"]} new_selections = [(binding, result) for binding, result in completed_selects @@ -556,7 +558,7 @@ def verify_end_to_end_submission(runtime, payload, history, submission): row = rows.get(binding["work_id"], {}) if row.get("state") != "completed" or json.loads(row["payload_json"]) != binding: continue - result = json.loads(row["result_json"] or "{}") + result = _decode_study_result_json(row["result_json"]) differences = result.get("paired_differences_right_minus_left") or [] if (result.get("status") != "SUCCEEDED" or result.get("operation") != "compare" or result.get("comparison_record") != current or result.get("metric") != binding["metric"] diff --git a/research/experiments/study_memory.py b/research/experiments/study_memory.py index 6a3f2f5..3f10814 100644 --- a/research/experiments/study_memory.py +++ b/research/experiments/study_memory.py @@ -11,6 +11,9 @@ from research.agents.context_store import ContextStore +_LOSSLESS_STORAGE_ENCODING = 'forest.lossless-json-subtrees.v1' + + class ContextQueryError(ValueError): """A caller supplied an invalid query for existing public evidence.""" @@ -35,9 +38,26 @@ def save_json(self, key, value, *, audience='arm-public', origin=None): if audience != 'arm-public': raise ValueError('Study memory accepts only an already filtered arm-public view') text = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False) - identifier = self.store.put('study-public:' + str(key), text, + stored_text = text + stored_encoding = None + from .study_context import intern_json_with_stats + encoded, stats = intern_json_with_stats(json.loads(text)) + if stats['saved_characters'] > 0: + stored_text = json.dumps(encoded, ensure_ascii=False, sort_keys=True, + separators=(',', ':'), allow_nan=False) + stored_encoding = _LOSSLESS_STORAGE_ENCODING + identifier = self.store.put('study-public:' + str(key), stored_text, origin=origin or str(key), authority='untrusted_evidence') metadata = self.store.index['segments'][identifier] + # Catalog character counts and page cursors describe the logical source, + # while the immutable backing file can use a lossless shared-subtree + # representation. Existing plain segments remain readable unchanged. + metadata['characters'] = len(text) + if stored_encoding: + metadata['storage_encoding'] = stored_encoding + else: + metadata.pop('storage_encoding', None) + self.store.save() return {**metadata, 'segment_id': identifier, 'audience': audience, 'read_operation': 'read_context_segment'} @@ -76,6 +96,11 @@ def read_json(self, segment_id, *, pointer='', offset=0, limit=6000, metadata = self.store.index['segments'][segment_id] path = self.store.root / (segment_id + '.txt') value = json.loads(path.read_text(encoding='utf-8')) + if metadata.get('storage_encoding') == _LOSSLESS_STORAGE_ENCODING: + from .study_context import restore_json + value = restore_json(value) + elif metadata.get('storage_encoding') is not None: + raise ContextQueryError('Unknown study-memory storage encoding') if not isinstance(pointer, str) or pointer and not pointer.startswith('/'): raise ContextQueryError('A JSON Pointer must be empty or start with /') if pointer: diff --git a/research/experiments/study_runtime.py b/research/experiments/study_runtime.py index 4b470c1..65baec7 100644 --- a/research/experiments/study_runtime.py +++ b/research/experiments/study_runtime.py @@ -68,6 +68,23 @@ campaign_id TEXT NOT NULL, work_id TEXT NOT NULL, node_id TEXT NOT NULL, code TEXT NOT NULL, program_json TEXT NOT NULL, recorded_at REAL NOT NULL, PRIMARY KEY(campaign_id,work_id,node_id)); +CREATE INDEX IF NOT EXISTS forest_study_work_live_state + ON forest_study_work(campaign_id,state,created_at,work_id,kind); +CREATE INDEX IF NOT EXISTS forest_study_work_live_ready_refs + ON forest_study_work(campaign_id,state,created_at,work_id,kind,supervisor_work_id) + WHERE state IN ('queued','retryable'); +CREATE INDEX IF NOT EXISTS forest_study_work_live_unresolved_refs + ON forest_study_work(campaign_id,state,work_id,supervisor_work_id) + WHERE state NOT IN ('completed','failed','cancelled','not_run','budget_exhausted'); +CREATE INDEX IF NOT EXISTS forest_study_work_pending_parent_state + ON forest_study_work(campaign_id,state,kind,work_id,decision_json,updated_at) + WHERE state='awaiting_reconciliation' AND kind IN ('episode','replay'); +CREATE INDEX IF NOT EXISTS forest_study_work_pending_compute_refs + ON forest_study_work(campaign_id,state,kind,work_id,supervisor_work_id) + WHERE state='awaiting_reconciliation' AND kind IN ('fit','infer'); +CREATE INDEX IF NOT EXISTS forest_study_model_call_unsettled + ON forest_study_model_calls(campaign_id,work_id) + WHERE status='requested'; """ TERMINAL = {"completed", "failed", "cancelled", "not_run", "budget_exhausted"} @@ -80,6 +97,112 @@ def _json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, allow_nan=False) +_SOURCE_FRONTIER_REF = "source_frontier_ref_turn" + + +def _compact_action_history_frontiers(history: Sequence[Any]) -> list[Any]: + """Store each unchanged public evidence frontier once, with lossless turn references.""" + compacted = [] + representatives: list[tuple[int, Any]] = [] + for index, value in enumerate(history): + if not isinstance(value, Mapping): + compacted.append(value) + continue + turn = dict(value) + frontier = turn.pop("source_frontier", None) + if frontier is not None: + reference = next((prior_index for prior_index, prior_frontier in reversed(representatives) + if prior_frontier is frontier or prior_frontier == frontier), None) + if reference is None: + turn["source_frontier"] = frontier + representatives.append((index, frontier)) + else: + turn[_SOURCE_FRONTIER_REF] = reference + compacted.append(turn) + return compacted + + +def _expand_action_history_frontiers(history: Sequence[Any]) -> list[Any]: + """Reconstruct complete history while sharing immutable frontier objects in memory.""" + expanded = [] + representatives: dict[int, Any] = {} + for index, value in enumerate(history): + if not isinstance(value, Mapping): + expanded.append(value) + continue + turn = dict(value) + reference = turn.pop(_SOURCE_FRONTIER_REF, None) + if reference is not None: + if isinstance(reference, bool) or not isinstance(reference, int) or reference not in representatives: + raise ValueError("Study history contains an invalid source-frontier reference") + turn["source_frontier"] = representatives[reference] + elif "source_frontier" in turn: + frontier = turn["source_frontier"] + same = next((prior_index for prior_index, prior_frontier in reversed(list(representatives.items())) + if prior_frontier is frontier or prior_frontier == frontier), None) + if same is not None: + turn["source_frontier"] = representatives[same] + else: + representatives[index] = frontier + expanded.append(turn) + return expanded + + +def _compact_study_result(result: Mapping[str, Any]) -> dict[str, Any]: + compact = dict(result) + history = compact.get("action_history") + if isinstance(history, list): + compact["action_history"] = _compact_action_history_frontiers(history) + return compact + + +def _decode_study_result_json(raw: str | None) -> dict[str, Any]: + value = json.loads(raw or "{}") + if not isinstance(value, dict): + return {} + history = value.get("action_history") + if isinstance(history, list): + value["action_history"] = _expand_action_history_frontiers(history) + return value + + +def _matching_frontier_turns(history: Sequence[Mapping[str, Any]], frontier: Mapping[str, Any]) -> set[int]: + """Compare once per distinct frontier epoch, not once for every repeated history turn.""" + matched_frontiers = set() + for index, turn in enumerate(history): + if not isinstance(turn, Mapping): + continue + previous = turn.get("source_frontier") + if previous is None or id(previous) in matched_frontiers: + continue + if previous is frontier or previous == frontier: + matched_frontiers.add(id(previous)) + return {index for index, turn in enumerate(history) + if isinstance(turn, Mapping) and isinstance(turn.get("source_frontier"), Mapping) + and id(turn["source_frontier"]) in matched_frontiers} + + +def _result_event_delta(result: Mapping[str, Any], previous_history: Sequence[Any]) -> dict[str, Any]: + """Append only changed history and reference unchanged frontiers from earlier turns.""" + history = result.get("action_history") + if not isinstance(history, list): + return dict(result) + previous = list(previous_history) if isinstance(previous_history, Sequence) else [] + shared = 0 + while shared < min(len(previous), len(history)): + if previous[shared] is history[shared] or previous[shared] == history[shared]: + shared += 1 + else: + break + compact = {key: value for key, value in result.items() if key != "action_history"} + compact["action_history_delta"] = { + "replace_from_turn": shared, + "turn_count": len(history), + "turns": _compact_action_history_frontiers(history)[shared:], + } + return compact + + def _transport_trace_reference(campaign: Path, request_id: str, provider: Mapping[str, Any]) -> str | None: """Return a durable, request-unique local Codex trace location.""" if provider.get("kind") != "codex_cli": @@ -222,6 +345,13 @@ def _intern_delivered_messages(messages: Sequence[Mapping[str, Any]]) -> tuple[l return intern_messages(messages) +def _decision_schema_note(schema: Mapping[str, Any], *, provider_kind: str) -> str: + """Avoid serializing Codex CLI's already-attached output schema twice.""" + if provider_kind == "codex_cli": + return "\n\nReturn JSON matching the attached response schema." + return "\n\nReturn exactly one JSON object satisfying this decision contract:\n" + _json(schema) + + def _validate_action_bindings(payload: Mapping[str, Any], actions: Sequence[Any]) -> None: """Check an entire proposed batch before any registered operation can run.""" bindings = payload.get("action_bindings") @@ -308,7 +438,7 @@ def register_action(self, campaign_id: str, work_id: str, payload: Mapping[str, raise ValueError(f"Unsupported bound action kind: {kind}") with self.connect() as db: db.execute("BEGIN IMMEDIATE") - existing = db.execute("SELECT * FROM forest_study_work WHERE work_id=?", (work_id,)).fetchone() + existing = db.execute("SELECT campaign_id,payload_json FROM forest_study_work WHERE work_id=?", (work_id,)).fetchone() encoded = _json(value) if existing: if existing["campaign_id"] != campaign_id or existing["payload_json"] != encoded: @@ -350,7 +480,10 @@ def register(self, manifest: Mapping[str, Any]) -> dict[str, int]: "calibration_decision", "snapshot_stage"}: raise ValueError(f"Runnable work {work_id} has unsupported or missing kind") stage = str(item.get("stage", item.get("phase", item.get("category", "")))) - prior = db.execute("SELECT * FROM forest_study_work WHERE work_id=?", (work_id,)).fetchone() + # Re-registering the manifest only compares its immutable plan + # fields. Do not hydrate result_json: completed episodes may + # retain hundreds of MB of complete scientific history there. + prior = db.execute("SELECT campaign_id,payload_json,state FROM forest_study_work WHERE work_id=?", (work_id,)).fetchone() payload_json = _json(dict(item)) if prior: if prior["campaign_id"] != campaign_id or prior["payload_json"] != payload_json: @@ -453,6 +586,17 @@ def claim(self, campaign_id: str, owner: str, *, work_ids: Sequence[str] | None value["lease_owner"] = owner value["lease_at"] = now value["payload"] = json.loads(value.pop("payload_json")) + value["result"] = _decode_study_result_json(value.get("result_json")) + # Older campaign checkpoints may predate lossless frontier + # interning. Compact them as part of the existing claim + # transaction so repeated immutable research context is + # stored once without dropping any evidence or events. + compact_result_json = _json(_compact_study_result(value["result"])) + prior_result_json = value.get("result_json") or "{}" + if len(compact_result_json) < len(prior_result_json): + db.execute("UPDATE forest_study_work SET result_json=? WHERE campaign_id=? AND work_id=?", + (compact_result_json, campaign_id, value["work_id"])) + value["result_json"] = compact_result_json return value return None @@ -475,7 +619,7 @@ def release_claim(self, campaign_id: str, work_id: str, owner: str, *, state: st raise ValueError(f"Cannot release a claim not held in claimed state by {owner}: {work_id}") def set_state(self, campaign_id: str, work_id: str, state: str, *, result=None, error=None, - supervisor_work_id=None, decision=None): + supervisor_work_id=None, decision=None, event_result=None): if state not in TERMINAL | {"queued", "claimed", "running", "awaiting_reconciliation", "retryable", "blocked", "paused_no_progress"}: raise ValueError(f"Unknown study state: {state}") now = time.time() @@ -484,10 +628,11 @@ def set_state(self, campaign_id: str, work_id: str, state: str, *, result=None, if not row: raise ValueError(f"Unknown study work ID: {work_id}") db.execute("UPDATE forest_study_work SET state=?,result_json=COALESCE(?,result_json),error=?,supervisor_work_id=COALESCE(?,supervisor_work_id),decision_json=COALESCE(?,decision_json),updated_at=? WHERE work_id=?", - (state, _json(result) if result is not None else None, error, supervisor_work_id, + (state, _json(_compact_study_result(result)) if result is not None else None, error, supervisor_work_id, _json(decision) if decision is not None else None, now, work_id)) self.event(campaign_id, "work_state", {"state": state, "error": error, - "supervisor_work_id": supervisor_work_id, "result": result}, work_id) + "supervisor_work_id": supervisor_work_id, + "result": result if event_result is None else event_result}, work_id) def resume_no_progress(self, campaign_id: str, work_id: str, *, operator: str) -> None: """Explicitly reopen a paused episode while retaining its prior evidence.""" @@ -510,12 +655,15 @@ def resume_no_progress(self, campaign_id: str, work_id: str, *, operator: str) - (campaign_id, work_id)).fetchone() if session: state = json.loads(session["state_json"]) + state["action_history"] = _expand_action_history_frontiers(state.get("action_history", [])) pending = state.get("pending") if isinstance(pending, Mapping) and (pending.get("execution") or {}).get("state") == "paused_no_progress": pending = dict(pending) state["pending"] = None state["resume_after_no_progress"] = {"at": now, "operator": operator, "previous_pending_node_id": pending.get("node_id")} + state["action_history"] = _compact_action_history_frontiers( + state.get("action_history", [])) db.execute("UPDATE forest_study_aide_sessions SET state_json=?,updated_at=? WHERE campaign_id=? AND work_id=?", (_json(state), now, campaign_id, work_id)) db.execute("INSERT INTO forest_study_events VALUES(?,?,?,?,?,?)", @@ -1121,8 +1269,10 @@ def _measure_public_page_request(self, client: Any, payload: Mapping[str, Any], "Observed results from completed source-bound operations in this episode:\n" + _json(context)}) schema = next_payload.get("decision_schema", {}) if messages: + provider_kind = ((getattr(client, "provider", {}) or {}).get("kind") + or getattr(client, "api", "")) messages[-1]["content"] = str(messages[-1].get("content", "")) + \ - "\n\nReturn exactly one JSON object satisfying this decision contract:\n" + _json(schema) + _decision_schema_note(schema, provider_kind=str(provider_kind)) from research.agents.policy import ensure_research_messages messages = ensure_research_messages(messages, client.config, str(next_payload.get("output_stage", "experiment"))) messages, _ = _intern_delivered_messages(messages) @@ -1226,7 +1376,9 @@ def _refit_latest_context_page(self, payload: Mapping[str, Any], history: list[d "delivery_status": fitted.get("delivery_status", "delivered"), "response_capacity": capacity, "response_capacity_sources": copy.deepcopy(getattr(self, "_context_capacity_sources", [])), - "result": fitted}, str(payload.get("work_id", ""))) + "result_reference": {"segment_id": fitted["segment_id"], "pointer": fitted["pointer"], + "offset": fitted["offset"], "next_offset": fitted["next_offset"]}}, + str(payload.get("work_id", ""))) return history def _attach_public_context(self, source_payload: Mapping[str, Any], prepared: Mapping[str, Any], *, @@ -1278,7 +1430,8 @@ def _attach_public_context(self, source_payload: Mapping[str, Any], prepared: Ma (self.campaign_id, str(payload.get("work_id", "")))).fetchone() if row: try: - stored_turns = json.loads(row[0]).get("action_history", []) + stored_turns = _expand_action_history_frontiers( + json.loads(row[0]).get("action_history", [])) public_turns = [] for turn in stored_turns: if not isinstance(turn, Mapping): @@ -1586,7 +1739,10 @@ def episode_resource_usage(self, item: Mapping[str, Any]) -> dict[str, float]: row = db.execute("SELECT state_json FROM forest_study_aide_sessions WHERE campaign_id=? AND work_id=?", (self.campaign_id, work_id)).fetchone() if row: - native_active_seconds = float(json.loads(row[0]).get("local_active_seconds", 0.0)) + native_state = json.loads(row[0]) + native_state["action_history"] = _expand_action_history_frontiers( + native_state.get("action_history", [])) + native_active_seconds = float(native_state.get("local_active_seconds", 0.0)) in_progress = max(0.0, time.monotonic() - self._active_episode_started[work_id]) if work_id in self._active_episode_started else 0.0 active_before_compute = max(model_seconds + float(cpu_actions or 0.0), native_active_seconds + in_progress) @@ -1786,15 +1942,17 @@ def execute_aide_node(self, item: Mapping[str, Any], *, node_id: str, code: str, "action_work_ids": [], "pending_work_ids": [], "verified_metrics": []} decision = program.get("decision") def reject_decision(reason: str, proposed: Any = None) -> dict[str, Any]: - frontier = _episode_source_frontier(payload, self._aide_action_history(str(item["work_id"]))) + prior_history = self._aide_action_history(str(item["work_id"])) + frontier = _episode_source_frontier(payload, prior_history) + frontier_turns = _matching_frontier_turns(prior_history, frontier) signature = _semantic_action_proposal(proposed) prior = [] - for turn in self._aide_action_history(str(item["work_id"])): + for turn_index, turn in enumerate(prior_history): execution = turn.get("execution") if isinstance(turn, Mapping) else None if not isinstance(execution, Mapping) or execution.get("exc_type") != "DecisionValidationError": continue if execution.get("validation_reason") == reason and execution.get("semantic_proposal") == signature \ - and _json(turn.get("source_frontier")) == _json(frontier): + and turn_index in frontier_turns: prior.append(execution) diagnosed = bool(prior) observation = {"same_source_bound_results": True, "new_evidence_added": False, @@ -1831,9 +1989,10 @@ def reject_decision(reason: str, proposed: Any = None) -> dict[str, Any]: if str(payload.get("arm", "")).upper() != "FULL_NO_GATE" and validation.get("valid") is not True: frontier = _episode_source_frontier(payload, self._aide_action_history(str(item["work_id"]))) semantic_submission = _semantic_submission(submission) - matching = [turn for turn in history + frontier_turns = _matching_frontier_turns(history, frontier) + matching = [turn for turn_index, turn in enumerate(history) if _semantic_submission(turn.get("submission", {})) == semantic_submission - and _json(turn.get("source_frontier")) == _json(frontier) + and turn_index in frontier_turns and isinstance(turn.get("submission_validation"), Mapping) and turn["submission_validation"].get("valid") is False and _json(turn["submission_validation"].get("source_validation_signature") @@ -1886,11 +2045,12 @@ def reject_decision(reason: str, proposed: Any = None) -> dict[str, Any]: prior is not None for prior in duplicates): frontier = _episode_source_frontier(payload, self._aide_action_history(str(item["work_id"]))) identities = sorted(str(row.get("action_work_id")) for row in duplicates) + prior_history = self._aide_action_history(str(item["work_id"])) + frontier_turns = _matching_frontier_turns(prior_history, frontier) same_source_diagnostic = any( isinstance(turn.get("no_progress_observation"), Mapping) and turn["no_progress_observation"].get("action_work_ids") == identities - and _json(turn.get("source_frontier")) == _json(frontier) - for turn in self._aide_action_history(str(item["work_id"]))) + for turn_index, turn in enumerate(prior_history) if turn_index in frontier_turns) feedback = "The requested registered action IDs already completed with the same saved source-bound results. This program added no new evidence. Choose a different available operation, use the existing evidence to submit, or state what new source-bound evidence is needed." observation = {"action_work_ids": identities, "same_source_bound_results": True, "new_evidence_added": False, "diagnostic_previously_delivered": same_source_diagnostic} @@ -1968,11 +2128,14 @@ def _record_active_interval(self, work_id: str, seconds: float, source: str) -> def _aide_action_history(self, work_id: str) -> list[dict[str, Any]]: with self.ledger.connect() as db: + table = db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='forest_study_aide_sessions'").fetchone() + if not table: + return [] row = db.execute("SELECT state_json FROM forest_study_aide_sessions WHERE campaign_id=? AND work_id=?", (self.campaign_id, work_id)).fetchone() if not row: return [] - return list(json.loads(row[0]).get("action_history", [])) + return _expand_action_history_frontiers(json.loads(row[0]).get("action_history", [])) def _context_read_no_progress(self, work_id: str, payload: Mapping[str, Any], actions: Sequence[Mapping[str, Any]]) -> dict[str, Any] | None: @@ -2225,7 +2388,7 @@ def execute_model_episode(self, item: Mapping[str, Any], *, client: Any, """ client = self._budgeted_episode_client(item, client) payload = dict(item["payload"]) - prior_result = item.get("result_json") + prior_result = item.get("result") if isinstance(prior_result, str): try: prior_result = json.loads(prior_result) @@ -2234,6 +2397,29 @@ def execute_model_episode(self, item: Mapping[str, Any], *, client: Any, elif not isinstance(prior_result, Mapping): prior_result = {} validation_feedback = prior_result.get("response_validation_feedback") + if not validation_feedback: + prior_diagnostic = next((turn.get("no_progress_observation") + for turn in reversed(history) if isinstance(turn, Mapping) + and isinstance(turn.get("no_progress_observation"), Mapping)), None) + if prior_diagnostic: + action_ids = prior_diagnostic.get("action_work_ids") or [] + if action_ids and prior_diagnostic.get("same_source_bound_results"): + instruction = ("The registered action IDs listed here already completed with the same saved results " + "under unchanged evidence. Do not execute those same IDs again unless their source evidence " + "changes. Use the saved results to submit or analyze the evidence, or choose a different " + "registered action that can add new evidence.") + elif prior_diagnostic.get("context_page_requests"): + instruction = ("The previous context-page reads returned no new source characters. Use the source " + "already retrieved, or choose a different registered page/offset that can add new evidence.") + else: + instruction = ("The previous decision produced no new evidence. Replan from the saved action " + "results and current evidence; choose an action that advances the task or submit a conclusion " + "supported by the existing evidence.") + validation_feedback = {"reason": "NO_PROGRESS_REPLAN", + "action_work_ids": list(action_ids), + "same_source_bound_results": prior_diagnostic.get("same_source_bound_results") is True, + "new_evidence_added": prior_diagnostic.get("new_evidence_added") is True, + "observation": dict(prior_diagnostic), "instruction": instruction} saved_prepared = prior_result.get("prepared_episode_payload") if (validation_feedback and isinstance(saved_prepared, Mapping) and not prior_result.get("no_progress_observation")): @@ -2265,7 +2451,9 @@ def execute_model_episode(self, item: Mapping[str, Any], *, client: Any, "Use the registered context and current feedback to make a new decision. Validation feedback: " + _json(payload["response_validation_feedback"])}) if delivered_messages: - schema_note = "\n\nReturn exactly one JSON object satisfying this decision contract:\n" + _json(payload["decision_schema"]) + provider_kind = ((getattr(client, "provider", {}) or {}).get("kind") + or getattr(client, "api", "")) + schema_note = _decision_schema_note(payload["decision_schema"], provider_kind=str(provider_kind)) delivered_messages[-1]["content"] = str(delivered_messages[-1].get("content", "")) + schema_note delivered_messages = ensure_research_messages(delivered_messages, client.config, output_stage) source_messages = copy.deepcopy(delivered_messages) @@ -2383,7 +2571,7 @@ def execute_model_episode(self, item: Mapping[str, Any], *, client: Any, same_rejection = (isinstance(prior_rejection, Mapping) and prior_rejection.get("proposal") == proposal and prior_rejection.get("reason") == str(validation_error) - and _json(prior_rejection.get("source_frontier")) == _json(frontier)) + and prior_rejection.get("source_frontier") == frontier) diagnosed = bool(same_rejection and prior_rejection.get("diagnostic_delivered")) feedback = {"error_type": type(validation_error).__name__, "reason": str(validation_error), "request_id": request_id, "raw_response_persisted": True} @@ -2395,11 +2583,11 @@ def execute_model_episode(self, item: Mapping[str, Any], *, client: Any, with self.ledger.connect() as db: db.execute("UPDATE forest_study_work SET state=?,result_json=?,decision_json=NULL,updated_at=? WHERE campaign_id=? AND work_id=?", ("paused_no_progress" if diagnosed else "retryable", - _json({"request_id": request_id, + _json(_compact_study_result({"request_id": request_id, "response_validation_feedback": None if diagnosed else feedback, "validation_rejection": rejection, "no_progress_observation": no_progress if diagnosed else None, - "prepared_episode_payload": payload, "action_history": list(history)}), + "prepared_episode_payload": payload, "action_history": list(history)})), finished_at, self.campaign_id, str(item["work_id"]))) return {"request_id": request_id, "decision": {}, "usage": usage, "model": result.get("model"), "response_id": result.get("response_id"), @@ -2416,8 +2604,8 @@ def execute_model_episode(self, item: Mapping[str, Any], *, client: Any, db.execute("UPDATE forest_study_model_calls SET response_json=?,usage_json=?,status='completed',finished_at=?,elapsed_seconds=? WHERE request_id=?", (_json(response_record), _json(usage), finished_at, elapsed_seconds, request_id)) db.execute("UPDATE forest_study_work SET state='running',result_json=?,decision_json=?,updated_at=? WHERE campaign_id=? AND work_id=?", - (_json({"request_id": request_id, "decision_received": True, "action_history": list(history), - "prepared_episode_payload": payload}), + (_json(_compact_study_result({"request_id": request_id, "decision_received": True, + "action_history": list(history), "prepared_episode_payload": payload})), _json(decision), finished_at, self.campaign_id, str(item["work_id"]))) if decision.get("mode") == "submit": submission = decision.get("submission") if isinstance(decision.get("submission"), Mapping) else {} @@ -2429,9 +2617,10 @@ def execute_model_episode(self, item: Mapping[str, Any], *, client: Any, if payload.get("arm") != "FULL_NO_GATE" and action_result.get("valid") is not True: frontier = _episode_source_frontier(payload, history) semantic_submission = _semantic_submission(submission) - repeated_prior = [turn for turn in history + frontier_turns = _matching_frontier_turns(history, frontier) + repeated_prior = [turn for turn_index, turn in enumerate(history) if _semantic_submission(turn.get("submission", {})) == semantic_submission - and _json(turn.get("source_frontier")) == _json(frontier) + and turn_index in frontier_turns and isinstance(turn.get("submission_validation"), Mapping) and turn["submission_validation"].get("valid") is False and _json(turn["submission_validation"].get("source_validation_signature") @@ -2441,7 +2630,6 @@ def execute_model_episode(self, item: Mapping[str, Any], *, client: Any, already_diagnosed = any( isinstance(turn.get("no_progress_observation"), Mapping) and _semantic_submission(turn["no_progress_observation"].get("repeated_submission", {})) == semantic_submission - and _json(turn.get("source_frontier")) == _json(frontier) for turn in repeated_prior) feedback = ("The same final proposal received the same source-validation result against unchanged evidence. " "Revise the submission using the validation feedback, execute a different available action, " @@ -2484,11 +2672,11 @@ def execute_model_episode(self, item: Mapping[str, Any], *, client: Any, if current_rows and repeated and len(repeated) == len(current_rows): identities = sorted(str(row.get("action_work_id")) for row in repeated) frontier = _episode_source_frontier(payload, history) + frontier_turns = _matching_frontier_turns(history, frontier) already_diagnosed = any( isinstance(turn.get("no_progress_observation"), Mapping) and turn["no_progress_observation"].get("action_work_ids") == identities - and _json(turn.get("source_frontier")) == _json(frontier) - for turn in history) + for turn_index, turn in enumerate(history) if turn_index in frontier_turns) feedback = ("These registered actions already completed with the same source-bound results: " + ", ".join(identities) + ". Repeating them produced no new evidence. Choose a different currently available action, " @@ -2659,7 +2847,9 @@ def execute_decision_actions(self, episode: Mapping[str, Any], actions: Sequence "requested_limit": page.get("requested_limit"), "returned_characters": page.get("returned_characters"), "complete": page.get("complete"), "capacity_limited": page.get("capacity_limited"), "response_capacity": page.get("response_capacity"), - "response_capacity_sources": list(context_response_capacity_sources), "result": page}, + "response_capacity_sources": list(context_response_capacity_sources), + "result_reference": {"segment_id": page.get("segment_id"), "pointer": page.get("pointer"), + "offset": page.get("offset"), "next_offset": page.get("next_offset")}}, str(episode.get("work_id", ""))) selected = (submission or {}).get("selected_configuration_id") selection_results = [r for r in results if r["kind"] == "analysis" and r.get("result", {}).get("operation") == "select"] @@ -2776,11 +2966,33 @@ def without_timeout(value): value["inference_spec"] = spec return value if without_timeout(saved) != without_timeout(binding): - raise ValueError(f"Action work ID {work_id} was reused with different source-bound conditions") + differing_fields = sorted(key for key in set(saved) | set(binding) + if key != "fit_timeout_seconds" + and (saved.get(key) != binding.get(key))) + feedback = { + "reason": "source_bound_action_id_conflict", + "action_work_id": work_id, + "registered_state": prior["state"], + "differing_fields": differing_fields, + "saved_result_available": prior["result_json"] is not None, + "instruction": ( + "This action work ID is already registered with different source-bound conditions. " + "It was not overwritten or executed. Inspect and use its saved result if it satisfies " + "the current research contract; otherwise choose a different predeclared action ID " + "whose conditions match the intended work. Do not repeat this ID." + ), + } + return {"action_work_id": work_id, "kind": binding["kind"], + "state": "retryable", "reason": feedback["reason"], + "request_new_decision": True, + "response_validation_feedback": feedback, + "result": {"status": "SOURCE_BOUND_ACTION_ID_CONFLICT", + "work_id": work_id, "registered_state": prior["state"], + "differing_fields": differing_fields, "execution_started": False}} binding = saved if prior["state"] == "completed": return {"action_work_id": work_id, "kind": binding["kind"], "state": "completed", - "result": json.loads(prior["result_json"] or "{}")} + "result": _decode_study_result_json(prior["result_json"])} else: budget = self.episode_budget_status(episode) if binding["kind"] in {"fit", "infer"}: @@ -2803,7 +3015,7 @@ def without_timeout(value): from .study_prerequisites import inspect_analysis_prerequisites with self.ledger.connect() as db: def source_work(source_id): - row = db.execute("SELECT * FROM forest_study_work WHERE campaign_id=? AND work_id=?", + row = db.execute("SELECT kind,state,supervisor_work_id FROM forest_study_work WHERE campaign_id=? AND work_id=?", (self.campaign_id, source_id)).fetchone() return dict(row) if row else None compute_exists = db.execute("SELECT 1 FROM sqlite_master WHERE name='forest_compute_work'").fetchone() @@ -2833,13 +3045,13 @@ def supervisor_work(source_id): "response_validation_feedback": feedback, "result": prerequisite} if prior and prior["state"] in {"failed", "cancelled", "budget_exhausted"}: return {"action_work_id": work_id, "kind": binding["kind"], "state": prior["state"], - "result": json.loads(prior["result_json"] or "{}")} + "result": _decode_study_result_json(prior["result_json"])} if binding.get("analysis_operation") == "compare" and binding.get("selection_work_id"): with self.ledger.connect() as db: selection = db.execute("SELECT state,result_json FROM forest_study_work WHERE campaign_id=? AND work_id=?", (self.campaign_id, str(binding["selection_work_id"]))).fetchone() target = (binding.get("compared_configuration_ids") or [None, None])[-1] - selected = json.loads(selection["result_json"] or "{}") if selection and selection["state"] == "completed" else {} + selected = _decode_study_result_json(selection["result_json"]) if selection and selection["state"] == "completed" else {} def action_selectors(entry): return {"action_work_id": entry["work_id"], "kind": entry.get("action_name", entry.get("analysis_operation", action)), @@ -3376,6 +3588,7 @@ def run_claimed(self, item: Mapping[str, Any], *, owner: str | None = None, ((nested_episode or {}).get("arm") if isinstance(nested_episode, Mapping) else ""))).upper() if payload_arm == "AIDE": from .aide_runtime import run_aide_episode + previous_aide_history = self._aide_action_history(str(item["work_id"])) if _is_replay_episode(item) or _is_replay_episode(raw_episode_payload): client = self._budgeted_episode_client(item, client) client = self._replay_request_client(item, client) @@ -3389,10 +3602,11 @@ def run_claimed(self, item: Mapping[str, Any], *, owner: str | None = None, if state not in {"completed", "failed", "retryable", "awaiting_reconciliation", "budget_exhausted", "paused_no_progress"}: raise ValueError(f"Native AIDE returned unknown episode state: {state}") self.ledger.set_state(self.campaign_id, item["work_id"], state, result=outcome, + event_result=_result_event_delta(outcome, previous_aide_history), error=outcome.get("reason") if state in {"failed", "budget_exhausted"} else None) return {"work_id": item["work_id"], "kind": item["kind"], **outcome} prior_decision = json.loads(item["decision_json"]) if item.get("decision_json") else None - prior_result = json.loads(item["result_json"]) if item.get("result_json") else {} + prior_result = item.get("result") or _decode_study_result_json(item.get("result_json")) history = list(prior_result.get("action_history", [])) previous_action_result = prior_result.get("action_result") or {} nested_replan = any(isinstance(action, Mapping) and action.get("request_new_decision") is True @@ -3486,7 +3700,8 @@ def run_claimed(self, item: Mapping[str, Any], *, owner: str | None = None, outcome["no_progress_observation"] = action_result["no_progress_observation"] outcome["state"] = final_state self.ledger.set_state(self.campaign_id, item["work_id"], final_state, - result=outcome, decision=decision if decision else None, + result=outcome, event_result=_result_event_delta(outcome, history), + decision=decision if decision else None, error=(action_result.get("reason") if final_state == "failed" else None)) return {"work_id": item["work_id"], "kind": item["kind"], **outcome} raise NotImplementedError(f"No factual executor is registered for {item['kind']}") @@ -3595,15 +3810,18 @@ def stable(value): def _semantic_submission(value: Mapping[str, Any]) -> dict[str, Any]: """Keep decision-bearing fields while ignoring prose that does not change a claim.""" prose = {"claim", "rationale", "explanation", "notes", "justification"} + unordered_ids = {"invalidate_record_ids", "reuse_record_ids"} if not isinstance(value, Mapping): return {} - return {str(key): copy.deepcopy(item) for key, item in value.items() if str(key) not in prose} + return {str(key): (sorted(copy.deepcopy(item), key=_json) if str(key) in unordered_ids + and isinstance(item, list) else copy.deepcopy(item)) + for key, item in value.items() if str(key) not in prose} def _validation_signature(value: Mapping[str, Any]) -> dict[str, Any]: volatile = {"state", "reason", "response_validation_feedback", "request_new_decision", "pause_reason", "no_progress_observation", "source_validation_reason", - "source_validation_signature", "observed_request_usage"} + "source_validation_signature", "observed_request_usage", "submitted_claim"} if not isinstance(value, Mapping): return {} return {str(key): copy.deepcopy(item) for key, item in value.items() if str(key) not in volatile} @@ -3646,8 +3864,11 @@ def _context_page_interval(action: Mapping[str, Any]) -> dict[str, Any] | None: or start < 0 or end <= start or returned != end - start or len(content) != returned or requested_limit < returned): return None - return {"segment_id": segment_id, "pointer": pointer, "start": start, "end": end, - "requested_limit": requested_limit} + source_key = result.get("source_key") + if source_key is not None and (not isinstance(source_key, str) or not source_key): + return None + return {"segment_id": segment_id, "source_key": source_key, "pointer": pointer, + "start": start, "end": end, "content": content, "requested_limit": requested_limit} def _repeated_context_page_ranges(actions: Sequence[Mapping[str, Any]], @@ -3655,9 +3876,10 @@ def _repeated_context_page_ranges(actions: Sequence[Mapping[str, Any]], frontier: Mapping[str, Any]) -> tuple[list[dict[str, Any]], bool]: """Return covered read ranges and whether this turn returned any new source characters.""" intervals: dict[tuple[str, str], list[tuple[int, int]]] = {} - frontier_json = _json(frontier) - for turn in history: - if not isinstance(turn, Mapping) or _json(turn.get("source_frontier")) != frontier_json: + exact_pages: list[tuple[str, str, int, int, str]] = [] + frontier_turns = _matching_frontier_turns(history, frontier) + for turn_index, turn in enumerate(history): + if not isinstance(turn, Mapping) or turn_index not in frontier_turns: continue for prior in turn.get("actions", []): if not isinstance(prior, Mapping) or not _is_context_read_action(prior): @@ -3666,6 +3888,9 @@ def _repeated_context_page_ranges(actions: Sequence[Mapping[str, Any]], if page is not None: key = (page["segment_id"], page["pointer"]) intervals.setdefault(key, []).append((page["start"], page["end"])) + if page["source_key"] is not None: + exact_pages.append((page["source_key"], page["pointer"], page["start"], + page["end"], page["content"])) repeated: list[dict[str, Any]] = [] new_characters = False @@ -3675,6 +3900,12 @@ def _repeated_context_page_ranges(actions: Sequence[Mapping[str, Any]], page = _context_page_interval(action) if page is None: continue + same_source_page = page["source_key"] is not None and any( + (page["source_key"], page["pointer"], page["start"], page["end"], page["content"]) == prior + for prior in exact_pages) + if same_source_page: + repeated.append(page) + continue key = (page["segment_id"], page["pointer"]) existing = sorted(intervals.get(key, [])) cursor = page["start"] @@ -3691,6 +3922,9 @@ def _repeated_context_page_ranges(actions: Sequence[Mapping[str, Any]], else: new_characters = True intervals.setdefault(key, []).append((page["start"], page["end"])) + if page["source_key"] is not None: + exact_pages.append((page["source_key"], page["pointer"], page["start"], + page["end"], page["content"])) return repeated, new_characters @@ -3709,10 +3943,16 @@ def _context_page_no_progress(actions: Sequence[Mapping[str, Any]], normalized_requests = [{"segment_id": segment, "pointer": pointer, "offset": offset, "requested_limit": limit} for segment, pointer, offset, limit in requests] + request_keys = sorted({(row["source_key"] or row["segment_id"], row["pointer"], + row["start"], row["requested_limit"]) for row in repeated}) + normalized_request_keys = [{"source": source, "pointer": pointer, + "offset": offset, "requested_limit": limit} + for source, pointer, offset, limit in request_keys] + frontier_turns = _matching_frontier_turns(history, frontier) already_diagnosed = any(isinstance(turn.get("no_progress_observation"), Mapping) - and turn["no_progress_observation"].get("context_page_requests") == normalized_requests - and _json(turn.get("source_frontier")) == _json(frontier) - for turn in history if isinstance(turn, Mapping)) + and (turn["no_progress_observation"].get("context_page_request_keys") == normalized_request_keys + or turn["no_progress_observation"].get("context_page_requests") == normalized_requests) + for index, turn in enumerate(history) if index in frontier_turns) return {"state": "paused_no_progress" if already_diagnosed else "retryable", "request_new_decision": not already_diagnosed, "reason": "repeated_context_page_without_new_source_characters", @@ -3724,6 +3964,7 @@ def _context_page_no_progress(actions: Sequence[Mapping[str, Any]], if already_diagnosed else None), "no_progress_observation": {"context_page_ranges": normalized, "context_page_requests": normalized_requests, + "context_page_request_keys": normalized_request_keys, "same_source_bound_results": True, "new_evidence_added": False, "diagnostic_previously_delivered": already_diagnosed}} @@ -4004,6 +4245,22 @@ def _verify_required_controller_restart(runtime: Any, episode: Mapping[str, Any] previous_identity = previous_exit.get("identity") if isinstance(previous_exit, Mapping) else None if isinstance(previous_exit, Mapping): completion = completion or previous_exit.get("completion_receipt") + if isinstance(completion, str): + # ManagedProcesses publishes the terminal receipt as a path. Resolve + # it only through the persisted source dispatch record, then read the + # exact completion.json belonging to that original worker. + dispatch_id = str(row["request_dispatch_id"]) + with runtime.ledger.connect() as db: + process = db.execute("SELECT folder FROM forest_study_controller_processes " + "WHERE campaign_id=? AND work_id=? AND dispatch_id=?", + (runtime.campaign_id, str(episode.get("work_id", "")), dispatch_id)).fetchone() + if process is not None: + expected_completion = Path(str(process["folder"])) / "processes" / dispatch_id / "completion.json" + try: + if Path(completion).resolve(strict=True) == expected_completion.resolve(strict=True): + completion = _read_json(expected_completion) + except (OSError, RuntimeError, ValueError, TypeError): + completion = None previous_identity = previous_identity or previous_exit def process_key(value): if not isinstance(value, Mapping): @@ -4050,6 +4307,12 @@ def same_worker(value, pid, created): and isinstance(previous_exit, Mapping) and str(previous_exit.get("status", "")).lower() == "completed" and isinstance(completion, Mapping) and completion.get("exit_code") == 0 + and str(completion.get("status", "")).lower() == "completed" + and str(completion.get("work_id", "")) == str(row["request_dispatch_id"]) + and completion.get("launch_id") == previous_exit.get("launch_id") + and completion.get("configuration_version") == previous_exit.get("configuration_version") + and completion.get("host_pid") == previous_exit.get("host_pid") + and completion.get("host_created") == previous_exit.get("host_created") and isinstance(previous_identity, Mapping) and same_worker(previous_identity, row["request_pid"], row["request_process_created"]) and same_worker(previous_worker, row["request_pid"], row["request_process_created"]) diff --git a/research/experiments/study_scheduler.py b/research/experiments/study_scheduler.py index 796127e..106a755 100644 --- a/research/experiments/study_scheduler.py +++ b/research/experiments/study_scheduler.py @@ -18,6 +18,7 @@ import argparse from concurrent.futures import ThreadPoolExecutor +from collections.abc import Mapping from copy import copy, deepcopy import json import importlib @@ -31,7 +32,9 @@ import uuid from research.agents.processes import ManagedProcesses, artifact_matches, atomic_json, is_alive -from .study_runtime import BLOCKING_RELATIONS, SETTLEMENT_RELATIONS, StudyLedger, StudyRuntime, load_saved_codex_client +from .study_runtime import (BLOCKING_RELATIONS, SETTLEMENT_RELATIONS, StudyLedger, StudyRuntime, + _decode_study_result_json, _expand_action_history_frontiers, _manifest_work_items, + load_saved_codex_client) MODEL_KINDS = ('episode', 'replay', 'calibration_decision') @@ -63,6 +66,9 @@ span_id TEXT PRIMARY KEY, campaign_id TEXT NOT NULL, work_id TEXT NOT NULL, lane TEXT NOT NULL, ready_at REAL NOT NULL, claimed_at REAL); CREATE INDEX IF NOT EXISTS forest_study_dispatch_work ON forest_study_dispatches(work_id,state); +CREATE INDEX IF NOT EXISTS forest_study_dispatch_live_session + ON forest_study_dispatches(campaign_id,state,session_id) + WHERE state IN ('claiming','accepted','running'); CREATE TABLE IF NOT EXISTS forest_study_controller_processes ( dispatch_id TEXT PRIMARY KEY, campaign_id TEXT NOT NULL, work_id TEXT NOT NULL, folder TEXT NOT NULL, state TEXT NOT NULL, worker_pid INTEGER, worker_created REAL, @@ -110,6 +116,9 @@ def _controller_checkpoint(runtime, item): (runtime.campaign_id, item['work_id'])).fetchone() if value: saved[table] = json.loads(value[0]) + if table == 'forest_study_aide_sessions': + saved[table]['action_history'] = _expand_action_history_frontiers( + saved[table].get('action_history', [])) return {'work': work, 'children': children, 'model_calls': calls, 'ordinary_state': saved} @@ -117,7 +126,7 @@ def _checkpoint_reload_proof(checkpoint): """Actually reconstruct native Journal/tree/RNG without requesting a model.""" native = checkpoint['ordinary_state'].get('forest_study_aide_sessions') if native is None: - return {'arm': 'FULL', 'action_history': json.loads(checkpoint['work']['result_json'] or '{}').get('action_history', []), + return {'arm': 'FULL', 'action_history': _decode_study_result_json(checkpoint['work']['result_json']).get('action_history', []), 'model_request_ids': [value['request_id'] for value in checkpoint['model_calls']]} from .aide_runtime import load_upstream, _tuple_tree, _node_record from omegaconf import OmegaConf @@ -238,7 +247,7 @@ def _request_controller_restart(runtime, item, dispatch, worker): native = checkpoint['ordinary_state'].get('forest_study_aide_sessions') if native and native.get('pending') is not None: return None - history = native.get('action_history', []) if native else json.loads(checkpoint['work']['result_json'] or '{}').get('action_history', []) + history = native.get('action_history', []) if native else _decode_study_result_json(checkpoint['work']['result_json']).get('action_history', []) if (comparison['selection_work_id'] not in _history_work_ids(history) or not any(value['status'] == 'completed' for value in checkpoint['model_calls']) or any(value['status'] == 'requested' for value in checkpoint['model_calls'])): @@ -427,13 +436,38 @@ def __init__(self, runtime, *, scheduler_id=None, maximum_model_concurrency=2, self._tested_degrees = set() self._pending_decisions = [] self._reconciliation_seen = {} + self._parent_reconciliation_cache = {} + self._provider_manifest_identity = object() + self._provider_profiles_by_work = {} + self._aide_session_index_ready = False + self._refresh_provider_profiles() self._executors = {'model': ThreadPoolExecutor(max_workers=self.maximum, thread_name_prefix='forest-model'), 'gpu': ThreadPoolExecutor(max_workers=1, thread_name_prefix='forest-gpu'), 'cpu': ThreadPoolExecutor(max_workers=self.cpu_degree, thread_name_prefix='forest-audit')} with self.ledger.connect() as db: db.executescript(SCHEMA) + compute_exists = db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='forest_compute_work'").fetchone() + if compute_exists: + db.execute("CREATE INDEX IF NOT EXISTS forest_compute_work_live_state " + "ON forest_compute_work(state,work_id) " + "WHERE state NOT IN ('completed','failed','cancelled','budget_exhausted')") self._read_pilot() + def _refresh_provider_profiles(self): + """Index per-work provider metadata from the already loaded manifest.""" + identity = getattr(self.runtime, '_manifest_identity', None) + if identity == self._provider_manifest_identity: + return + profiles = {} + default_profile = self.runtime.manifest.get('model_provider') + for item in _manifest_work_items(self.runtime.manifest): + work_id = str(item.get('work_id') or item.get('cell_id') or item.get('id') or '') + profile = item.get('model_provider') or default_profile + if work_id and isinstance(profile, Mapping): + profiles[work_id] = dict(profile) + self._provider_profiles_by_work = profiles + self._provider_manifest_identity = identity + def _configuration(self): return {'version': 1, 'maximum_model_concurrency': self.maximum, 'model_concurrency': self.degree, 'cpu_concurrency': self.cpu_degree, @@ -486,14 +520,36 @@ def acquire(self): self.scheduler_id = previous['scheduler_id'] saved = json.loads(previous['configuration_json']) if saved.get('maximum_model_concurrency') == self.maximum and saved.get('model_profile') == self.runtime.manifest.get('model_provider', {}) and saved.get('pilot_path') == (str(self.pilot_path) if self.pilot_path else None): - self.degree = min(self.maximum, saved.get('model_concurrency', self.degree)) - self._observed_samples = saved.get('observed_samples', self._observed_samples) + saved_degree = min(self.maximum, saved.get('model_concurrency', self.degree)) + # Preserve both the measured pilot baseline and successful + # live samples across coordinator recovery. Replacing the + # pilot samples with a stale, downrated saved degree could + # strand all ready work behind an unrelated unresolved + # request even when the pilot proved another degree faster. + samples = {value['degree']: value for value in self._observed_samples + if isinstance(value, Mapping) and isinstance(value.get('degree'), int)} + for value in saved.get('observed_samples', []): + if not isinstance(value, Mapping) or not isinstance(value.get('degree'), int): + continue + current = samples.get(value['degree']) + rank = lambda sample: (sample.get('calls_per_second', 0), + -sample.get('p95_seconds', float('inf')), -sample['degree']) + if current is None or rank(value) > rank(current): + samples[value['degree']] = dict(value) + self._observed_samples = list(samples.values()) self._tested_degrees.update(saved.get('tested_degrees', [])) + if saved_degree in self._tested_degrees and self._observed_samples: + self.degree = max(self._observed_samples, + key=lambda value: (value.get('calls_per_second', 0), + -value.get('p95_seconds', float('inf')), -value['degree']))['degree'] + else: + self.degree = saved_degree self._wave = saved.get('wave') db.execute('INSERT INTO forest_study_schedulers VALUES(?,?,?,?,?,?,?,?,?,?) ON CONFLICT(campaign_id) DO UPDATE SET session_id=excluded.session_id,hostname=excluded.hostname,pid=excluded.pid,process_created=excluded.process_created,state=excluded.state,configuration_json=excluded.configuration_json,updated_at=excluded.updated_at', (self.campaign_id, self.scheduler_id, self.session_id, socket.gethostname(), os.getpid(), created, 'running', _json(self._configuration()), now, now)) self._owned = True + self._recover_provider_configuration_circuits() self._recover_dispatches() for value in self._pending_decisions: self._decision(value) @@ -540,6 +596,28 @@ def _recover_dispatches(self): for dispatch,item,lane in observers: future = self._executors[lane].submit(self._execute, dispatch, item, lane) self._futures[future] = {'item': item, 'lane': lane, 'dispatch_id': dispatch} + # Client construction can fail after the durable claim but before + # run_claimed changes the work state to running (for example, a + # temporarily unavailable saved-provider database). Such dispatches + # are terminal while their work lease remains claimed. Requeue only + # when the same dispatch has no model-call receipt at all; any recorded + # request, completed or ambiguous, remains untouched for reconciliation. + with self.ledger.connect() as db: + orphaned = db.execute("SELECT w.work_id,w.lease_owner,d.dispatch_id,d.accepted_at FROM forest_study_work w JOIN forest_study_dispatches d ON d.campaign_id=w.campaign_id AND d.owner=w.lease_owner WHERE w.campaign_id=? AND w.state='claimed' AND d.session_id<>? AND d.lane='model' AND d.state='failed' AND d.finished_at IS NOT NULL AND w.kind IN ('episode','replay','calibration_decision')", + (self.campaign_id, self.session_id)).fetchall() + recovered_claims = [] + for work in orphaned: + call = db.execute('SELECT 1 FROM forest_study_model_calls WHERE campaign_id=? AND work_id=? AND started_at>=? LIMIT 1', + (self.campaign_id, work['work_id'], work['accepted_at'])).fetchone() + if call: + continue + changed = db.execute("UPDATE forest_study_work SET state='retryable',lease_owner=NULL,lease_at=NULL,error=NULL,updated_at=? WHERE campaign_id=? AND work_id=? AND state='claimed' AND lease_owner=?", + (time.time(), self.campaign_id, work['work_id'], work['lease_owner'])).rowcount + if changed: + recovered_claims.append((work['work_id'], work['dispatch_id'])) + for work_id, dispatch_id in recovered_claims: + self.ledger.event(self.campaign_id, 'unstarted_claim_recovered', + {'dispatch_id': dispatch_id, 'reason': 'failed before any model request was recorded'}, work_id) def _interval(self, kind, start, end, *, work_id=None, reference=None): if end <= start: @@ -561,6 +639,75 @@ def _source_blocked(self, item): rows = db.execute('SELECT state FROM forest_study_scheduler_circuits WHERE campaign_id=? AND source_id IN (' + ','.join('?' for _ in sources) + ')', [self.campaign_id, *sources]).fetchall() return any(row['state'] == 'blocked' for row in rows) + def _recover_provider_configuration_circuits(self): + """Release only work circuits made stale by a repaired provider DB path. + + Earlier provider-table lookup failures happened before a model request + was submitted. Keep their original last_error and counter, but once the + exact saved local provider validates again, let its registered work be + claimed. Ambiguous or completed model-call receipts are not affected. + """ + with self.ledger.connect() as db: + rows = db.execute("SELECT source_id,last_error,consecutive_errors FROM forest_study_scheduler_circuits " + "WHERE campaign_id=? AND state='blocked' AND instr(last_error,'no such table: providers')>0", + (self.campaign_id,)).fetchall() + if not rows: + return [] + + verified: dict[str, bool] = {} + recoverable = [] + for row in rows: + try: + source = json.loads(row['source_id']) + except (TypeError, ValueError): + continue + work_id = source.get('work_id') if isinstance(source, Mapping) else None + profile = self._provider_profiles_by_work.get(str(work_id)) if work_id else None + if not isinstance(profile, Mapping): + continue + provider_id = str(profile.get('id') or '') + if not provider_id: + continue + if provider_id not in verified: + try: + load_saved_codex_client(provider_id, model=str(profile.get('model') or ''), + reasoning_effort=str((profile.get('config') or {}).get('reasoning_effort', 'medium'))) + except Exception: + verified[provider_id] = False + else: + verified[provider_id] = True + if verified[provider_id]: + # Do not release a circuit while the same logical work still + # owns an unsettled provider request. Its outcome must be + # reconciled before a new invocation can become eligible. + with self.ledger.connect() as db: + unsettled = db.execute("SELECT 1 FROM forest_study_model_calls " + "WHERE campaign_id=? AND work_id=? AND status='requested' LIMIT 1", + (self.campaign_id, str(work_id))).fetchone() + if unsettled: + continue + recoverable.append((row['source_id'], str(work_id), int(row['consecutive_errors']))) + + if not recoverable: + return [] + now = time.time() + with self.ledger.connect() as db: + db.execute('BEGIN IMMEDIATE') + changed = [] + for source_id, work_id, prior_errors in recoverable: + if db.execute("UPDATE forest_study_scheduler_circuits SET state='ready' " + "WHERE campaign_id=? AND source_id=? AND state='blocked' " + "AND instr(last_error,'no such table: providers')>0", + (self.campaign_id, source_id)).rowcount: + changed.append({'work_id': work_id, 'preserved_error_count': prior_errors}) + if changed: + db.execute('INSERT INTO forest_study_events VALUES(?,?,?,?,?,?)', + (str(uuid.uuid4()), self.campaign_id, None, 'provider_configuration_revalidated', + _json({'provider_ids': sorted(verified), 'released_work_circuits': changed, + 'prior_errors_preserved': True, + 'verification': 'saved local provider kind/model/reasoning validated; no request sent'}), now)) + return [value['work_id'] for value in changed] + def _unsettled_model_owners(self): """A dead coordinator cannot release an unconfirmed external model slot.""" active = {entry['item']['work_id'] for entry in self._futures.values() if entry['lane'] == 'model'} @@ -595,13 +742,27 @@ def _record_outcome(self, item, error, *, failure=None): db.execute('INSERT INTO forest_study_scheduler_circuits VALUES(?,?,?,?,?) ON CONFLICT(campaign_id,source_id) DO UPDATE SET last_error=excluded.last_error,consecutive_errors=excluded.consecutive_errors,state=excluded.state', (self.campaign_id, source, error, count, state)) - def _accepted_compute(self): + def _accepted_compute(self, work_ids=None): from .gpu_supervisor import ComputeSupervisor with self.ledger.connect() as db: exists = db.execute("SELECT 1 FROM sqlite_master WHERE name='forest_compute_work'").fetchone() if not exists: return {} - rows = db.execute("SELECT work_id,state,receipt_json,json_extract(configuration_json,'$.gpu_devices') AS gpu_devices_json FROM forest_compute_work").fetchall() + if work_ids is None: + rows = db.execute("SELECT work_id,state,receipt_json,json_extract(configuration_json,'$.gpu_devices') AS gpu_devices_json FROM forest_compute_work").fetchall() + else: + identifiers = {str(value) for value in work_ids if value} + # Keep every live physical job visible, even if its owner is + # outside this campaign; then hydrate receipts only for those + # owners and the currently unresolved study work. + identifiers.update(row[0] for row in db.execute( + "SELECT work_id FROM forest_compute_work WHERE state NOT IN ('completed','failed','cancelled','budget_exhausted')")) + ordered = sorted(identifiers) + rows = [] + for offset in range(0, len(ordered), 500): + chunk = ordered[offset:offset + 500] + rows.extend(db.execute("SELECT work_id,state,receipt_json,json_extract(configuration_json,'$.gpu_devices') AS gpu_devices_json " + "FROM forest_compute_work WHERE work_id IN (" + ','.join('?' for _ in chunk) + ')', chunk).fetchall()) supervisor = ComputeSupervisor(self.runtime.ledger_path, self.runtime.ledger_id) result = {} for row in rows: @@ -614,9 +775,7 @@ def _accepted_compute(self): except ValueError: # A different logical owner still shares physical GPU usage; # unknown original work retains its capacity until reconciled. - with self.ledger.connect() as db: - original = dict(db.execute("SELECT state,json_extract(configuration_json,'$.gpu_devices') AS gpu_devices_json FROM forest_compute_work WHERE work_id=?", (row['work_id'],)).fetchone()) - result[row['work_id']] = {'status': original['state'], 'configuration': {'gpu_devices': json.loads(original['gpu_devices_json'] or '[]')}, 'unknown_owner': True} + result[row['work_id']] = {'status': row['state'], 'configuration': {'gpu_devices': json.loads(row['gpu_devices_json'] or '[]')}, 'unknown_owner': True} return result def _hard_allowance(self): @@ -650,10 +809,17 @@ def _reconcile_completed_gpu(self, compute): from .gpu_supervisor import ComputeSupervisor active = {entry['item']['work_id'] for entry in self._futures.values()} with self.ledger.connect() as db: - rows = [dict(row) for row in db.execute("SELECT * FROM forest_study_work WHERE campaign_id=? AND state='awaiting_reconciliation' AND kind IN ('fit','infer')", (self.campaign_id,))] + rows = [dict(row) for row in db.execute("SELECT work_id,kind,supervisor_work_id,payload_json FROM forest_study_work INDEXED BY forest_study_work_pending_compute_refs WHERE campaign_id=? AND state='awaiting_reconciliation' AND kind IN ('fit','infer')", (self.campaign_id,))] attempts = {row['logical_work_id']: dict(row) for row in db.execute( - 'SELECT * FROM forest_study_fit_attempts WHERE campaign_id=? ORDER BY attempt_number', (self.campaign_id,))} - accepted = {row['work_id']: row['ledger_id'] for row in db.execute('SELECT work_id,ledger_id FROM forest_compute_work')} if compute else {} + 'SELECT logical_work_id,supervisor_work_id,output_directory FROM forest_study_fit_attempts WHERE campaign_id=? ORDER BY attempt_number', (self.campaign_id,))} + physical_ids = sorted({row['supervisor_work_id'] or row['work_id'] for row in rows}) + accepted = {} + compute_table = db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='forest_compute_work'").fetchone() + if compute_table: + for offset in range(0, len(physical_ids), 500): + chunk = physical_ids[offset:offset + 500] + accepted.update({row['work_id']: row['ledger_id'] for row in db.execute( + 'SELECT work_id,ledger_id FROM forest_compute_work WHERE work_id IN (' + ','.join('?' for _ in chunk) + ')', chunk)}) recovered = [] for row in rows: logical = row['work_id'] @@ -753,60 +919,103 @@ def _wake_reconciled_parents(self): """Resume a durable pending decision only after all its children settle.""" active = {entry['item']['work_id'] for entry in self._futures.values()} with self.ledger.connect() as db: - db.execute('BEGIN IMMEDIATE') - rows = db.execute("SELECT * FROM forest_study_work WHERE campaign_id=? AND state='awaiting_reconciliation' AND kind IN ('episode','replay')", (self.campaign_id,)).fetchall() + rows = db.execute("SELECT work_id,decision_json,updated_at FROM forest_study_work INDEXED BY forest_study_work_pending_parent_state WHERE campaign_id=? AND state='awaiting_reconciliation' AND kind IN ('episode','replay')", (self.campaign_id,)).fetchall() + session_table = db.execute("SELECT 1 FROM sqlite_master WHERE name='forest_study_aide_sessions'").fetchone() + sessions = {} + if session_table: + if not self._aide_session_index_ready: + db.execute("CREATE INDEX IF NOT EXISTS forest_study_aide_session_live_updated " + "ON forest_study_aide_sessions(campaign_id,work_id,updated_at)") + self._aide_session_index_ready = True + sessions = {row['work_id']: row['updated_at'] for row in db.execute( + 'SELECT work_id,updated_at FROM forest_study_aide_sessions ' + 'INDEXED BY forest_study_aide_session_live_updated WHERE campaign_id=?', + (self.campaign_id,))} states = {row['work_id']: row['state'] for row in db.execute('SELECT work_id,state FROM forest_study_work WHERE campaign_id=?', (self.campaign_id,))} edges = db.execute("SELECT source_id,target_id FROM forest_study_edges WHERE campaign_id=? AND relation='derived_from'", (self.campaign_id,)).fetchall() awakened = [] for row in rows: if row['work_id'] in active or db.execute("SELECT 1 FROM forest_study_model_calls WHERE campaign_id=? AND work_id=? AND status='requested'", (self.campaign_id, row['work_id'])).fetchone(): continue - saved = json.loads(row['result_json'] or '{}') has_decision = bool(row['decision_json']) + session_updated = sessions.get(row['work_id']) + signature = (row['updated_at'], session_updated) + cached = self._parent_reconciliation_cache.get(row['work_id']) + if cached and cached['signature'] == signature: + saved, pending = cached['saved'], cached['pending'] + else: + # These reconciliation decisions use only two scalar fields. + # Episode result_json may contain the complete durable history + # (hundreds of MB in long campaigns); avoid materializing and + # decoding that snapshot just to recover its pending action. + raw = db.execute("SELECT json_extract(result_json,'$.pending_action') AS pending_action, " + "json_extract(result_json,'$.node_id') AS node_id " + 'FROM forest_study_work WHERE campaign_id=? AND work_id=?', + (self.campaign_id, row['work_id'])).fetchone() + saved = {'pending_action': raw['pending_action'], 'node_id': raw['node_id']} if raw else {} + pending = None + if session_table and session_updated is not None: + native = db.execute("SELECT json_extract(state_json,'$.pending.node_id') AS node_id, " + "json_extract(state_json,'$.pending.node.code') AS code " + 'FROM forest_study_aide_sessions WHERE campaign_id=? AND work_id=?', + (self.campaign_id, row['work_id'])).fetchone() + if native and native['node_id'] is not None: + pending = {'node_id': native['node_id'], 'code': native['code']} + self._parent_reconciliation_cache[row['work_id']] = { + 'signature': signature, 'saved': saved, 'pending': pending} if not has_decision and saved.get('pending_action') and saved.get('node_id'): program = db.execute('SELECT code,program_json FROM forest_study_aide_programs WHERE campaign_id=? AND work_id=? AND node_id=?', (self.campaign_id, row['work_id'], saved['node_id'])).fetchone() - table = db.execute("SELECT 1 FROM sqlite_master WHERE name='forest_study_aide_sessions'").fetchone() - native = db.execute('SELECT state_json FROM forest_study_aide_sessions WHERE campaign_id=? AND work_id=?', (self.campaign_id, row['work_id'])).fetchone() if table else None - pending = json.loads(native[0]).get('pending') if native else None - has_decision = bool(program and (json.loads(program['program_json']).get('decision') or {}).get('actions') and pending and pending.get('node_id') == saved['node_id'] and (pending.get('node') or {}).get('code') == program['code']) + has_decision = bool(program and (json.loads(program['program_json']).get('decision') or {}).get('actions') and pending and pending.get('node_id') == saved['node_id'] and pending.get('code') == program['code']) children = {edge['target_id'] for edge in edges if edge['source_id'] == row['work_id']} - children.update(map(str, saved.get('pending_work_ids') or [])) if has_decision and children and all(states.get(child) in TERMINAL for child in children): db.execute("UPDATE forest_study_work SET state='retryable',lease_owner=NULL,lease_at=NULL,error=NULL,updated_at=? WHERE campaign_id=? AND work_id=? AND state='awaiting_reconciliation'", (time.time(), self.campaign_id, row['work_id'])) awakened.append((row['work_id'], sorted(children))) + self._parent_reconciliation_cache.pop(row['work_id'], None) for work_id, children in awakened: self.ledger.event(self.campaign_id, 'pending_decision_reconciled', {'settled_child_work_ids': children}, work_id) return [work_id for work_id, _ in awakened] def _ready_ids(self, kinds, compute): with self.ledger.connect() as db: - rows = db.execute("SELECT * FROM forest_study_work WHERE campaign_id=? AND state IN ('queued','retryable') ORDER BY created_at,work_id", (self.campaign_id,)).fetchall() + # Queue polling is frequent. Do not materialize each candidate's + # complete payload and result (which can contain the full durable + # research history) just to decide whether its identity is ready. + # Read those JSON subdocuments only after the cheap dependency + # checks need them, and pass only the provider profile to circuit + # classification. The full payload remains untouched for the + # eventual execution claim. + rows = db.execute("SELECT work_id,kind,state,supervisor_work_id FROM forest_study_work INDEXED BY forest_study_work_live_ready_refs WHERE campaign_id=? AND state IN ('queued','retryable') ORDER BY created_at,work_id", (self.campaign_id,)).fetchall() states = {row['work_id']: row['state'] for row in db.execute('SELECT work_id,state FROM forest_study_work WHERE campaign_id=?', (self.campaign_id,))} edges = db.execute('SELECT source_id,target_id,relation FROM forest_study_edges WHERE campaign_id=?', (self.campaign_id,)).fetchall() - active = {entry['item']['work_id'] for entry in self._futures.values()} - ready = [] - for row in rows: - if row['kind'] not in kinds or row['work_id'] in active: - continue - if any(edge['target_id'] == row['work_id'] and ( - edge['relation'] in BLOCKING_RELATIONS and states.get(edge['source_id']) != 'completed' or - edge['relation'] in SETTLEMENT_RELATIONS and states.get(edge['source_id']) not in TERMINAL) for edge in edges): - continue - if row['state'] == 'retryable': - original = compute.get(row['supervisor_work_id'] or row['work_id']) - if original and original['status'] not in ('completed', 'failed', 'cancelled', 'budget_exhausted', 'reconciliation'): - continue - children = [edge['target_id'] for edge in edges if edge['source_id'] == row['work_id'] and edge['relation'] == 'derived_from'] - if row['decision_json'] and any(states.get(child) not in TERMINAL for child in children): + active = {entry['item']['work_id'] for entry in self._futures.values()} + ready = [] + for row in rows: + if row['kind'] not in kinds or row['work_id'] in active: continue - saved = json.loads(row['result_json']) if row['result_json'] else {} - pending = saved.get('pending_work_ids') or [] - if any(states.get(work) not in TERMINAL for work in pending): + if any(edge['target_id'] == row['work_id'] and ( + edge['relation'] in BLOCKING_RELATIONS and states.get(edge['source_id']) != 'completed' or + edge['relation'] in SETTLEMENT_RELATIONS and states.get(edge['source_id']) not in TERMINAL) for edge in edges): continue - item = {**dict(row), 'payload': json.loads(row['payload_json'])} - if not self._source_blocked(item): - ready.append(row['work_id']) - return ready + if row['state'] == 'retryable': + original = compute.get(row['supervisor_work_id'] or row['work_id']) + if original and original['status'] not in ('completed', 'failed', 'cancelled', 'budget_exhausted', 'reconciliation'): + continue + # Pending action IDs are registered as derived_from edges + # before execution. Use this compact relational index for + # readiness instead of json_extract(result_json): SQLite + # must scan and parse the entire (potentially huge) durable + # episode history even when only pending_work_ids is asked + # for. This mirrors checkpoint engines that route on small + # state metadata and hydrate full snapshots only on resume. + children = [edge['target_id'] for edge in edges + if edge['source_id'] == row['work_id'] and edge['relation'] == 'derived_from'] + if any(states.get(child) not in TERMINAL for child in children): + continue + provider = self._provider_profiles_by_work.get(row['work_id']) + item = {'work_id': row['work_id'], 'kind': row['kind'], 'payload': {'model_provider': provider}} + if not self._source_blocked(item): + ready.append(row['work_id']) + return ready def _make_client(self, item): if self.client_factory: @@ -933,8 +1142,10 @@ def _execute(self, dispatch, item, lane): with self.ledger.connect() as db: db.execute('UPDATE forest_study_dispatches SET state=?,finished_at=?,error=? WHERE dispatch_id=?', ('failed' if error else 'returned', time.time(), error, dispatch)) - db.execute('UPDATE forest_study_dispatches SET result_json=? WHERE dispatch_id=?', - (_json(result) if result is not None else None, dispatch)) + # The complete result is already durably stored by StudyRuntime + # in forest_study_work (and its event delta). Duplicating it in + # every dispatch row multiplied episode histories in the ledger; + # dispatch.work_id is the canonical result reference. self._record_outcome(item, error, failure=failure) def _evaluate_wave(self): @@ -952,6 +1163,19 @@ def _evaluate_wave(self): self._wave = None if not calls: return # Resuming saved actions is not a new inference sample. + # A crashed coordinator can leave a provider request in ``requested`` + # while its external outcome is being reconciled. That is neither a + # successful latency sample nor evidence that the chosen concurrency + # failed. Keep the request reserved by _unsettled_model_owners(), but + # let unrelated work use other capacity measured by the pilot. + unresolved = [call for call in calls + if call['status'] == 'requested' and call['finished_at'] is None] + if unresolved: + self._decision({'type': 'live_concurrency_observation_deferred', + 'degree': self.degree, + 'reason': 'An external model request remains unresolved; preserve its reservation without treating it as a failed throughput sample', + 'request_ids': [call['request_id'] for call in unresolved]}) + return failed = [call for call in calls if call['status'] != 'completed' or call['finished_at'] is None] if failed: previous = self.degree @@ -1039,9 +1263,15 @@ def tick(self): except ValueError: pass # Runtime may have just committed its terminal state. self.runtime.reload_manifest() + self._refresh_provider_profiles() self._read_pilot() self._evaluate_wave() - compute = self._accepted_compute() + with self.ledger.connect() as db: + candidates = db.execute("SELECT work_id,supervisor_work_id FROM forest_study_work INDEXED BY forest_study_work_live_unresolved_refs " + "WHERE campaign_id=? AND state NOT IN ('completed','failed','cancelled','not_run','budget_exhausted')", + (self.campaign_id,)).fetchall() + compute_ids = {value for row in candidates for value in (row['work_id'], row['supervisor_work_id']) if value} + compute = self._accepted_compute(compute_ids) self._reconcile_completed_gpu(compute) self._wake_reconciled_parents() gpu_busy = any(value['configuration'].get('gpu_devices') and diff --git a/tests/test_shared_message_context.py b/tests/test_shared_message_context.py index 30029b3..4861556 100644 --- a/tests/test_shared_message_context.py +++ b/tests/test_shared_message_context.py @@ -11,13 +11,23 @@ DOCUMENT_HEADER, FORMAT, MESSAGE_FORMAT, MESSAGE_HEADER, _message_json_blocks, intern_json_with_stats, restore_json, restore_messages, ) -from research.experiments.study_runtime import _intern_delivered_messages +from research.experiments.study_runtime import _decision_schema_note, _intern_delivered_messages, _json def compact(value): return json.dumps(value, ensure_ascii=False, separators=(",", ":")) +def test_only_codex_cli_uses_its_out_of_band_schema_without_inline_duplication(): + schema = {"type": "object", "properties": {"decision": {"type": "string"}}, + "required": ["decision"], "additionalProperties": False} + codex_note = _decision_schema_note(schema, provider_kind="codex_cli") + other_note = _decision_schema_note(schema, provider_kind="ollama") + assert "attached response schema" in codex_note + assert compact(schema) not in codex_note + assert _json(schema) in other_note + + def public_receipt(size=1400): return {"data_manifest": {"sample_ids": list(range(size)), "unicode": "森林🌲"}, "configuration": {"seed": 1001, "z": 1.0, "a": 1, "signed_zero": -0.0}, @@ -167,26 +177,58 @@ def test_saved_failed_request_projects_below_observed_capacity_read_only(): """Opt-in exact incident proof; never opens a campaign in writable mode.""" database = os.environ.get("FOREST_SHARED_CONTEXT_REQUEST_DB") request_id = os.environ.get("FOREST_SHARED_CONTEXT_REQUEST_ID") - if not database or not request_id: - pytest.skip("Set an explicit read-only saved-request fixture to verify the incident") - with sqlite3.connect("file:" + str(Path(database).resolve()) + "?mode=ro", uri=True) as db: - raw = db.execute("SELECT request_json FROM forest_study_model_calls WHERE request_id=?", (request_id,)).fetchone()[0] - request = json.loads(raw) - trace = json.loads(Path(request["transport_trace_path"]).read_text(encoding="utf-8")) - payload = deepcopy(request["transport_payload"]) + trace_path = os.environ.get("FOREST_SHARED_CONTEXT_TRACE_PATH") + raw_request = None + request = None + if trace_path: + trace = json.loads(Path(trace_path).read_text(encoding="utf-8")) + payload = json.loads(trace["stdin"]) + if database and request_id: + with sqlite3.connect("file:" + str(Path(database).resolve()) + "?mode=ro", uri=True) as db: + raw_request = db.execute("SELECT request_json FROM forest_study_model_calls WHERE request_id=?", + (request_id,)).fetchone()[0] + request = json.loads(raw_request) + elif database and request_id: + with sqlite3.connect("file:" + str(Path(database).resolve()) + "?mode=ro", uri=True) as db: + raw_request = db.execute("SELECT request_json FROM forest_study_model_calls WHERE request_id=?", + (request_id,)).fetchone()[0] + request = json.loads(raw_request) + trace = json.loads(Path(request["transport_trace_path"]).read_text(encoding="utf-8")) + # Older study-call records used ``transport_payload``; the current episode + # runtime stores the same exact wire object as ``request_payload``. + original_payload = request.get("transport_payload", request.get("request_payload")) + assert isinstance(original_payload, dict) + payload = deepcopy(original_payload) + else: + pytest.skip("Set a saved trace path or read-only saved-request fixture to verify the incident") + original_schema = deepcopy(payload.get("response_schema")) + original_messages = deepcopy((request or {}).get("original_messages", payload["messages"])) + schema_note_schema = (request or {}).get("decision_schema") or original_schema + if schema_note_schema and "--output-schema" in trace.get("argv", []): + from research.experiments.study_runtime import _decision_schema_note, _json + old_note = "\n\nReturn exactly one JSON object satisfying this decision contract:\n" + _json(schema_note_schema) + new_note = _decision_schema_note(schema_note_schema, provider_kind="codex_cli") + matching = [index for index, message in enumerate(original_messages) + if isinstance(message.get("content"), str) and message["content"].endswith(old_note)] + if matching: + assert len(matching) == 1 + index = matching[0] + original_messages[index]["content"] = original_messages[index]["content"][:-len(old_note)] + new_note assert compact(payload) == trace["stdin"] from research.agents.codex_transport import _input_rejection capacity = _input_rejection(trace["stderr"])["max_chars"] assert len(compact(payload)) > capacity - delivered, stats = _intern_delivered_messages(request["original_messages"]) - assert restore_messages(delivered) == request["original_messages"] + delivered, stats = _intern_delivered_messages(original_messages) + assert restore_messages(delivered) == original_messages payload["messages"] = delivered assert len(compact(payload)) < capacity - assert stats[0]["scope"] == "request_messages" + assert stats and sum(int(row.get("saved_characters", 0)) for row in stats) > 0 # The original source schema and immutable receipt are unaffected. - assert payload["response_schema"] == request["transport_payload"]["response_schema"] - with sqlite3.connect("file:" + str(Path(database).resolve()) + "?mode=ro", uri=True) as db: - assert db.execute("SELECT request_json FROM forest_study_model_calls WHERE request_id=?", (request_id,)).fetchone()[0] == raw + assert payload["response_schema"] == original_schema + if raw_request is not None: + with sqlite3.connect("file:" + str(Path(database).resolve()) + "?mode=ro", uri=True) as db: + assert db.execute("SELECT request_json FROM forest_study_model_calls WHERE request_id=?", + (request_id,)).fetchone()[0] == raw_request def _legacy_delivery(messages): diff --git a/tests/test_study_memory.py b/tests/test_study_memory.py index 120e0c4..de44217 100644 --- a/tests/test_study_memory.py +++ b/tests/test_study_memory.py @@ -27,6 +27,38 @@ def test_exact_complete_pages_and_restart(tmp_path): assert metadata['segment_id'] == reader.save_json('source', value)['segment_id'] +def test_repeated_public_history_is_stored_losslessly_without_repeated_payloads(tmp_path): + shared = {'data': {'train_ids': list(range(400)), 'validation_ids': list(range(500, 700))}, + 'training': {'width': 64, 'layers': 4, 'learning_rate': 0.0003}, 'seed': 1001} + value = {'action_history': [{'action_id': f'action-{index}', + 'artifacts': [shared, shared], 'observations': [shared, shared], + 'result': {'score': index / 10}} for index in range(24)]} + expected = compact(value) + memory = StudyMemory(tmp_path) + metadata = memory.save_json('public-context', value) + source_path = memory.store.root / (metadata['segment_id'] + '.txt') + stored = source_path.read_text(encoding='utf-8') + + assert metadata['storage_encoding'] == 'forest.lossless-json-subtrees.v1' + assert metadata['characters'] == len(expected) + assert len(stored) < len(expected) * .4 + reader = StudyMemory(tmp_path) + pages, offset = [], 0 + while True: + page = reader.read_json(metadata['segment_id'], offset=offset, limit=997) + pages.append(page['content']) + assert page['total_characters'] == len(expected) + offset = page['next_offset'] + if page['complete']: + break + reconstructed = ''.join(pages) + assert reconstructed == expected + assert json.loads(reconstructed) == value + pointer = reader.read_json(metadata['segment_id'], pointer='/action_history/0/artifacts/0') + assert json.loads(pointer['content']) == shared + assert metadata['segment_id'] == reader.save_json('public-context', value)['segment_id'] + + def test_versions_never_destroy_original_or_leak_other_episode(tmp_path): memory = StudyMemory(tmp_path/'episode-one') original = memory.save_json('same-key', {'value': 'old'}) diff --git a/tests/test_study_runtime.py b/tests/test_study_runtime.py index a506649..b33b70e 100644 --- a/tests/test_study_runtime.py +++ b/tests/test_study_runtime.py @@ -6,10 +6,12 @@ from research.experiments.study_runtime import (StudyLedger, StudyRuntime, _manifest_work_items, _manifest_edges, _known_source_preflight_failure, + _result_event_delta, _compact_action_history_frontiers, _decode_study_result_json, numpy_score_predictions, _normalize_episode_payload, _episode_mutation_id, _c2_evaluation_overrides, _episode_metric_contract, verify_episode_submission, execute_registered_analysis, _codex_strict_schema, _restore_strict_optional_nulls, _transport_trace_reference, _validate_episode_schema) +from research.experiments.study_runtime import _semantic_submission, _validation_signature def manifest(items, edges=(), *, version=1): @@ -49,6 +51,174 @@ def test_manifest_registration_versions_only_unlaunched_work_revisions(tmp_path) assert db.execute("SELECT count(*) FROM forest_study_work").fetchone()[0] == 1 +def test_manifest_and_action_reregistration_do_not_hydrate_saved_results(tmp_path): + ledger = StudyLedger(tmp_path / "study.sqlite", "ledger-test") + data = manifest([item("episode-long", kind="replay")]) + ledger.register(data) + large_result = {"state": "retryable", "retained_evidence": "evidence" * 50_000} + ledger.set_state("campaign-test", "episode-long", "retryable", result=large_result) + action = {"work_id": "episode-child", "kind": "analysis", "action_name": "inspect"} + ledger.register_action("campaign-test", "episode-child", action, parent_id="episode-long") + ledger.set_state("campaign-test", "episode-child", "completed", result=large_result) + + statements = [] + original_connect = ledger.connect + def traced_connect(): + db = original_connect() + db.set_trace_callback(statements.append) + return db + ledger.connect = traced_connect + assert ledger.register(data)["unchanged"] == 1 + ledger.register_action("campaign-test", "episode-child", action, parent_id="episode-long") + + assert not any("select * from forest_study_work" in query.lower() for query in statements) + with original_connect() as db: + saved = json.loads(db.execute("SELECT result_json FROM forest_study_work WHERE work_id='episode-child'").fetchone()[0]) + assert saved["retained_evidence"] == large_result["retained_evidence"] + + +def test_completed_action_id_conflict_returns_replan_feedback_without_reexecution(tmp_path, monkeypatch): + campaign = tmp_path / "campaign" + campaign.mkdir() + manifest_path = campaign / "manifest.json" + manifest_path.write_text(json.dumps(manifest([])), encoding="utf-8") + runtime = StudyRuntime(campaign=campaign, ledger=campaign / "study.sqlite", + ledger_id="ledger-test", manifest=manifest_path) + work_id = "fit-c4-anchor-1001" + old_binding = {"work_id": work_id, "kind": "fit", "action_name": "train", + "configuration_id": "T02_c0", "seed": 1001, + "configuration_path": str(tmp_path / "anchor.json"), + "output_directory": str(tmp_path / "fit-output"), + "effective_config": {"seed": 1001, "optimizer_updates": 200, + "dataset_version_id": "etth1-current-researcher-v1"}, + "source_files": ["/previous/FOREST-source/research/experiments/study_runtime.py"], + "fit_timeout_seconds": 90} + runtime.ledger.register_action("campaign-test", work_id, old_binding, parent_id="episode-test") + saved_result = {"status": "SUCCEEDED", "work_id": work_id, + "effective_config": old_binding["effective_config"], + "validation_metrics": {"rmse": 0.4, "mae": 0.3}} + runtime.ledger.set_state("campaign-test", work_id, "completed", result=saved_result, + supervisor_work_id=work_id) + + new_binding = {**old_binding, + "source_files": ["/current/worktree/research/experiments/study_runtime.py"]} + episode = {"work_id": "episode-test", "payload": {"action_bindings": [new_binding]}} + executed = [] + monkeypatch.setattr(runtime, "execute_fit", lambda *args, **kwargs: executed.append(args)) + outcome = runtime.execute_decision_actions(episode, [{"kind": "train", + "action_work_id": work_id, "configuration_id": "T02_c0", "seed": 1001}]) + + assert outcome["state"] == "retryable" + assert outcome["request_new_decision"] is True + assert outcome["response_validation_feedback"]["reason"] == "source_bound_action_id_conflict" + assert outcome["response_validation_feedback"]["saved_result_available"] is True + assert outcome["actions"][0]["result"]["execution_started"] is False + assert not executed + with runtime.ledger.connect() as db: + saved = db.execute("SELECT state,result_json FROM forest_study_work WHERE work_id=?", (work_id,)).fetchone() + assert saved["state"] == "completed" + assert _decode_study_result_json(saved["result_json"]) == saved_result + + +def test_submission_validation_signature_ignores_claim_rewording_but_keeps_source_evidence(): + first = {"valid": False, "reason": "unverified_claim", + "submitted_claim": "The revised result is unresolved.", + "unavailable_action_work_ids": ["fit-a"], "scientific_repair_complete": False} + reworded = {**first, "submitted_claim": "No revised result has been established."} + changed_evidence = {**reworded, "unavailable_action_work_ids": ["fit-b"]} + + assert _validation_signature(first) == _validation_signature(reworded) + assert _validation_signature(first) != _validation_signature(changed_evidence) + + +def test_semantic_submission_treats_record_id_order_as_non_scientific(): + first = {"invalidate_record_ids": ["fit-b", "fit-a"], "reuse_record_ids": ["pred-2", "pred-1"], + "scientific_repair_complete": False} + reordered = {**first, "invalidate_record_ids": ["fit-a", "fit-b"], + "reuse_record_ids": ["pred-1", "pred-2"]} + changed = {**reordered, "invalidate_record_ids": ["fit-a", "fit-c"]} + + assert _semantic_submission(first) == _semantic_submission(reordered) + assert _semantic_submission(first) != _semantic_submission(changed) + + +def test_action_history_events_store_only_changed_turns_and_keep_full_checkpoint(tmp_path): + ledger = StudyLedger(tmp_path / "study.sqlite", "ledger-test") + ledger.register(manifest([item("episode-1", kind="replay")])) + frontier = {"evidence": [{"record": "source" * 4000}]} + first_turn = {"source_frontier": frontier, + "actions": [{"action_work_id": "inspect-1", "detail": "A" * 20000}]} + first = {"state": "retryable", "action_history": [first_turn]} + ledger.set_state("campaign-test", "episode-1", "retryable", result=first, + event_result=_result_event_delta(first, [])) + second_turn = {"source_frontier": frontier, + "actions": [{"action_work_id": "inspect-2", "detail": "B" * 20000}]} + second = {"state": "retryable", "action_history": [first_turn, second_turn]} + ledger.set_state("campaign-test", "episode-1", "retryable", result=second, + event_result=_result_event_delta(second, first["action_history"])) + + with ledger.connect() as db: + checkpoint_raw = db.execute( + "SELECT result_json FROM forest_study_work WHERE work_id='episode-1'").fetchone()[0] + checkpoint = _decode_study_result_json(checkpoint_raw) + events = [json.loads(row[0]) for row in db.execute( + "SELECT payload_json FROM forest_study_events WHERE work_id='episode-1' AND event_type='work_state' ORDER BY rowid")] + event_sizes = [len(_json_value) for _json_value in db.execute( + "SELECT payload_json FROM forest_study_events WHERE work_id='episode-1' AND event_type='work_state' ORDER BY rowid")] + assert checkpoint["action_history"] == [first_turn, second_turn] + assert all("action_history" not in event["result"] for event in events) + restored = [] + for event in events: + delta = event["result"]["action_history_delta"] + restored[delta["replace_from_turn"]:] = delta["turns"] + assert len(restored) == delta["turn_count"] + assert _decode_study_result_json(json.dumps({"action_history": restored}))["action_history"] == checkpoint["action_history"] + assert event_sizes[1] < len(json.dumps(checkpoint)) + assert event_sizes[1] < 30000 + claimed = ledger.claim("campaign-test", "recovery-worker") + assert claimed["result"]["action_history"] == checkpoint["action_history"] + assert claimed["result"]["action_history"][0]["source_frontier"] == frontier + + +def test_action_history_reuses_repeated_frontiers_losslessly(): + frontier = {"evidence": [{"record": "large" * 1000}]} + history = [{"turn": index, "source_frontier": json.loads(json.dumps(frontier)), + "actions": [{"action_work_id": f"action-{index}"}]} for index in range(8)] + compact = _compact_action_history_frontiers(history) + expanded = _decode_study_result_json(json.dumps({"action_history": compact}))["action_history"] + + assert [turn["source_frontier"] for turn in expanded] == [frontier] * len(history) + assert all(turn["source_frontier"] is expanded[0]["source_frontier"] for turn in expanded) + assert len(json.dumps(compact)) < len(json.dumps(history)) / 5 + + +def test_claim_losslessly_compacts_legacy_checkpoint_frontiers(tmp_path): + ledger = StudyLedger(tmp_path / "study.sqlite", "ledger-test") + ledger.register(manifest([item("legacy-replay", kind="replay")])) + message = "shared immutable research evidence " * 500 + history = [] + frontier = {"messages": [{"role": "user", "content": message}], + "completed_source_actions": {"source-action": {"receipt": "validated"}}} + for index in range(12): + history.append({"source_frontier": json.loads(json.dumps(frontier)), + "actions": [{"action_work_id": f"action-{index}"}]}) + legacy = {"state": "retryable", "action_history": history} + with ledger.connect() as db: + db.execute("UPDATE forest_study_work SET result_json=? WHERE campaign_id=? AND work_id=?", + (json.dumps(legacy), "campaign-test", "legacy-replay")) + before = len(json.dumps(legacy)) + claimed = ledger.claim("campaign-test", "recovery-worker") + assert claimed["result"]["action_history"] == history + assert all(turn["source_frontier"] is claimed["result"]["action_history"][0]["source_frontier"] + for turn in claimed["result"]["action_history"]) + with ledger.connect() as db: + after = db.execute("SELECT result_json FROM forest_study_work WHERE work_id='legacy-replay'").fetchone()[0] + event_count = db.execute("SELECT count(*) FROM forest_study_events WHERE work_id='legacy-replay'").fetchone()[0] + assert len(after) < before / 5 + assert _decode_study_result_json(after)["action_history"] == history + assert event_count == 1 # Compaction adds no synthetic research event. + + @pytest.mark.parametrize("external_evidence", ["claim", "released_claim", "supervisor", "model_call", "fit_attempt"]) def test_manifest_revision_rejected_after_claim_or_external_execution_evidence(tmp_path, external_evidence): ledger = StudyLedger(tmp_path / "study.sqlite", "ledger-test") @@ -240,6 +410,14 @@ def test_arm_public_context_is_exactly_paged_and_reads_are_observational(tmp_pat assert page["content"] == encoded_source_string[19:56] with runtime.ledger.connect() as db: assert db.execute("SELECT count(*) FROM forest_study_work WHERE work_id=?", (context_binding["work_id"],)).fetchone()[0] == 0 + event = db.execute("SELECT payload_json FROM forest_study_events WHERE event_type='context_page_read'").fetchone() + payload_reference = json.loads(event[0]) + assert "result" not in payload_reference + assert payload_reference["result_reference"] == {"segment_id": source_segment, + "pointer": "/messages/1/content", "offset": 19, "next_offset": 56} + restored_page = memory.read_json(source_segment, pointer=payload_reference["pointer"], + offset=payload_reference["offset"], limit=payload_reference["next_offset"]-payload_reference["offset"]) + assert restored_page["content"] == page["content"] history = [{"actions": result["actions"]}] repeated_payload = runtime._attach_public_context(prepared, payload, history=history) assert repeated_payload["context_memory"]["segment_id"] == source_segment @@ -1149,6 +1327,37 @@ def read(offset, end, segment="segment-v1", requested_limit=None): assert _episode_source_frontier(context_note) == _episode_source_frontier(frontier) +def test_context_page_repeat_is_detected_across_immutable_segment_revisions(): + from research.experiments.study_runtime import _context_page_no_progress + + frontier = {"messages": [{"role": "user", "content": "question"}], + "completed_source_actions": {}} + + def read(segment, content="same source page"): + return {"kind": "analysis", "state": "completed", "result": { + "operation": "read_context_segment", "segment_id": segment, + "source_key": "study-public:public-context", "pointer": "/messages/0", + "offset": 0, "next_offset": len(content), "returned_characters": len(content), + "requested_limit": 100, "content": content, "delivery_status": "delivered"}} + + original = {"source_frontier": frontier, "actions": [read("snapshot-v1")]} + first_repeat = _context_page_no_progress([read("snapshot-v2")], [original], frontier) + assert first_repeat["state"] == "retryable" + assert first_repeat["request_new_decision"] is True + diagnosed_turn = {"source_frontier": frontier, "actions": [read("snapshot-v2")], + "no_progress_observation": first_repeat["no_progress_observation"]} + second_repeat = _context_page_no_progress( + [read("snapshot-v3")], [original, diagnosed_turn], frontier) + assert second_repeat["state"] == "paused_no_progress" + assert second_repeat["no_progress_observation"]["context_page_request_keys"] == [{ + "source": "study-public:public-context", "pointer": "/messages/0", + "offset": 0, "requested_limit": 100}] + + changed_page = _context_page_no_progress( + [read("snapshot-v4", "changed source page")], [original], frontier) + assert changed_page is None + + def test_action_batch_is_fully_bound_before_first_operation(tmp_path, monkeypatch): campaign = tmp_path / "campaign" campaign.mkdir() @@ -1184,6 +1393,59 @@ def test_paused_no_progress_requires_explicit_resume_and_preserves_result(tmp_pa assert json.loads(row["result_json"]) == prior_result +def test_resumed_episode_delivers_no_progress_diagnostic_and_uses_saved_action_results(tmp_path, monkeypatch): + campaign = tmp_path / "campaign" + campaign.mkdir() + payload = {"arm": "FULL", "phase": "INFRA_TEST", + "messages": [{"role": "user", "content": "Use the evidence already in the ledger."}], + "decision_schema": {"type": "object", "required": ["mode", "actions", "submission", "rationale"], + "properties": {"mode": {"type": "string"}, "actions": {"type": "array"}, + "submission": {"type": "object"}, "rationale": {"type": "string"}}}, + "allowed_decisions": ["select"], "action_bindings": []} + episode = {"work_id": "replanned-episode", "kind": "episode", "phase": "INFRA_TEST", + "payload": payload} + manifest_path = campaign / "manifest.json" + manifest_path.write_text(json.dumps(manifest([episode])), encoding="utf-8") + runtime = StudyRuntime(campaign=campaign, ledger=campaign / "study.sqlite", + ledger_id="ledger-test", manifest=manifest_path) + action_id = "saved-select-result" + prior_result = {"action_history": [{"actions": [{"action_work_id": action_id, + "kind": "analysis", "state": "completed", "result": {"status": "SUCCEEDED", + "selected_configuration_id": "candidate"}}], + "no_progress_observation": {"action_work_ids": [action_id], + "same_source_bound_results": True, "new_evidence_added": False, + "diagnostic_previously_delivered": True}}]} + runtime.ledger.set_state("campaign-test", episode["work_id"], "paused_no_progress", result=prior_result) + runtime.ledger.resume_no_progress("campaign-test", episode["work_id"], operator="codex-test") + + class SubmitClient: + provider = {"id": "fixture-provider", "model": "fixture-model"} + config = {"reasoning_effort": "low"} + def __init__(self): self.messages = [] + def complete(self, messages, **_kwargs): + self.messages.append(messages) + return {"text": json.dumps({"mode": "submit", "actions": [], + "submission": {"claim": "Use the completed selector result."}, + "rationale": "The saved source result is sufficient."}), + "usage": {"input_tokens": 10, "output_tokens": 5}} + + monkeypatch.setattr("research.experiments.study_runtime.verify_episode_submission", + lambda *_args, **_kwargs: {"valid": True, "state": "completed"}) + client = SubmitClient() + result = runtime.run_one(owner="codex-test", work_id=episode["work_id"], client=client) + + assert result["state"] == "completed" + assert len(client.messages) == 1 + delivered = "\n".join(message["content"] for message in client.messages[0]) + assert "NO_PROGRESS_REPLAN" in delivered + assert action_id in delivered + assert "Do not execute those same IDs again unless their source evidence changes" in delivered + assert "selected_configuration_id" not in json.dumps(result["decision"]["submission"]) + with runtime.ledger.connect() as db: + assert db.execute("SELECT state FROM forest_study_work WHERE work_id=?", (action_id,)).fetchone() is None + assert db.execute("SELECT state FROM forest_study_work WHERE work_id=?", (episode["work_id"],)).fetchone()[0] == "completed" + + def test_codex_transport_schema_is_strict_without_changing_runtime_schema(): schema = {"type": "object", "required": ["mode", "actions", "submission", "rationale"], "properties": {"mode": {"enum": ["action", "submit"]}, @@ -1795,7 +2057,8 @@ def test_cli_leaves_mixed_campaign_provider_resolution_to_each_work_item(tmp_pat assert runtime._model_provider_spec(document["execution_state"]["work_items"][0]) == document["execution_state"]["work_items"][0]["model_provider"] -def test_required_controller_restart_final_gate_checks_the_source_receipt(tmp_path, monkeypatch): +@pytest.mark.parametrize("completion_reference", ["embedded", "recorded_path", "foreign_path"]) +def test_required_controller_restart_final_gate_checks_the_source_receipt(tmp_path, monkeypatch, completion_reference): from research.experiments.study_runtime import _verify_required_controller_restart campaign = tmp_path / "campaign" campaign.mkdir() @@ -1815,6 +2078,26 @@ def test_required_controller_restart_final_gate_checks_the_source_receipt(tmp_pa publication = {"checkpoint": checkpoint, "policy": policy, "request_id": "request", "dispatch_id": "dispatch-a", "worker": {"pid": 101, "process_created": 1.0}} checkpoint_path.write_text(json.dumps(publication), encoding="utf-8") + completion = {"status": "completed", "exit_code": 0, "work_id": "dispatch-a", + "launch_id": "launch-a", "configuration_version": "config-a", + "host_pid": 303, "host_created": 3.0} + previous_exit = {"status": "completed", "pid": 101, "process_created": 1.0, + "work_id": "dispatch-a", "launch_id": "launch-a", "configuration_version": "config-a", + "host_pid": 303, "host_created": 3.0, "artifacts": [artifact]} + if completion_reference == "embedded": + completion_value = completion + else: + folder = campaign / ".forest-study-controllers" / "dispatch-a" + completion_path = folder / "processes" / "dispatch-a" / "completion.json" + completion_path.parent.mkdir(parents=True) + completion_path.write_text(json.dumps(completion), encoding="utf-8") + with runtime.ledger.connect() as db: + db.execute("""CREATE TABLE forest_study_controller_processes ( + dispatch_id TEXT, campaign_id TEXT, work_id TEXT, folder TEXT)""") + db.execute("INSERT INTO forest_study_controller_processes VALUES(?,?,?,?)", + ("dispatch-a", runtime.campaign_id, "restart-cell", str(folder))) + completion_value = str(completion_path if completion_reference == "recorded_path" + else campaign / "unrelated-completion.json") monkeypatch.setattr("research.agents.processes.artifact_matches", lambda *_args: True) with runtime.ledger.connect() as db: db.execute("""CREATE TABLE forest_study_controller_restarts ( @@ -1824,8 +2107,7 @@ def test_required_controller_restart_final_gate_checks_the_source_receipt(tmp_pa resume_process_created REAL, receipt_json TEXT, requested_at REAL, resumed_at REAL, PRIMARY KEY(campaign_id, work_id))""") receipt = {"milestone": policy["milestone"], "previous_exit": { - "status": "completed", "pid": 101, "process_created": 1.0, - "completion_receipt": {"exit_code": 0}, "artifacts": [artifact]}, + **previous_exit, "completion_receipt": completion_value}, "previous_worker": {"pid": 101, "process_created": 1.0}, "resumed_worker": {"pid": 202, "process_created": 2.0}, "checkpoint_reloaded": True, @@ -1837,8 +2119,12 @@ def test_required_controller_restart_final_gate_checks_the_source_receipt(tmp_pa (runtime.campaign_id, "restart-cell", "request", json.dumps(policy), "resumed", json.dumps(checkpoint), "dispatch-a", 101, 1.0, "dispatch-b", 202, 2.0, json.dumps(receipt), 1.0, 2.0)) checked = _verify_required_controller_restart(runtime, episode, valid_result) - assert checked["valid"] is True - assert checked["controller_restart"]["previous_worker"]["pid"] != checked["controller_restart"]["resumed_worker"]["pid"] + if completion_reference == "foreign_path": + assert checked["valid"] is False + assert checked["reason"] == "required_controller_restart_receipt_failed_source_validation" + else: + assert checked["valid"] is True + assert checked["controller_restart"]["previous_worker"]["pid"] != checked["controller_restart"]["resumed_worker"]["pid"] def test_aide_verified_metrics_accepts_actual_common_rescore_artifact_shape(tmp_path): diff --git a/tests/test_study_scheduler.py b/tests/test_study_scheduler.py index 3291da8..7671b49 100644 --- a/tests/test_study_scheduler.py +++ b/tests/test_study_scheduler.py @@ -13,6 +13,7 @@ import pytest from research.experiments.study_runtime import StudyLedger +import research.experiments.study_scheduler as scheduler_module from research.experiments.study_scheduler import ( CoordinatorBusy, StudyScheduler, interval_measure, intersect_intervals, measured_concurrency) @@ -150,6 +151,53 @@ def test_current_formal_progress_excludes_historical_and_withdrawn_cells(tmp_pat service.close() +def test_ready_poll_reads_only_small_state_from_large_durable_records(tmp_path, monkeypatch): + history = 'retained research evidence ' * 10000 + provider_profile = {'id': 'work-provider', 'kind': 'codex_cli', 'model': 'gpt-6-luna', + 'config': {'reasoning_effort': 'medium'}} + runtime = ChildRuntime(tmp_path, [ + {'work_id': 'large-queued', 'kind': 'replay', 'context': history, + 'model_provider': provider_profile}, + {'work_id': 'large-retryable', 'kind': 'replay', 'context': history}]) + runtime.ledger.register_action(runtime.campaign_id, 'pending-source', + {'work_id': 'pending-source', 'kind': 'infer', 'phase': 'DEV'}, parent_id='large-retryable') + runtime.ledger.set_state(runtime.campaign_id, 'large-retryable', 'retryable', + result={'pending_work_ids': ['pending-source'], 'history': history}) + service = scheduler(runtime) + original_loads = scheduler_module.json.loads + statements = [] + original_connect = runtime.ledger.connect + + def traced_connect(): + db = original_connect() + db.set_trace_callback(statements.append) + return db + + monkeypatch.setattr(runtime.ledger, 'connect', traced_connect) + assert service._provider_profiles_by_work['large-queued'] == provider_profile + + def reject_full_record(value, *args, **kwargs): + if isinstance(value, str) and ('"context"' in value or '"history"' in value): + raise AssertionError('queue polling materialized a full durable research record') + return original_loads(value, *args, **kwargs) + + monkeypatch.setattr(scheduler_module.json, 'loads', reject_full_record) + try: + assert service._ready_ids(('replay',), {}) == ['large-queued'] + with runtime.ledger.connect() as db: + db.execute("UPDATE forest_study_work SET state='completed' WHERE work_id='pending-source'") + assert service._ready_ids(('replay',), {}) == ['large-queued', 'large-retryable'] + assert not any('json_extract(result_json' in query.lower() for query in statements) + assert not any('json_extract(payload_json' in query.lower() for query in statements) + with runtime.ledger.connect() as db: + assert original_loads(db.execute('SELECT payload_json FROM forest_study_work WHERE work_id=?', + ('large-queued',)).fetchone()[0])['context'] == history + assert original_loads(db.execute('SELECT result_json FROM forest_study_work WHERE work_id=?', + ('large-retryable',)).fetchone()[0])['history'] == history + finally: + service.close() + + def pilot(): return {'pilot_id': 'actual-fixture-timing', 'model': 'infrastructure-child', 'reasoning_effort': 'medium', 'batches': [ @@ -168,6 +216,29 @@ def test_pilot_uses_only_observed_successful_authorized_degrees(): assert measured_concurrency(data, 9)['degree'] == 1 +def test_resume_merges_pilot_and_saved_live_concurrency_samples(tmp_path): + runtime = ChildRuntime(tmp_path, []) + pilot_path = tmp_path / 'pilot.json' + pilot_path.write_text(json.dumps(pilot()), encoding='utf-8') + service = scheduler(runtime, maximum_model_concurrency=2, model_concurrency=2, + pilot_path=pilot_path) + try: + saved = service._configuration() + saved['model_concurrency'] = 1 + saved['observed_samples'] = [{'degree': 1, 'calls_per_second': .06, + 'p95_seconds': 15., 'completed_calls': 2, 'wall_seconds': 30.}] + saved['tested_degrees'] = [1, 2] + with runtime.ledger.connect() as db: + db.execute('INSERT INTO forest_study_schedulers VALUES(?,?,?,?,?,?,?,?,?,?)', + (runtime.campaign_id, 'previous-scheduler', 'previous-session', 'old-host', + None, None, 'stopped', json.dumps(saved), 1., 2.)) + service.acquire() + assert service.degree == 2 + assert {sample['degree'] for sample in service._observed_samples} == {1, 2} + finally: + service.close() + + def test_serial_calls_are_not_evidence_of_declared_parallel_capacity(): data=pilot() batch=data['batches'][1] @@ -212,6 +283,33 @@ def test_live_sequential_requests_use_actual_overlap_degree(tmp_path): service.close() +def test_unresolved_external_request_does_not_downrate_pilot_concurrency(tmp_path): + runtime = ChildRuntime(tmp_path, []) + service = scheduler(runtime, model_concurrency=2, maximum_model_concurrency=2) + with runtime.ledger.connect() as db: + db.execute('INSERT INTO forest_study_dispatches(dispatch_id,campaign_id,scheduler_id,session_id,owner,work_id,lane,state,accepted_at,started_at,finished_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)', + ('dispatch-unresolved-wave', runtime.campaign_id, service.scheduler_id, service.session_id, + 'owner', 'native-node', 'model', 'reconciled', 0., 0., 5.)) + for request_id,status,started,finished,elapsed in ( + ('completed-wave-call', 'completed', 1., 2., 1.), + ('unresolved-wave-call', 'requested', 1.1, None, None)): + db.execute('INSERT INTO forest_study_model_calls(request_id,campaign_id,work_id,request_json,status,recorded_at,started_at,finished_at,elapsed_seconds) VALUES(?,?,?,?,?,?,?,?,?)', + (request_id, runtime.campaign_id, 'native-node', json.dumps({'model': 'infrastructure-child'}), + status, started, started, finished, elapsed)) + service._wave = {'degree': 2, 'dispatch_ids': ['dispatch-unresolved-wave']} + try: + service._evaluate_wave() + assert service.degree == 2 + assert service._wave is None + with runtime.ledger.connect() as db: + request = db.execute("SELECT status FROM forest_study_model_calls WHERE request_id=?", + ('unresolved-wave-call',)).fetchone() + assert service._pending_decisions[-1]['type'] == 'live_concurrency_observation_deferred' + assert request['status'] == 'requested' + finally: + service.close() + + def test_exhausted_cell_is_terminal_for_scheduler(tmp_path): runtime = ChildRuntime(tmp_path, [{'work_id': 'exhausted', 'kind': 'episode', 'context': 'own'}]) runtime.ledger.set_state(runtime.campaign_id, 'exhausted', 'budget_exhausted') @@ -279,6 +377,12 @@ def test_real_independent_children_overlap_and_gpu_lane_is_serial(tmp_path): with runtime.ledger.connect() as db: assert db.execute('SELECT lane FROM forest_study_dispatches ORDER BY accepted_at LIMIT 1').fetchone()[0] == 'gpu' assert db.execute('SELECT count(*) FROM forest_study_dispatches WHERE state=\'returned\'').fetchone()[0] == 4 + # Full outputs belong to the canonical study-work checkpoint. + # Dispatch rows retain the work ID and lifecycle receipt only. + assert db.execute("SELECT count(*) FROM forest_study_dispatches " + "WHERE state='returned' AND result_json IS NOT NULL").fetchone()[0] == 0 + assert all(json.loads(row[0]) for row in db.execute( + "SELECT result_json FROM forest_study_work WHERE state='completed'")) value = service.metrics() assert value['categories']['model_wait']['summed_seconds'] > value['categories']['model_wait']['wall_union_seconds'] assert value['completed_work_items'] == 4 and value['remaining_ready'] == 0 diff --git a/tests/test_study_scheduler_recovery.py b/tests/test_study_scheduler_recovery.py index 9e32e04..e009d30 100644 --- a/tests/test_study_scheduler_recovery.py +++ b/tests/test_study_scheduler_recovery.py @@ -194,6 +194,183 @@ def hold_exit(folder, dispatch): service.close() +def test_parent_reconciliation_projects_pending_fields_without_loading_large_checkpoint(tmp_path, monkeypatch): + """Scheduler reconciliation reads its scalar needs from the stored snapshot.""" + import research.experiments.study_scheduler as module + + runtime = ChildRuntime(tmp_path, [ + {"work_id": "large-parent", "kind": "episode", "context": "episode"}, + {"work_id": "settled-child", "kind": "infer", "context": "inference"}, + ]) + runtime.ledger.set_state(runtime.campaign_id, "settled-child", "completed", result={"receipt": "valid"}) + runtime.ledger.set_state(runtime.campaign_id, "large-parent", "awaiting_reconciliation", + result={"pending_action": True, "node_id": "node-1"}, + decision={"actions": [{"action_work_id": "settled-child"}]}) + with runtime.ledger.connect() as db: + db.execute("INSERT INTO forest_study_edges VALUES(?,?,?,?)", + (runtime.campaign_id, "large-parent", "settled-child", "derived_from")) + db.execute("CREATE TABLE forest_study_aide_sessions(campaign_id TEXT, work_id TEXT, state_json TEXT, updated_at REAL, PRIMARY KEY(campaign_id,work_id))") + db.execute("INSERT INTO forest_study_aide_sessions VALUES(?,?,?,?)", + (runtime.campaign_id, "large-parent", json.dumps({"pending": {"node_id": "node-1", + "node": {"code": "analysis"}}, "action_history": [], "retained_evidence": "x" * 300_000}), time.time())) + + service = scheduler(runtime) + original_loads = module.json.loads + large_values = [] + def track_loads(value, *args, **kwargs): + if isinstance(value, str) and len(value) > 4096: + large_values.append(len(value)) + raise AssertionError("Scheduler decoded a full historical checkpoint during reconciliation") + return original_loads(value, *args, **kwargs) + monkeypatch.setattr(module.json, "loads", track_loads) + try: + assert service._wake_reconciled_parents() == ["large-parent"] + with runtime.ledger.connect() as db: + state = db.execute("SELECT state FROM forest_study_work WHERE work_id='large-parent'").fetchone()[0] + assert state == "retryable" + assert large_values == [] + finally: + service.close() + + +def test_scheduler_live_poll_paths_use_partial_indexes(tmp_path): + runtime = ChildRuntime(tmp_path, [{"work_id": "ready-work", "kind": "replay", "context": "saved"}]) + service = scheduler(runtime) + try: + with runtime.ledger.connect() as db: + db.execute("INSERT INTO forest_study_dispatches(dispatch_id,campaign_id,scheduler_id,session_id,owner,lane,state,accepted_at) VALUES(?,?,?,?,?,?,?,?)", + ("active-dispatch", runtime.campaign_id, "scheduler", "current-session", "owner", "model", "running", time.time())) + db.execute("INSERT INTO forest_study_model_calls(request_id,campaign_id,work_id,request_json,status,recorded_at) VALUES(?,?,?,?,?,?)", + ("pending-request", runtime.campaign_id, "ready-work", "{}", "requested", time.time())) + db.execute("CREATE TABLE forest_study_aide_sessions(campaign_id TEXT,work_id TEXT,state_json TEXT,updated_at REAL,PRIMARY KEY(campaign_id,work_id))") + db.execute("INSERT INTO forest_study_aide_sessions VALUES(?,?,?,?)", + (runtime.campaign_id, "ready-work", json.dumps({"history": "x" * 100_000}), time.time())) + plans = { + "work": db.execute("EXPLAIN QUERY PLAN SELECT work_id,kind,state FROM forest_study_work " + "WHERE campaign_id=? AND state IN ('queued','retryable') ORDER BY created_at,work_id", + (runtime.campaign_id,)).fetchall(), + "ready_refs": db.execute("EXPLAIN QUERY PLAN SELECT work_id,kind,state,supervisor_work_id FROM forest_study_work INDEXED BY forest_study_work_live_ready_refs " + "WHERE campaign_id=? AND state IN ('queued','retryable') ORDER BY created_at,work_id", + (runtime.campaign_id,)).fetchall(), + "unresolved_refs": db.execute("EXPLAIN QUERY PLAN SELECT work_id,supervisor_work_id FROM forest_study_work INDEXED BY forest_study_work_live_unresolved_refs " + "WHERE campaign_id=? AND state NOT IN ('completed','failed','cancelled','not_run','budget_exhausted')", + (runtime.campaign_id,)).fetchall(), + "pending_parent": db.execute("EXPLAIN QUERY PLAN SELECT work_id,decision_json,updated_at FROM forest_study_work INDEXED BY forest_study_work_pending_parent_state " + "WHERE campaign_id=? AND state='awaiting_reconciliation' AND kind IN ('episode','replay')", + (runtime.campaign_id,)).fetchall(), + "pending_compute": db.execute("EXPLAIN QUERY PLAN SELECT work_id,kind,supervisor_work_id,payload_json FROM forest_study_work INDEXED BY forest_study_work_pending_compute_refs " + "WHERE campaign_id=? AND state='awaiting_reconciliation' AND kind IN ('fit','infer')", + (runtime.campaign_id,)).fetchall(), + "dispatch": db.execute("EXPLAIN QUERY PLAN SELECT dispatch_id FROM forest_study_dispatches " + "WHERE campaign_id=? AND session_id<>? AND state IN ('claiming','accepted','running')", + (runtime.campaign_id, "other-session")).fetchall(), + "model_call": db.execute("EXPLAIN QUERY PLAN SELECT work_id FROM forest_study_model_calls " + "WHERE campaign_id=? AND status='requested'", (runtime.campaign_id,)).fetchall(), + } + assert "forest_study_work_live_ready_refs" in " ".join(str(row[3]) for row in plans["work"]) + assert "COVERING INDEX forest_study_work_live_ready_refs" in " ".join( + str(row[3]) for row in plans["ready_refs"]) + assert "COVERING INDEX forest_study_work_live_unresolved_refs" in " ".join( + str(row[3]) for row in plans["unresolved_refs"]) + assert "COVERING INDEX forest_study_work_pending_parent_state" in " ".join( + str(row[3]) for row in plans["pending_parent"]) + assert "forest_study_work_pending_compute_refs" in " ".join( + str(row[3]) for row in plans["pending_compute"]) + service._wake_reconciled_parents() + with runtime.ledger.connect() as db: + aide_plan = db.execute("EXPLAIN QUERY PLAN SELECT work_id,updated_at FROM forest_study_aide_sessions INDEXED BY forest_study_aide_session_live_updated WHERE campaign_id=?", + (runtime.campaign_id,)).fetchall() + assert "COVERING INDEX forest_study_aide_session_live_updated" in " ".join( + str(row[3]) for row in aide_plan) + assert "forest_study_dispatch_live_session" in " ".join(str(row[3]) for row in plans["dispatch"]) + assert "forest_study_model_call_unsettled" in " ".join(str(row[3]) for row in plans["model_call"]) + finally: + service.close() + + +def test_recovered_saved_provider_releases_only_unsettled_environment_circuits(tmp_path, monkeypatch): + import research.experiments.study_scheduler as module + + profile = {'id': 'saved-local', 'kind': 'codex_cli', 'model': 'gpt-6-luna', + 'config': {'reasoning_effort': 'medium'}} + runtime = ChildRuntime(tmp_path, [ + {'work_id': 'safe-retry', 'kind': 'replay', 'context': 'saved', 'model_provider': profile}, + {'work_id': 'uncertain-retry', 'kind': 'replay', 'context': 'saved', 'model_provider': profile}, + {'work_id': 'different-error', 'kind': 'replay', 'context': 'saved', 'model_provider': profile}]) + service = scheduler(runtime) + provider_table_error = 'RuntimeError: OperationalError: no such table: providers' + with runtime.ledger.connect() as db: + for work_id, error in (('safe-retry', provider_table_error), + ('uncertain-retry', provider_table_error), + ('different-error', 'RuntimeError: unrelated failure')): + source = service._source({'work_id': work_id, 'kind': 'replay', + 'payload': {'model_provider': profile}}) + db.execute('INSERT INTO forest_study_scheduler_circuits VALUES(?,?,?,?,?)', + (runtime.campaign_id, source, error, 2, 'blocked')) + db.execute("INSERT INTO forest_study_model_calls(request_id,campaign_id,work_id,request_json,status,recorded_at) " + "VALUES(?,?,?,?,?,?)", ('unsettled-original', runtime.campaign_id, 'uncertain-retry', '{}', + 'requested', time.time())) + + loaded = [] + monkeypatch.setattr(module, 'load_saved_codex_client', + lambda provider_id, **kwargs: loaded.append((provider_id, kwargs)) or object()) + try: + assert service._recover_provider_configuration_circuits() == ['safe-retry'] + assert loaded == [('saved-local', {'model': 'gpt-6-luna', 'reasoning_effort': 'medium'})] + with runtime.ledger.connect() as db: + states = {row['source_id']: (row['state'], row['consecutive_errors'], row['last_error']) + for row in db.execute('SELECT source_id,state,consecutive_errors,last_error ' + 'FROM forest_study_scheduler_circuits WHERE campaign_id=?', (runtime.campaign_id,))} + event = db.execute("SELECT payload_json FROM forest_study_events " + "WHERE event_type='provider_configuration_revalidated'").fetchone() + assert states[service._source({'work_id': 'safe-retry', 'kind': 'replay', + 'payload': {'model_provider': profile}})] == ('ready', 2, provider_table_error) + assert states[service._source({'work_id': 'uncertain-retry', 'kind': 'replay', + 'payload': {'model_provider': profile}})] == ('blocked', 2, provider_table_error) + assert states[service._source({'work_id': 'different-error', 'kind': 'replay', + 'payload': {'model_provider': profile}})] == ('blocked', 2, 'RuntimeError: unrelated failure') + assert json.loads(event['payload_json'])['prior_errors_preserved'] is True + finally: + service.close() + + +def test_tick_compute_reconciliation_hydrates_only_referenced_receipts(tmp_path, monkeypatch): + from research.experiments.gpu_supervisor import ComputeSupervisor + import research.experiments.study_scheduler as module + + runtime = ChildRuntime(tmp_path, [{"work_id": "candidate", "kind": "infer", "context": "registered"}]) + ComputeSupervisor(runtime.ledger_path, runtime.ledger_id) + with runtime.ledger.connect() as db: + for work_id, state, receipt in ( + ("candidate", "completed", json.dumps({"status": "completed", "finished_at": 2})), + ("unrelated-complete", "completed", json.dumps({"large_receipt": "x" * 300_000})), + ("external-active", "running", "{}"), + ): + db.execute("INSERT INTO forest_compute_work(work_id,ledger_id,configuration_json,state,accepted_at,launch_requested,receipt_json) VALUES(?,?,?,?,?,?,?)", + (work_id, runtime.ledger_id, json.dumps({"gpu_devices": ["GPU-test"]}), state, time.time(), 1, receipt)) + + def unknown_status(_self, _work_id): + raise ValueError("No original process publication") + monkeypatch.setattr(ComputeSupervisor, "status", unknown_status) + service = scheduler(runtime) + statements = [] + original_connect = runtime.ledger.connect + def traced_connect(): + db = original_connect() + db.set_trace_callback(statements.append) + return db + runtime.ledger.connect = traced_connect + try: + value = service._accepted_compute({"candidate"}) + assert set(value) == {"candidate", "external-active"} + assert value["candidate"]["status"] == "completed" + assert value["external-active"]["unknown_owner"] is True + assert not any("from forest_compute_work\"" in query.lower() and "where work_id in" not in query.lower() + and "state not in" not in query.lower() for query in statements) + finally: + service.close() + + def test_changed_raw_checkpoint_cannot_satisfy_controller_restart(tmp_path): runtime, _ = controller_fixture_runtime(tmp_path) service = controller_service(runtime) @@ -483,11 +660,24 @@ def test_pending_parent_wakes_only_after_children_settle_without_extra_query(tmp runtime.ledger.set_state(runtime.campaign_id, 'parent', 'awaiting_reconciliation', decision={'child_id': 'child'}, result={'pending_work_ids': ['child']}) service = scheduler(runtime) + statements = [] + original_connect = runtime.ledger.connect + def traced_connect(): + db = original_connect() + db.set_trace_callback(statements.append) + return db + runtime.ledger.connect = traced_connect try: service.tick() assert next(row for row in rows(runtime) if row['work_id'] == 'parent')['state'] == 'awaiting_reconciliation' runtime.ledger.set_state(runtime.campaign_id, 'child', 'completed', result={'nonformal_fixture': True}) run_until(service, lambda: next(row for row in rows(runtime) if row['work_id'] == 'parent')['state'] == 'completed') + assert sum('json_extract(result_json' in query.lower() and 'FROM forest_study_work WHERE campaign_id=' in query + for query in statements) == 1 + assert not any('length(result_json)' in query.lower() or 'length(state_json)' in query.lower() + for query in statements) + assert not any('select * from forest_study_fit_attempts' in query.lower() + for query in statements) with runtime.ledger.connect() as db: assert db.execute('SELECT count(*) FROM forest_study_model_calls').fetchone()[0] == 0 assert json.loads(next(row for row in rows(runtime) if row['work_id'] == 'parent')['result_json'])['resumed_saved_decision'] is True @@ -495,6 +685,36 @@ def test_pending_parent_wakes_only_after_children_settle_without_extra_query(tmp service.close() +def test_recovery_requeues_failed_client_setup_only_before_any_model_call(tmp_path): + runtime = ChildRuntime(tmp_path, [{'work_id': 'orphaned-model-work', 'kind': 'replay', + 'context': 'saved replay checkpoint'}]) + def unavailable_client(_item): + raise RuntimeError('nonformal provider setup failure') + failed = StudyScheduler(runtime, client_factory=unavailable_client, poll_seconds=.01) + try: + failed.tick() + deadline = time.monotonic() + 3 + while time.monotonic() < deadline and not all(future.done() for future in failed._futures): + time.sleep(.01) + assert failed._futures and all(future.done() for future in failed._futures) + assert rows(runtime)[0]['state'] == 'claimed' + with runtime.ledger.connect() as db: + assert db.execute("SELECT count(*) FROM forest_study_model_calls WHERE work_id='orphaned-model-work'").fetchone()[0] == 0 + finally: + failed.close() + + recovered = scheduler(runtime) + try: + recovered.acquire() + assert rows(runtime)[0]['state'] == 'retryable' + run_until(recovered, lambda: rows(runtime)[0]['state'] == 'completed') + with runtime.ledger.connect() as db: + assert db.execute("SELECT count(*) FROM forest_study_model_calls WHERE work_id='orphaned-model-work' AND status='completed'").fetchone()[0] == 1 + assert db.execute("SELECT count(*) FROM forest_study_events WHERE work_id='orphaned-model-work' AND event_type='unstarted_claim_recovered'").fetchone()[0] == 1 + finally: + recovered.close() + + def test_saved_completed_action_still_gets_client_for_next_actual_decision(tmp_path): action_id = 'nonformal-inspect' payload = {'arm': 'FULL', 'messages': [{'role': 'user', 'content': 'Nonformal two-turn fixture'}],