diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 925b747f..5bba6131 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,7 +7,7 @@ permissions: jobs: local: runs-on: ubuntu-latest - timeout-minutes: 30 + timeout-minutes: 45 env: FOREST_MODEL: '' FOREST_DATABASE_URL: sqlite:////tmp/forest-ci.db @@ -54,6 +54,19 @@ jobs: "$install_dir/tectonic" --version - run: python -m pip install -r requirements.lock.txt - run: python -m pip install --no-deps -e . + - name: Install CPU training and native AIDE test dependencies + shell: bash + run: | + set -euo pipefail + python -m pip install torch==2.14.1 --index-url https://download.pytorch.org/whl/cpu + python -m pip install -r requirements-study-ci.txt + aide_root="$RUNNER_TEMP/aide-upstream" + git init "$aide_root" + git -C "$aide_root" remote add origin https://github.com/WecoAI/aideml.git + git -C "$aide_root" fetch --depth=1 origin 60b3978ddf65b71f86eb7c64506965048a1398cf + git -C "$aide_root" checkout --detach FETCH_HEAD + echo "FOREST_AIDE_UPSTREAM=$aide_root" >> "$GITHUB_ENV" + FOREST_AIDE_UPSTREAM="$aide_root" python -c 'import os; from pathlib import Path; from research.experiments.aide_runtime import load_upstream; load_upstream(Path(os.environ["FOREST_AIDE_UPSTREAM"]))' - run: forest --help - run: python scripts/export_api_contract.py --check - run: python -m pytest -q diff --git a/README.md b/README.md index 613ad7ca..ef6034b0 100644 --- a/README.md +++ b/README.md @@ -114,6 +114,16 @@ After installing the Web workspace: ./scripts/test.sh ``` +FOREST approaches scientific research as a **continuous, evidence-driven process rather than a sequence of isolated experiments**. Its persistent Research Graph connects hypotheses, experimental results, and research decisions, enabling agents to iteratively explore, evaluate, and refine scientific solutions. We evaluated this approach using **GPT-6 Luna (Medium)** as the research model, demonstrating autonomous experimentation and measurable scientific improvements across multiple research domains. + +| Research Task | Dataset | Metric | Baseline | FOREST | Improvement | +|---|---|---|---:|---:|---:| +| PDE Gradient Optimization | Darcy64 | Relative Gradient L2 | 0.303645 | **0.256810** | **15.42% ** | +| Time-Series Forecasting (Study 1) | ETTh1 | MAE | 2.914060 | **2.553352** | **12.38% ** | +| PDE Solution Optimization | Darcy64 | Relative L2 | 0.108735 | **0.099608** | **8.39% ** | +| Time-Series Forecasting (Study 2) | ETTh1 | MAE | 2.913240 | **2.728309** | **6.35% ** | +| Medical Image Classification | BloodMNIST | Macro-F1 | 0.965410 | **0.969351** | **+0.394 pp** | + ### Research Modes - **Manual** — edit the research graph and choose tasks to execute. diff --git a/requirements-study-ci.txt b/requirements-study-ci.txt new file mode 100644 index 00000000..3017e6d3 --- /dev/null +++ b/requirements-study-ci.txt @@ -0,0 +1,47 @@ +annotated-types==0.8.0 +anthropic==1.13.0 +anyio==4.15.1 +attrs==26.1.0 +backoff==2.2.1 +black==26.10.1 +click==8.5.0 +coolname==5.0.0 +dataclasses-json==0.6.7 +docstring-parser==0.18.0 +funcy==2.1 +genson==1.4.0 +h11==0.16.0 +httpcore2==2.13.1 +httpx2==2.13.1 +humanize==4.16.0 +idna==3.20 +igraph==1.0.0 +jiter==0.17.0 +jsonschema==4.26.0 +jsonschema-specifications==2025.9.1 +loguru==0.7.3 +markdown-it-py==4.2.0 +marshmallow==3.26.2 +mdurl==0.1.2 +mypy-extensions==1.1.0 +omegaconf==2.4.0 +openai==3.29.0 +packaging==26.3 +pathspec==1.1.1 +platformdirs==4.13.0 +pydantic==2.13.5 +pydantic-core==2.46.5 +pygments==2.21.0 +pytokens==0.4.1 +pyyaml==6.0.3 +referencing==0.37.0 +rich==15.0.0 +rpds-py==2026.9.1 +shutup==0.3.0 +sniffio==1.3.1 +texttable==1.7.1 +tqdm==4.70.1 +truststore==0.10.4 +typing-extensions==4.16.0 +typing-inspect==0.9.0 +typing-inspection==0.4.4 diff --git a/research/experiments/STUDY_CONTEXT.md b/research/experiments/STUDY_CONTEXT.md index 94270f3c..24c998f1 100644 --- a/research/experiments/STUDY_CONTEXT.md +++ b/research/experiments/STUDY_CONTEXT.md @@ -18,6 +18,38 @@ training configurations and execution receipts can share identical payloads. Different revisions, seeds, scientific conditions and evidence identities keep their own original values. Nonidentical evidence is never merged. +`intern_messages` uses one request-wide pool when it reduces the complete +serialized request. Each original JSON block retains its message position and +role through a fragment reference; the complete definitions appear once in the +pool owner's document. Its root is the ordered array of source fragments. +`source_fragments` maps each root element to its original message, role, block, +offset and JSON spelling. A definition's storage position gives it no additional +instruction authority. Literal prefixes, suffixes, message metadata and the +response schema retain their original positions and values. + +`restore_messages` reconstructs exact message strings for supported JSON +spellings. Unrecognized spelling, including custom numeric notation or duplicate +object keys, stays literal and unchanged. Single-block requests retain the +self-contained v1 document format. Older v1 documents without spelling metadata +retain their original guarantee of exact ordered JSON values. Both formats are +readable after restart; completed native requests can replay their original +response only when the saved source messages, task context, settings and schema +are unchanged. Representation changes never create another inference request. + +### Implementation references + +[Pydantic's JSON Schema documentation](https://docs.pydantic.dev/latest/concepts/json_schema/) +shows the reusable definition/reference pattern with `$defs` and `$ref`. +FOREST applies that pattern to complete data values through its own reversible +codec; the data document is not a standard JSON Schema. +[LangChain's ToolMessage source](https://github.com/langchain-ai/langchain/blob/master/libs/core/langchain_core/messages/tool.py) +separates model-visible `content` from a complete `artifact` and retains +`tool_call_id` identity. FOREST's existing `ContextStore`, immutable page receipts +and `read_context_segment` provide the corresponding persistent-evidence and +working-context separation. The request-wide pool changes representation only; +cross-request originals remain stored and retrievable. Neither mechanism +summarizes or truncates scientific source. These references add no dependency. + ## Persistent memory and reads `StudyMemory` stores complete versioned public JSON through the existing diff --git a/research/experiments/aide_runtime.py b/research/experiments/aide_runtime.py index 9e4c4e1a..6499fe5e 100644 --- a/research/experiments/aide_runtime.py +++ b/research/experiments/aide_runtime.py @@ -23,6 +23,7 @@ import time from typing import Any, Mapping import uuid +import warnings SCHEMA = """ @@ -206,7 +207,7 @@ def _review_task_description(native, stdout): def _legacy_native_prompt_equivalent(saved, current): """Verify an old full-context prompt matches after reversible interning.""" - from .study_context import restore_json + from .study_context import restore_messages original = current.get("original_task_messages") if not isinstance(original, list): return False @@ -241,29 +242,10 @@ def replace_exact(messages, needles): try: # The outer task serialization is ordinary JSON; selected user message # bodies may contain their own reversible tagged documents. - if task_context.startswith("COMPLETE LOSSLESS JSON CONTEXT DOCUMENT follows."): - _, outer_document_text = task_context.split("\n", 1) - outer_document, _ = json.JSONDecoder().raw_decode(outer_document_text) - delivered_task = restore_json(outer_document) - else: - delivered_task = json.loads(task_context) - if not isinstance(delivered_task, list) or len(delivered_task) != len(original): + outer = restore_messages([{"role": "user", "content": task_context}])[0]["content"] + delivered_task = json.loads(outer) + if not isinstance(delivered_task, list) or restore_messages(delivered_task) != original: return False - for before, after in zip(original, delivered_task): - if not isinstance(before, Mapping) or not isinstance(after, Mapping) or before.get("role") != after.get("role"): - return False - before_content, after_content = before.get("content"), after.get("content") - if not isinstance(before_content, str) or not isinstance(after_content, str): - if before_content != after_content: - return False - continue - if after_content.startswith("COMPLETE LOSSLESS JSON CONTEXT DOCUMENT follows."): - _, encoded = after_content.split("\n", 1) - document, _ = json.JSONDecoder().raw_decode(encoded) - if restore_json(document) != json.loads(before_content): - return False - elif after_content != before_content: - return False old_normalized, old_count = replace_exact(saved_messages, old_needles) new_normalized, new_count = replace_exact(current_messages, new_needles) except (ValueError, TypeError, RecursionError): @@ -271,6 +253,27 @@ def replace_exact(messages, needles): return old_count > 0 and new_count > 0 and old_normalized == new_normalized +def _native_serialization_equivalent(saved, current): + """Replay an unchanged saved native source after a representation upgrade.""" + if (not isinstance(saved.get("original_messages"), list) + or saved["original_messages"] != current.get("original_messages")): + return False + # These fields derive from the delivered representation or the current + # remaining allowance. Every original source, setting and schema must match. + presentation = {"messages", "context_encoding", "request_payload", "source_output_allocation", + "effective_timeout_seconds", "max_output_tokens", "transport_trace_path"} + if ({key: value for key, value in saved.items() if key not in presentation} + != {key: value for key, value in current.items() if key not in presentation}): + return False + before, after = deepcopy(saved.get("request_payload")), deepcopy(current.get("request_payload")) + if not isinstance(before, dict) or not isinstance(after, dict): + return before == after + for payload in (before, after): + payload.pop("messages", None) + payload.pop("max_output_tokens", None) + return before == after + + def _query_router(*args, **kwargs): backend = _BACKEND.get() if backend is None: @@ -299,7 +302,14 @@ def load_upstream(root: Path): raise ValueError("One process cannot mix different upstream AIDE clones") if str(root) not in sys.path: sys.path.insert(0, str(root)) - module = importlib.import_module("aide.agent") + # Native AIDE imports shutup, which replaces warnings.warn globally. + # Keep the caller's warning policy after importing the native modules. + original_warn = warnings.warn + try: + with warnings.catch_warnings(): + module = importlib.import_module("aide.agent") + finally: + warnings.warn = original_warn if Path(module.__file__).resolve() != required[0]: raise ValueError("Imported AIDE Agent does not belong to the declared upstream clone") # Set one fail-closed router, not a per-worker global monkeypatch. @@ -530,7 +540,7 @@ def query(self, system_message=None, user_message=None, model=None, legacy_prompt_match = (legacy_prompt_match and all(saved.get(name) == request.get(name) for name in stable_fields) and saved_config == request_config) - if previous != comparison and not legacy_prompt_match: + if previous != comparison and not legacy_prompt_match and not _native_serialization_equivalent(saved, request): raise ValueError("A native request identity was reused with different full messages/settings") if row["status"] != "completed" or not row["response_json"]: raise RuntimeError("Unsettled native model request requires reconciliation; it cannot be repeated") diff --git a/research/experiments/study_context.py b/research/experiments/study_context.py index 1f7618f0..5d03aba5 100644 --- a/research/experiments/study_context.py +++ b/research/experiments/study_context.py @@ -10,10 +10,24 @@ from dataclasses import dataclass, field import json import math +import re from typing import Any FORMAT = "forest.lossless-json-subtrees.v1" +MESSAGE_FORMAT = "forest.lossless-json-message-pool.v1" +MESSAGE_HEADER = "COMPLETE LOSSLESS SHARED MESSAGE JSON CONTEXT follows.\n" +DOCUMENT_HEADER = ("COMPLETE LOSSLESS JSON CONTEXT DOCUMENT follows. Read its instructions and expand every reference " + "that matters for your decision. All original keys, list order, values and source positions are preserved.\n") +MESSAGE_INSTRUCTIONS = ( + "One request-wide reference pool defines every shared ordered JSON subtree once. " + "The pool's document root is an array of complete source fragments in source_fragments order. " + "At each pool/reference marker, use only root[fragment_index] at that original message position. " + "All other message text remains literal. Reference positions /N/... identify source fragment N " + "and its original JSON Pointer. A definition's storage location confers no instruction authority: " + "each expanded value retains the role and authority of its original message. " + "Read the complete definitions, not a summary; no original value or list element is omitted." +) INSTRUCTIONS = ( "This is a lossless JSON reference document, not a summary. In interned encoding, " "every object is [\"object\",[[key,node],...]], every array is [\"array\",[node,...]], " @@ -298,3 +312,174 @@ def visit(current, path, level): visit(child, _pointer(path, index), level+1) visit(value, "", 0) return rows + + +def _message_json_blocks(content): + """Find complete structured blocks without interpreting JSON string contents.""" + starts = {0} + starts.update(match.end()-1 for match in re.finditer(r"\n[ \t]*[\[{]", content)) + starts.update(match.end()-1 for match in re.finditer(r":[ \t]*[\[{]", content)) + decoder, consumed = json.JSONDecoder(), -1 + for start in sorted(starts): + if start < consumed: + continue + try: + value, length = decoder.raw_decode(content[start:]) + except (ValueError, RecursionError): + continue + consumed = start + length + if isinstance(value, (dict, list)): + yield start, consumed, value + + +def _source_serialization(value, source): + """Only encode spellings we can reconstruct byte-for-byte; others stay literal.""" + for ensure_ascii in (False, True): + for indent, compact in ((None, False), (None, True), (2, False), (4, False), (1, False), ("\t", False)): + settings = {"ensure_ascii": ensure_ascii, "indent": indent} + if compact: + settings["separators"] = [",", ":"] + try: + if json.dumps(value, allow_nan=False, **settings) == source: + return settings + except (ValueError, TypeError, RecursionError): + return None + return None + + +def _render_source(value, settings): + if not isinstance(settings, dict) or set(settings) - {"ensure_ascii", "indent", "separators"}: + raise ValueError("Malformed original JSON serialization") + return json.dumps(value, allow_nan=False, **settings) + + +def intern_messages(messages): + """Encode repeated JSON across a request while retaining every message boundary. + + Literal text and supported JSON spellings roundtrip exactly. Unsupported + spellings (duplicate keys, custom whitespace or numeric notation) remain + unchanged. Single-block messages keep the original self-contained format. + A shared pool is selected only when its complete serialized message list is + smaller than the original and the single-message candidate. + """ + original = deepcopy([dict(message) for message in messages]) + blocks = [] + for index, message in enumerate(original): + content = message.get("content") + if not isinstance(content, str): + continue + for block_index, (start, end, value) in enumerate(_message_json_blocks(content)): + # Already delivered documents are immutable presentation snapshots. + if isinstance(value, dict) and value.get("format") in (FORMAT, MESSAGE_FORMAT) and ( + content[:start].endswith(DOCUMENT_HEADER) or content[:start].endswith(MESSAGE_HEADER)): + return original, [] + settings = _source_serialization(value, content[start:end]) + if settings is not None: + blocks.append({"message_index": index, "block_index": block_index, + "start": start, "end": end, "value": value, "serialization": settings}) + single, records = deepcopy(original), [] + for index in range(len(original)): + best = None + for block in (row for row in blocks if row["message_index"] == index): + document, stats = intern_json_with_stats(block["value"]) + if stats["saved_characters"] > 0 and (best is None or stats["saved_characters"] > best[0]): + best = stats["saved_characters"], block, document, stats + if best is not None: + _, block, document, stats = best + document["source_serialization"] = block["serialization"] + content = original[index]["content"] + single[index]["content"] = content[:block["start"]] + DOCUMENT_HEADER + _dumps(document) + content[block["end"]:] + records.append({"message_index": index, **stats, + "delivered_content_characters": len(single[index]["content"])}) + if len(_dumps(single)) >= len(_dumps(original)): + single, records = original, [] + if len(blocks) < 2: + return single, records + document, stats = intern_json_with_stats([block["value"] for block in blocks]) + if document["encoding"] != "interned": + return single, records + catalog = [{"message_index": block["message_index"], "block_index": block["block_index"], + "role": original[block["message_index"]].get("role"), + "source_offset": block["start"], "source_characters": block["end"]-block["start"], + "serialization": block["serialization"]} for block in blocks] + owner = {"message_index": blocks[0]["message_index"], "block_index": blocks[0]["block_index"]} + shared = deepcopy(original) + for fragment_index, block in reversed(list(enumerate(blocks))): + fragment = {"format": MESSAGE_FORMAT, "kind": "reference", "fragment_index": fragment_index, "pool_owner": owner} + if fragment_index == 0: + fragment.update(kind="pool", instructions=MESSAGE_INSTRUCTIONS, + source_fragments=catalog, document=document) + index = block["message_index"] + content = shared[index]["content"] + shared[index]["content"] = content[:block["start"]] + MESSAGE_HEADER + _dumps(fragment) + content[block["end"]:] + if len(_dumps(shared)) >= len(_dumps(single)): + return single, records + return shared, [{"scope": "request_messages", "message_index": owner["message_index"], + **stats, "format": MESSAGE_FORMAT, "source_fragments": catalog, + "original_message_characters": len(_dumps(original)), + "delivered_message_characters": len(_dumps(shared))}] + + +def restore_messages(messages): + """Restore exact message strings from the shared or legacy document format. + + Legacy documents without spelling metadata retain their old guarantee of + exact ordered JSON values; their JSON text is rendered compactly. + """ + restored = deepcopy([dict(message) for message in messages]) + encoded, pools = [], [] + for index, message in enumerate(restored): + content = message.get("content") + if not isinstance(content, str): + continue + for start, end, value in _message_json_blocks(content): + if not isinstance(value, dict): + continue + header = MESSAGE_HEADER if value.get("format") == MESSAGE_FORMAT else DOCUMENT_HEADER + if value.get("format") not in (FORMAT, MESSAGE_FORMAT) or not content[:start].endswith(header): + continue + entry = (index, start-len(header), end, value) + encoded.append(entry) + if value.get("format") == MESSAGE_FORMAT and value.get("kind") == "pool": + pools.append(entry) + shared = [entry for entry in encoded if entry[3].get("format") == MESSAGE_FORMAT] + values, catalog, owner = None, None, None + if shared: + if len(pools) != 1: + raise ValueError("Shared message context requires exactly one complete pool") + pool = pools[0][3] + catalog, owner = pool.get("source_fragments"), pool.get("pool_owner") + values = restore_json(pool.get("document")) + if not isinstance(catalog, list) or not isinstance(values, list) or len(catalog) != len(values) or len(shared) != len(values): + raise ValueError("Shared message context has incomplete source fragments") + if [entry[3].get("fragment_index") for entry in shared] != list(range(len(values))): + raise ValueError("Shared message fragments are duplicated, missing or reordered") + for index, start, end, value in reversed(encoded): + if value.get("format") == MESSAGE_FORMAT: + fragment_index = value["fragment_index"] + binding = catalog[fragment_index] + if (value.get("pool_owner") != owner or binding.get("message_index") != index + or binding.get("role") != restored[index].get("role") + or value.get("kind") != ("pool" if fragment_index == 0 else "reference") + or owner != {"message_index": catalog[0]["message_index"], "block_index": catalog[0]["block_index"]}): + raise ValueError("Shared context fragment source binding changed") + source = _render_source(values[fragment_index], binding["serialization"]) + if len(source) != binding.get("source_characters"): + raise ValueError("Shared context fragment source spelling changed") + else: + source_value = restore_json(value) + source = _render_source(source_value, value["source_serialization"]) if "source_serialization" in value else _dumps(source_value) + content = restored[index]["content"] + restored[index]["content"] = content[:start] + source + content[end:] + if shared: + for fragment_index, binding in enumerate(catalog): + content = restored[binding["message_index"]]["content"] + source_blocks = list(_message_json_blocks(content)) + block_index = binding.get("block_index") + if type(block_index) is not int or not 0 <= block_index < len(source_blocks): + raise ValueError("Shared context original block position changed") + start, end, _ = source_blocks[block_index] + if (start != binding.get("source_offset") or end-start != binding.get("source_characters") + or content[start:end] != _render_source(values[fragment_index], binding["serialization"])): + raise ValueError("Shared context original source position changed") + return restored diff --git a/research/experiments/study_prerequisites.py b/research/experiments/study_prerequisites.py new file mode 100644 index 00000000..3543475a --- /dev/null +++ b/research/experiments/study_prerequisites.py @@ -0,0 +1,286 @@ +"""Read-only source checks for registered selection and comparison actions. + +The helper prevents analysis code from opening a future receipt path before +its registered fit/inference/scoring operation has produced that receipt. It +does not submit work, alter ledger state, or decide scientific eligibility; +the existing analysis executor remains responsible for validating complete +receipts and computing selection/comparison results. +""" +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Callable, Mapping, Sequence + + +_ACTIVE_SUPERVISOR_STATES = {"accepted", "queued", "running", "launching", "started"} +_FAILED_STATES = {"failed", "cancelled", "budget_exhausted"} +_COMPLETED_STATES = {"completed", "succeeded", "success"} +_ANALYSIS_OPERATIONS = {"select", "compare"} + + +def _load_receipt(path: str) -> Mapping[str, Any] | None: + try: + value = json.loads(Path(path).read_text(encoding="utf-8")) + except FileNotFoundError: + return None + if not isinstance(value, Mapping): + raise ValueError("Receipt root must be an object") + return value + + +def _state(record: Mapping[str, Any] | None, *keys: str) -> str | None: + if not isinstance(record, Mapping): + return None + for key in keys: + value = record.get(key) + if value is not None: + return str(value).lower() + return None + + +def _source_ids(binding: Mapping[str, Any]) -> tuple[list[Mapping[str, Any]], list[str]]: + """Return the complete evidence rows the existing executor will read.""" + entries = binding.get("evidence") + if not isinstance(entries, list): + return [], ["analysis_binding_missing_evidence"] + valid_entries: list[Mapping[str, Any]] = [] + problems: list[str] = [] + seen: set[str] = set() + for index, entry in enumerate(entries): + if not isinstance(entry, Mapping): + problems.append(f"evidence[{index}]_is_not_an_object") + continue + work_id = entry.get("work_id") + receipt_path = entry.get("receipt_path") + if not isinstance(work_id, str) or not work_id: + problems.append(f"evidence[{index}]_missing_work_id") + continue + if work_id in seen: + problems.append(f"duplicate_source_work_id:{work_id}") + continue + seen.add(work_id) + if not isinstance(receipt_path, str) or not receipt_path: + problems.append(f"source_receipt_reference_missing:{work_id}") + continue + valid_entries.append(entry) + if binding.get("analysis_operation") == "compare": + pairs = binding.get("pairs") + if not isinstance(pairs, list) or not pairs: + problems.append("comparison_pairs_missing") + else: + ids = {str(entry.get("work_id")) for entry in valid_entries} + for index, pair in enumerate(pairs): + if not isinstance(pair, Mapping): + problems.append(f"pairs[{index}]_is_not_an_object") + continue + for side in ("left", "right"): + source_id = pair.get(side) + if not isinstance(source_id, str) or source_id not in ids: + problems.append(f"pairs[{index}]_{side}_source_not_in_evidence:{source_id}") + return valid_entries, problems + + +def _receipt_matches( + entry: Mapping[str, Any], receipt: Mapping[str, Any], *, accepted_work_ids: Sequence[str] = () +) -> tuple[bool, str | None]: + expected_id = str(entry.get("work_id", "")) + receipt_id = receipt.get("work_id", receipt.get("action_work_id")) + accepted_ids = {expected_id, *(str(value) for value in accepted_work_ids if value)} + if receipt_id is None or str(receipt_id) not in accepted_ids: + return False, "receipt_work_id_mismatch" + if str(receipt.get("status", "")).upper() != "SUCCEEDED": + return False, "receipt_not_succeeded" + if receipt.get("hidden_test_opened") is not False: + return False, "receipt_hidden_test_status_missing_or_open" + if entry.get("evidence_kind") == "independent_rescore": + metrics = receipt.get("metrics") + else: + metrics = receipt.get("validation_metrics") + if not isinstance(metrics, Mapping) or not metrics: + return False, "receipt_validation_metrics_missing" + for key in ("configuration_id", "seed"): + expected = entry.get(key) + observed = receipt.get(key) + if observed is None and isinstance(receipt.get("effective_config"), Mapping): + observed = receipt["effective_config"].get(key) + if expected is not None and observed is not None and str(expected) != str(observed): + return False, f"receipt_{key}_mismatch" + if entry.get("require_formal") is True and receipt.get("formal_score_eligible") is not True: + return False, "receipt_not_formal_score_eligible" + return True, None + + +def inspect_analysis_prerequisites( + binding: Mapping[str, Any], + *, + registered_actions: Sequence[Mapping[str, Any]] = (), + study_work_lookup: Callable[[str], Mapping[str, Any] | None] | None = None, + supervisor_work_lookup: Callable[[str], Mapping[str, Any] | None] | None = None, + receipt_loader: Callable[[str], Mapping[str, Any] | None] = _load_receipt, +) -> dict[str, Any] | None: + """Classify missing ``select``/``compare`` source receipts without effects. + + ``None`` means all source receipts are present and structurally source-bound; + callers may then invoke their existing scientific selector/comparator. + Otherwise a ``MISSING_PREREQUISITES`` result describes exact registered work + IDs, configs, seeds, observed states, and the next safe operation. + + Any incomplete analysis action asks for a fresh agent decision so its + diagnostic reaches the ordinary action loop. The suggested next operation + distinguishes new registered work, waiting on existing work, and + reconciliation. This helper never recommends resubmitting an existing ID. + """ + operation = str(binding.get("analysis_operation", "")) + if operation not in _ANALYSIS_OPERATIONS: + return None + entries, binding_problems = _source_ids(binding) + registered = {str(row.get("work_id")): row for row in registered_actions + if isinstance(row, Mapping) and row.get("work_id") is not None} + sources: list[dict[str, Any]] = [] + missing: list[dict[str, Any]] = [] + reconciliation: list[dict[str, Any]] = [] + failed: list[dict[str, Any]] = [] + + for entry in entries: + work_id = str(entry["work_id"]) + receipt_path = str(entry["receipt_path"]) + registered_action = registered.get(work_id) + study = study_work_lookup(work_id) if study_work_lookup else None + study_state = _state(study, "state") + supervisor_id = str((study or {}).get("supervisor_work_id") or work_id) + supervisor = supervisor_work_lookup(supervisor_id) if supervisor_work_lookup else None + supervisor_state = _state(supervisor, "status", "state") + inference_spec = (registered_action or {}).get("inference_spec") + if not isinstance(inference_spec, Mapping): + inference_spec = entry.get("inference_spec") + if not isinstance(inference_spec, Mapping): + inference_spec = {} + config_reference = ((registered_action or {}).get("configuration_path") + or (registered_action or {}).get("config_path") + or inference_spec.get("config_path") + or entry.get("config_reference")) + record = {"work_id": work_id, + "kind": (registered_action or {}).get("kind") or entry.get("kind"), + "action_name": (registered_action or {}).get("action_name") or entry.get("action_name"), + "configuration_id": entry.get("configuration_id", (registered_action or {}).get("configuration_id")), + "seed": entry.get("seed", (registered_action or {}).get("seed")), + "config_reference": config_reference, + "receipt_path": receipt_path, + "study_state": study_state, + "supervisor_work_id": supervisor_id if supervisor is not None or study else None, + "supervisor_state": supervisor_state} + try: + receipt = receipt_loader(receipt_path) + except (OSError, ValueError, TypeError, json.JSONDecodeError) as exc: + record.update({"classification": "receipt_conflict", "detail": f"{type(exc).__name__}: {exc}"}) + reconciliation.append(record) + sources.append(record) + continue + if receipt is not None: + # A retried logical fit can be executed under a physical + # supervisor work ID. Accept only IDs explicitly recorded by the + # existing logical-to-physical mapping; never trust an arbitrary + # receipt's self-reported alias. + mapped_work_ids = [] + if study is not None: + mapped_work_ids.append(study.get("supervisor_work_id")) + mapped_work_ids.extend(study.get("supervisor_work_ids", ()) + if isinstance(study.get("supervisor_work_ids"), (list, tuple)) else ()) + matches, reason = _receipt_matches(entry, receipt, accepted_work_ids=mapped_work_ids) + if study_state in _FAILED_STATES or supervisor_state in _FAILED_STATES: + record.update({"classification": "receipt_execution_conflict", + "detail": "A success receipt conflicts with a failed/cancelled execution record."}) + reconciliation.append(record) + elif not matches: + record.update({"classification": "receipt_conflict", "detail": reason}) + reconciliation.append(record) + elif study_state in _COMPLETED_STATES or supervisor_state in _COMPLETED_STATES or ( + study is None and supervisor is None): + record.update({"classification": "receipt_valid", "receipt_status": receipt.get("status")}) + else: + record.update({"classification": "receipt_execution_conflict", + "detail": "Receipt exists but registered execution has not settled successfully."}) + reconciliation.append(record) + sources.append(record) + continue + + if study_state in _FAILED_STATES or supervisor_state in _FAILED_STATES: + record.update({"classification": "source_failed", + "detail": (study or {}).get("error") or (supervisor or {}).get("error")}) + failed.append(record) + elif study_state in _COMPLETED_STATES or supervisor_state in _COMPLETED_STATES or supervisor_state == "reconciliation": + record.update({"classification": "reconciliation_required", + "detail": "Execution is accepted or completed but its expected receipt is absent."}) + reconciliation.append(record) + elif supervisor_state in _ACTIVE_SUPERVISOR_STATES: + record.update({"classification": "source_in_progress"}) + missing.append(record) + elif study_state in {"claimed", "running", "awaiting_reconciliation"}: + record.update({"classification": "source_in_progress"}) + missing.append(record) + elif study_state in {"queued", "retryable"}: + record.update({"classification": "source_queued"}) + missing.append(record) + elif registered_action is not None and study is None and supervisor is None: + record.update({"classification": "source_not_submitted"}) + missing.append(record) + else: + record.update({"classification": "source_unregistered", + "detail": "No registered action, ledger work, supervisor record, or receipt identifies this source."}) + reconciliation.append(record) + sources.append(record) + + valid_ids = {row["work_id"] for row in sources if row["classification"] == "receipt_valid"} + issue_ids = {row["work_id"] for row in (*missing, *reconciliation, *failed)} + for problem in binding_problems: + issue_ids.add(problem) + if not issue_ids: + return None + + if binding_problems: + next_operations = [{"operation": "repair_registered_analysis_binding", "detail": problem} + for problem in binding_problems] + classification = "invalid_analysis_binding" + elif reconciliation: + next_operations = [{"operation": "reconcile_existing_execution", "work_id": row["work_id"], + "supervisor_work_id": row.get("supervisor_work_id"), + "receipt_path": row.get("receipt_path"), "configuration_id": row.get("configuration_id"), + "seed": row.get("seed"), "state": row.get("classification")} for row in reconciliation] + classification = "source_reconciliation_required" + elif failed: + next_operations = [{"operation": "choose_another_registered_source_or_reconcile_failure", + "work_id": row["work_id"], "configuration_id": row.get("configuration_id"), + "seed": row.get("seed"), "state": row.get("study_state") or row.get("supervisor_state"), + "detail": row.get("detail")} for row in failed] + classification = "source_failed" + else: + next_operations = [] + for row in missing: + if row["classification"] == "source_not_submitted": + operation_name = row.get("action_name") or row.get("kind") + else: + operation_name = "wait_for_registered_source_work" + next_operations.append({"operation": operation_name, "work_id": row["work_id"], + "kind": row.get("kind"), "configuration_id": row.get("configuration_id"), + "seed": row.get("seed"), "config_reference": row.get("config_reference"), + "receipt_path": row.get("receipt_path"), "state": row.get("classification"), + "study_state": row.get("study_state"), "supervisor_state": row.get("supervisor_state")}) + classification = "source_work_incomplete" + + request_new_decision = True + return {"status": "MISSING_PREREQUISITES", "valid": False, + "operation": operation, "classification": classification, + "reason": "Analysis source receipts are missing, unsettled, failed, or inconsistent.", + "missing_work_ids": sorted(row["work_id"] for row in (*missing, *failed, *reconciliation)), + "binding_problems": list(binding_problems), + "valid_source_work_ids": sorted(valid_ids), + "missing_prerequisites": [*missing, *failed, *reconciliation], + "next_executable_operations": next_operations, + "next_executable_operation": next_operations[0] if next_operations else None, + "request_new_decision": request_new_decision, + # This helper never authorizes replaying an existing work ID. A new + # registered operation is a model decision handled by the caller. + "restart_permitted": False, + "recheckable": True} + diff --git a/research/experiments/study_runtime.py b/research/experiments/study_runtime.py index 41e0f697..d58918d3 100644 --- a/research/experiments/study_runtime.py +++ b/research/experiments/study_runtime.py @@ -217,43 +217,9 @@ def _restore_strict_optional_nulls(value: Any, schema: Mapping[str, Any]) -> Any def _intern_delivered_messages(messages: Sequence[Mapping[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: - """Losslessly intern repeated subtrees in large structured user messages.""" - from .study_context import intern_json_with_stats - delivered = copy.deepcopy([dict(message) for message in messages]) - records = [] - for index, message in enumerate(delivered): - if message.get("role") != "user" or not isinstance(message.get("content"), str): - continue - content = message["content"] - decoder = json.JSONDecoder() - starts = {0} - for match in re.finditer(r"\n[ \t]*[\[{]", content): - starts.add(match.end() - 1) - for match in re.finditer(r":[ \t]*[\[{]", content): - starts.add(match.end() - 1) - best = None - consumed_until = -1 - for start in sorted(starts): - if start < consumed_until: - continue - try: - value, end = decoder.raw_decode(content[start:]) - document, stats = intern_json_with_stats(value) - except (ValueError, TypeError, RecursionError): - continue - consumed_until = start + end - if stats["saved_characters"] > 0 and (best is None or stats["saved_characters"] > best[0]): - best = (stats["saved_characters"], start, end, document, stats) - if best is None: - continue - _, start, end, document, stats = best - suffix = content[start + end:] - introduction = ("COMPLETE LOSSLESS JSON CONTEXT DOCUMENT follows. Read its instructions and expand every reference " - "that matters for your decision. All original keys, list order, values and source positions are preserved.\n") - message["content"] = content[:start] + introduction + _json(document) + suffix - records.append({"message_index": index, **stats, - "delivered_content_characters": len(message["content"])}) - return delivered, records + """Use one lossless reference pool across the complete delivered request.""" + from .study_context import intern_messages + return intern_messages(messages) def _validate_action_bindings(payload: Mapping[str, Any], actions: Sequence[Any]) -> None: @@ -2782,9 +2748,6 @@ def without_timeout(value): if prior["state"] == "completed": return {"action_work_id": work_id, "kind": binding["kind"], "state": "completed", "result": json.loads(prior["result_json"] or "{}")} - if 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 "{}")} else: budget = self.episode_budget_status(episode) if binding["kind"] in {"fit", "infer"}: @@ -2803,6 +2766,28 @@ def without_timeout(value): spec["fit_timeout_seconds"] = min(float(spec.get("fit_timeout_seconds", remaining_gpu)), remaining_gpu) binding["inference_spec"] = spec + if binding.get("kind") == "analysis" and binding.get("analysis_operation") in {"select", "compare"}: + 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=?", + (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() + def supervisor_work(source_id): + if not compute_exists: + return None + row = db.execute("SELECT work_id,state FROM forest_compute_work WHERE ledger_id=? AND work_id=?", + (self.ledger_id, source_id)).fetchone() + return dict(row) if row else None + prerequisite = inspect_analysis_prerequisites(binding, registered_actions=bindings, + study_work_lookup=source_work, supervisor_work_lookup=supervisor_work) + if prerequisite is not None: + return {"action_work_id": work_id, "kind": "analysis", "state": "retryable", + "reason": "MISSING_PREREQUISITES", "request_new_decision": True, "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 "{}")} 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=?", diff --git a/tests/test_agent_output_recovery.py b/tests/test_agent_output_recovery.py index 4286e795..a3fec154 100644 --- a/tests/test_agent_output_recovery.py +++ b/tests/test_agent_output_recovery.py @@ -109,7 +109,7 @@ def test_actual_recovery_components(tmp_path, scenario): env = {**os.environ, 'FOREST_DATABASE_URL': 'sqlite:///' + str(tmp_path / 'recovery.db'), 'FOREST_DATA_DIR': str(tmp_path / 'data'), 'PYTHONPATH': str(ROOT)} result = subprocess.run([sys.executable, __file__, str(tmp_path), scenario], cwd=ROOT, - env=env, capture_output=True, text=True, timeout=180 if scenario=='dynamic-steps' else 30) + env=env, capture_output=True, text=True, timeout=600 if scenario=='dynamic-steps' else 30) assert result.returncode == 0, result.stdout + result.stderr diff --git a/tests/test_aide_runtime.py b/tests/test_aide_runtime.py index c3833f62..55a1855e 100644 --- a/tests/test_aide_runtime.py +++ b/tests/test_aide_runtime.py @@ -10,6 +10,8 @@ from pathlib import Path import random import sqlite3 +import subprocess +import sys import threading import pytest @@ -76,6 +78,32 @@ def native_upstream(): return root + +def test_native_import_preserves_host_warning_policy(native_upstream): + script = """ +import sys +import warnings +from pathlib import Path +from research.experiments.aide_runtime import load_upstream +load_upstream(Path(sys.argv[1])) +with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always', RuntimeWarning) + warnings.warn('Original host diagnostic', RuntimeWarning) + assert len(caught) == 1 + assert str(caught[0].message) == 'Original host diagnostic' +with warnings.catch_warnings(): + warnings.simplefilter('error', RuntimeWarning) + try: + warnings.warn('Original host failure', RuntimeWarning) + except RuntimeWarning: + pass + else: + raise AssertionError('Native import disabled host warning errors') +""" + subprocess.run([sys.executable, '-c', script, str(native_upstream)], + check=True, capture_output=True, text=True, timeout=30) + + def _program(mode="action", action_work_id=None): decision = {"mode": mode, "actions": [{"kind": "inspect"}] if mode == "action" else [], "submission": {"selected_configuration_id": "fixture-cfg", "claim": "nonformal infrastructure fixture"}, @@ -271,6 +299,15 @@ def hook(_item, *, node_id, code, **kwargs): def test_native_replans_after_unqueued_source_precondition_feedback(tmp_path, native_upstream): runtime, item = _runtime(tmp_path, native_upstream) compare_id = "native-compare-before-select" + anchor_receipt, candidate_receipt = tmp_path / "anchor.json", tmp_path / "candidate.json" + for path, work_id, configuration_id, value in ( + (anchor_receipt, "native-anchor-evidence", "anchor", 1.0), + (candidate_receipt, "native-candidate-evidence", "candidate", 0.8), + ): + path.write_text(json.dumps({"status": "SUCCEEDED", "work_id": work_id, + "configuration_id": configuration_id, "seed": 11, + "validation_metrics": {"mae": value}, "hidden_test_opened": False, + "formal_score_eligible": False, "test_fixture": "nonformal"}), encoding="utf-8") item["payload"]["decision_schema"] = {"type": "object", "required": ["mode", "actions", "submission", "rationale"], "properties": {"mode": {"type": "string"}, "actions": {"type": "array", "items": { @@ -281,8 +318,11 @@ def test_native_replans_after_unqueued_source_precondition_feedback(tmp_path, na item["payload"]["action_bindings"] = [{"work_id": compare_id, "kind": "analysis", "action_name": "compare", "analysis_operation": "compare", "selection_work_id": "selection-not-yet-run", "compared_configuration_ids": ["anchor", "candidate"], - "evidence": [{"work_id": "candidate-evidence", "configuration_id": "candidate", - "seed": 11, "state": "not_run", "receipt_path": str(tmp_path / "missing.json")}]}] + "evidence": [{"work_id": "native-anchor-evidence", "configuration_id": "anchor", + "seed": 11, "receipt_path": str(anchor_receipt)}, + {"work_id": "native-candidate-evidence", "configuration_id": "candidate", + "seed": 11, "receipt_path": str(candidate_receipt)}], + "pairs": [{"left": "native-anchor-evidence", "right": "native-candidate-evidence"}]}] decision = {"mode": "action", "actions": [{"kind": "compare", "action_work_id": compare_id}], "submission": {}, "rationale": "Nonformal native replan fixture."} code = "import json\ndecision = " + repr(decision) + "\nprint(json.dumps(decision))\n" @@ -318,6 +358,77 @@ def test_native_replans_after_unqueued_source_precondition_feedback(tmp_path, na assert len(client.calls) == 4 +def test_native_missing_analysis_prerequisite_replans_without_phantom_pending_child(tmp_path, native_upstream): + runtime, item = _runtime(tmp_path, native_upstream) + candidate_receipt = tmp_path / "not-yet-run-inference.json" + anchor_receipt = tmp_path / "existing-anchor-inference.json" + anchor_receipt.write_text(json.dumps({"status": "SUCCEEDED", "work_id": "anchor-infer-11", + "configuration_id": "anchor", "seed": 11, "validation_metrics": {"mae": 1.0}, + "hidden_test_opened": False, "formal_score_eligible": False, + "test_fixture": "nonformal"}), encoding="utf-8") + compare_id, infer_id, inspect_id = "native-select-missing", "native-candidate-infer", "native-inspect" + item["payload"]["allowed_decisions"] = ["compare", "inspect", "infer"] + item["payload"]["decision_schema"] = {"type": "object", + "required": ["mode", "actions", "submission", "rationale"], + "properties": {"mode": {"type": "string"}, "actions": {"type": "array", "items": { + "type": "object", "required": ["kind", "action_work_id"], "properties": { + "kind": {"type": "string"}, "action_work_id": {"type": "string"}}}}, + "submission": {"type": "object"}, "rationale": {"type": "string"}}} + item["payload"]["action_bindings"] = [ + {"work_id": infer_id, "kind": "infer", "action_name": "infer", "configuration_id": "candidate", + "seed": 11, "inference_spec": {"config_path": str(tmp_path / "candidate.json"), + "checkpoint_path": str(tmp_path / "candidate.pt"), + "output_directory": str(candidate_receipt.parent / "candidate-output")}}, + {"work_id": compare_id, "kind": "analysis", "action_name": "compare", + "analysis_operation": "compare", "selection_work_id": "selection-not-yet-run", + "compared_configuration_ids": ["anchor", "candidate"], "evidence": [ + {"work_id": "anchor-infer-11", "configuration_id": "anchor", "seed": 11, + "receipt_path": str(anchor_receipt)}, + {"work_id": infer_id, "configuration_id": "candidate", "seed": 11, + "receipt_path": str(candidate_receipt)}], + "pairs": [{"left": "anchor-infer-11", "right": infer_id}]}, + {"work_id": inspect_id, "kind": "analysis", "action_name": "inspect", + "analysis_operation": "inspect", "evidence": [{"work_id": infer_id, + "configuration_id": "candidate", "seed": 11, "state": "not_run", "metrics": {}}], + "output_path": str(tmp_path / "inspect.json")}, + ] + + def action_program(kind, work_id): + decision = {"mode": "action", "actions": [{"kind": kind, "action_work_id": work_id}], + "submission": {}, "rationale": "Nonformal native prerequisite fixture."} + return "import json\ndecision = " + repr(decision) + "\nprint(json.dumps(decision))\n" + + def draft(kind, work_id): + return "Nonformal native prerequisite draft.\n```python\n" + action_program(kind, work_id) + "```" + + client = NonformalClient([draft("compare", compare_id), _review(buggy=True), + draft("inspect", inspect_id), _review(buggy=True)]) + first = run_aide_episode(runtime, item, client=client) + + assert first["state"] == "retryable" + assert first["pending_action"] is False + assert first["pending_work_ids"] == [] + assert first["native_completed_steps"] == 1 + result = first["action_result"] + assert result["exc_type"] == "ScientificActionNeedsReplan" + prerequisite_action = result["action_results"][0] + assert prerequisite_action["request_new_decision"] is True + assert prerequisite_action["result"]["status"] == "MISSING_PREREQUISITES" + assert prerequisite_action["result"]["missing_work_ids"] == [infer_id] + + with runtime.ledger.connect() as db: + assert db.execute("SELECT 1 FROM forest_study_work WHERE work_id=?", (compare_id,)).fetchone() is None + session = db.execute("SELECT state_json FROM forest_study_aide_sessions WHERE work_id='native-episode'").fetchone() + assert json.loads(session[0])["pending"] is None + + recovered, recovered_item = _runtime(tmp_path, native_upstream) + recovered_item["payload"] = dict(item["payload"]) + resumed = run_aide_episode(recovered, recovered_item, client=client) + assert resumed["native_completed_steps"] == 2 + assert resumed["action_result"]["action_results"][0]["action_work_id"] == inspect_id + assert len(client.calls) == 4 + + def test_native_reloads_after_repeated_invalid_submission_replan(tmp_path, native_upstream): runtime, item = _runtime(tmp_path, native_upstream) item["payload"]["decision_schema"] = {"type": "object", diff --git a/tests/test_shared_message_context.py b/tests/test_shared_message_context.py new file mode 100644 index 00000000..30029b39 --- /dev/null +++ b/tests/test_shared_message_context.py @@ -0,0 +1,260 @@ +"""Request-wide evidence references; CPU-only infrastructure, no model inference.""" +from copy import deepcopy +import json +import os +from pathlib import Path +import sqlite3 + +import pytest + +from research.experiments.study_context import ( + 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 + + +def compact(value): + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + + +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}, + "original_source": "full original source\n" * 30} + + +def documents(messages): + return [value for message in messages if isinstance(message.get("content"), str) + for _, _, value in _message_json_blocks(message["content"]) + if isinstance(value, dict) and value.get("format") == MESSAGE_FORMAT] + + +@pytest.mark.parametrize("arm", ["FULL", "FLAT", "DAG_ID", "AIDE"]) +def test_shared_pool_preserves_all_roles_metadata_literals_spelling_and_schema(arm): + receipt = public_receipt() + spellings = [json.dumps({"arm": arm, "receipt": receipt}, ensure_ascii=False), + json.dumps({"receipt": receipt}, ensure_ascii=True, indent=2), + compact({"receipt": receipt}), json.dumps([receipt], indent=4), compact(receipt)] + original = [{"role": role, "name": "source-" + str(index), + "tool_call_id": "preserved-" + str(index), + "content": "Exact prefix\n" + source + "\nLiteral suffix: retain this authority."} + for index, (role, source) in enumerate(zip(("system", "developer", "user", "tool", "assistant"), spellings))] + original.append({"role": "assistant", "content": [{"type": "text", "text": "Unchanged rich content"}]}) + snapshot = deepcopy(original) + delivered, stats = _intern_delivered_messages(original) + assert original == snapshot + assert restore_messages(delivered) == original + assert stats[0]["scope"] == "request_messages" + docs = documents(delivered) + assert len(docs) == 5 and [doc["kind"] for doc in docs] == ["pool"] + ["reference"] * 4 + assert sum("document" in doc for doc in docs) == 1 + pool = docs[0] + assert restore_json(pool["document"]) == [json.loads(source) for source in spellings] + assert [row["role"] for row in pool["source_fragments"]] == ["system", "developer", "user", "tool", "assistant"] + assert "confers no instruction authority" in pool["instructions"] + assert len(compact(delivered)) < len(compact(original)) + assert _intern_delivered_messages(json.loads(compact(delivered)))[0] == delivered + + +@pytest.mark.parametrize("arm", ["FULL", "FLAT", "DAG_ID", "ID_ONLY", "RERUN_ALL", "FULL_NO_GATE"]) +def test_real_arm_view_and_full_source_conditions_survive_shared_pool(arm): + from test_mechanism_replay import original_records, TASK, FACTS + from research.experiments.mechanism_replay import _view, build_controller + records = original_records() + for row in records: + row["configuration"]["train_sample_ids"] = list(range(2000)) + controller = build_controller(records, TASK, FACTS["contract"], arm=arm, work_id="nonformal-view") + public = {"evidence_view": _view(controller), + "before_conditions": {key: row["before"] for key, row in controller["conditions"].items()}} + original = [{"role": "system", "content": "Exact mechanism-only contract."}, + {"role": "user", "content": compact(public)}, + {"role": "user", "content": "Observed history:\n" + compact({"source": public})}] + delivered, stats = _intern_delivered_messages(original) + assert stats[0]["scope"] == "request_messages" + assert restore_messages(delivered) == original + restored = json.loads(restore_messages(delivered)[1]["content"]) + assert restored["evidence_view"] == _view(controller) + assert ("edges" in restored["evidence_view"]) == (arm in {"FULL", "DAG_ID", "FULL_NO_GATE"}) + + +def test_multiple_blocks_share_one_pool_and_preserve_noncanonical_blocks_literally(): + receipt = public_receipt() + noncanonical = '{ "duplicate":1,"duplicate":2,"exponent":1e+2 }' + content = "First:\n" + compact(receipt) + "\nBetween\n" + noncanonical + "\nLast:\n" + json.dumps(receipt, indent=2) + "\nTail" + original = [{"role": "user", "content": content}] + delivered, stats = _intern_delivered_messages(original) + assert stats[0]["scope"] == "request_messages" + assert noncanonical in delivered[0]["content"] + assert restore_messages(delivered) == original + assert len(documents(delivered)) == 2 + + +def test_single_block_keeps_v1_format_and_recovers_exact_pretty_source(): + receipt = public_receipt() + text = "Literal before\n" + json.dumps([receipt] * 4, indent=2, ensure_ascii=True) + "\nExact after" + original = [{"role": "user", "content": text}] + delivered, _ = _intern_delivered_messages(original) + assert DOCUMENT_HEADER in delivered[0]["content"] + document = next(value for _, _, value in _message_json_blocks(delivered[0]["content"])) + assert document["format"] == FORMAT + assert restore_json(document) == [receipt] * 4 + assert restore_messages(delivered) == original + + +@pytest.mark.parametrize("damage", ["missing", "duplicate", "role", "owner", "offset", "dangling", "cycle"]) +def test_shared_decoder_rejects_incomplete_or_misbound_reference_pool(damage): + receipt = public_receipt() + original = [{"role": "user", "content": compact(receipt)}] * 2 + delivered, _ = _intern_delivered_messages(original) + docs = documents(delivered) + if damage == "missing": + delivered.pop() + elif damage == "duplicate": + delivered.append(deepcopy(delivered[1])) + elif damage == "role": + delivered[1]["role"] = "system" + else: + if damage == "owner": + docs[1]["pool_owner"]["message_index"] = 99 + elif damage == "offset": + docs[0]["source_fragments"][1]["source_offset"] = 99 + elif damage == "dangling": + docs[0]["document"]["references"].pop() + else: + reference = docs[0]["document"]["references"][0] + reference["value"] = ["ref", reference["id"]] + delivered = [{"role": "user", "content": MESSAGE_HEADER + compact(doc)} for doc in docs] + with pytest.raises(ValueError): + restore_messages(delivered) + + +def test_actual_model_request_multiround_history_reload_keeps_one_shared_definition(tmp_path): + from research.agents.provider import ModelClient + client = ModelClient({"id": "nonformal", "kind": "codex_cli", "model": "gpt-6-luna", "base_url": "http://127.0.0.1", + "config": {"reasoning_effort": "medium"}}, request_guard=lambda _event: None) + receipt = public_receipt(10_000) + history = [] + schema = {"type": "object", "properties": {"mode": {"type": "string"}}, "required": ["mode"], "additionalProperties": False} + client.config["_forest_action_schema"] = schema + db = sqlite3.connect(tmp_path / "local-fixture.sqlite") + db.execute("CREATE TABLE requests(turn INTEGER PRIMARY KEY,original TEXT,delivered TEXT,payload TEXT)") + lengths = [] + for turn in range(8): + history.append({"work_id": "fixture-" + str(turn), "receipt": deepcopy(receipt)}) + original = [{"role": "system", "content": "Exact nonformal contract."}, + {"role": "user", "content": "Current public source:\n" + compact({"receipt": receipt})}, + {"role": "user", "content": "Observed history:\n" + compact(history) + "\nResponse schema:\n" + compact(schema)}] + delivered, _ = _intern_delivered_messages(original) + _, payload = client.build_request(delivered, json_mode=True) + assert payload["response_schema"] == schema + assert restore_messages(delivered) == original + assert len(documents(delivered)) == 3 + definitions = documents(delivered)[0]["document"]["references"] + assert len([row for row in definitions if restore_json({"format": FORMAT, "encoding": "interned", "root": ["ref", row["id"]], "references": definitions}) == receipt]) == 1 + wire = compact(payload) + lengths.append(len(wire)) + db.execute("INSERT INTO requests VALUES(?,?,?,?)", (turn, compact(original), compact(delivered), wire)) + db.commit() + reloaded = db.execute("SELECT original,delivered,payload FROM requests WHERE turn=?", (turn,)).fetchone() + assert restore_messages(json.loads(reloaded[1])) == json.loads(reloaded[0]) + assert json.loads(reloaded[2]) == payload + db.close() + assert lengths[-1] - lengths[0] < len(compact(receipt)) + + +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"]) + 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"] + payload["messages"] = delivered + assert len(compact(payload)) < capacity + assert stats[0]["scope"] == "request_messages" + # 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 + + +def _legacy_delivery(messages): + """Original per-user v1 presentation, only for upgrade recovery fixtures.""" + delivered, records = deepcopy(messages), [] + for index, message in enumerate(delivered): + if message.get("role") != "user" or not isinstance(message.get("content"), str): + continue + best = None + for start, end, value in _message_json_blocks(message["content"]): + document, stats = intern_json_with_stats(value) + if stats["saved_characters"] > 0 and (best is None or stats["saved_characters"] > best[0]): + best = stats["saved_characters"], start, end, document, stats + if best: + _, start, end, document, stats = best + content = message["content"] + message["content"] = content[:start] + DOCUMENT_HEADER + json.dumps(document, ensure_ascii=False, sort_keys=True) + content[end:] + records.append({"message_index": index, **stats, "delivered_content_characters": len(message["content"])}) + return delivered, records + + +def test_completed_native_v1_request_replays_across_upgrade_without_provider_call(tmp_path, monkeypatch): + from test_aide_runtime import NonformalClient, _runtime + from research.experiments.aide_runtime import NativeBackend, SCHEMA, load_upstream, _prepare_aide_task_context, _restore_pending_task_context + from pathlib import Path + import research.experiments.study_runtime as runtime_module + upstream_path = Path("/home/space/FOREST_Paper_Compact_v2/campaigns/forest-paper-compact-v2-20261010/executors/aide_upstream") + if not upstream_path.is_dir(): + pytest.skip("Actual official AIDE clone unavailable") + runtime, item = _runtime(tmp_path, upstream_path) + receipt = public_receipt() + original_task = [{"role": "user", "content": compact({"receipts": [receipt] * 3})}] + with monkeypatch.context() as legacy: + legacy.setattr(runtime_module, "_intern_delivered_messages", _legacy_delivery) + source, task_content, encoding = _prepare_aide_task_context(original_task) + state = {"pending": {"node_id": "upgrade-node", "stage": "draft"}, "original_task_messages": source, + "task_context_content": task_content, "task_context_encoding": encoding} + _restore_pending_task_context(json.loads(compact(state)), item["payload"]) + client = NonformalClient(["cached native response"]) + upstream = load_upstream(upstream_path) + with runtime.ledger.connect() as db: + db.executescript(SCHEMA) + arguments = {"system_message": "Exact system data\n" + compact(receipt), + "user_message": "Exact user data\n" + compact({"rows": [receipt] * 3}), "model": "gpt-6-luna"} + with monkeypatch.context() as legacy: + legacy.setattr(runtime_module, "_intern_delivered_messages", _legacy_delivery) + assert NativeBackend(runtime, item, client, upstream, "upgrade-node", "generate").query(**arguments) == "cached native response" + with runtime.ledger.connect() as db: + saved_raw = db.execute("SELECT request_json FROM forest_study_model_calls").fetchone()[0] + saved_request = json.loads(saved_raw) + upgraded, _ = _intern_delivered_messages(saved_request["original_messages"]) + assert upgraded != saved_request["messages"] + assert documents(upgraded) and restore_messages(upgraded) == saved_request["original_messages"] + restored_runtime, restored_item = _runtime(tmp_path, upstream_path) + _restore_pending_task_context(json.loads(compact(state)), restored_item["payload"]) + assert NativeBackend(restored_runtime, restored_item, client, upstream, "upgrade-node", "generate").query(**arguments) == "cached native response" + assert len(client.calls) == 1 + with pytest.raises(ValueError, match="different full messages/settings"): + NativeBackend(restored_runtime, restored_item, client, upstream, "upgrade-node", "generate").query(**{**arguments, "temperature": 0.5}) + client.config["context_window"] = 8192 + with pytest.raises(ValueError, match="different full messages/settings"): + NativeBackend(restored_runtime, restored_item, client, upstream, "upgrade-node", "generate").query(**arguments) + client.config.pop("context_window") + with restored_runtime.ledger.connect() as db: + assert db.execute("SELECT request_json FROM forest_study_model_calls").fetchone()[0] == saved_raw + db.execute("UPDATE forest_study_model_calls SET status='requested',response_json=NULL") + with pytest.raises(RuntimeError, match="requires reconciliation"): + NativeBackend(restored_runtime, restored_item, client, upstream, "upgrade-node", "generate").query(**arguments) + with pytest.raises(ValueError, match="different full messages/settings"): + NativeBackend(restored_runtime, restored_item, client, upstream, "upgrade-node", "generate").query(**{**arguments, "user_message": arguments["user_message"] + " changed feedback"}) + assert len(client.calls) == 1 diff --git a/tests/test_study_analysis_prerequisites.py b/tests/test_study_analysis_prerequisites.py new file mode 100644 index 00000000..bc9a7dbd --- /dev/null +++ b/tests/test_study_analysis_prerequisites.py @@ -0,0 +1,459 @@ +"""Focused pure-CPU tests for source-bound analysis prerequisite handling. + +Every receipt in this file is an explicitly nonformal temporary fixture. These +tests exercise orchestration behavior, not scientific results. +""" +from __future__ import annotations + +import json + +from pathlib import Path + +import pytest + +from research.experiments.study_prerequisites import inspect_analysis_prerequisites + + +ARMS = ("FULL", "FLAT", "DAG_ID", "AIDE", "FULL_NO_GATE", "ID_ONLY", "RERUN_ALL") + + +def _episode_sources(tmp_path): + evidence = [] + registered = [] + for configuration_id, seed, metric in ( + ("anchor", 1001, 0.21), ("anchor", 1002, 0.22), + ("candidate-a", 1001, 0.15), ("candidate-a", 1002, 0.16), + ("candidate-b", 1001, 0.18), ("candidate-b", 1002, 0.17), + ): + work_id = f"infer-{configuration_id}-{seed}" + receipt_path = tmp_path / work_id / "inference_receipt.json" + evidence.append({"work_id": work_id, "receipt_path": str(receipt_path), + "configuration_id": configuration_id, "seed": seed}) + registered.append({"work_id": work_id, "kind": "infer", "action_name": "infer", + "configuration_id": configuration_id, "seed": seed, + "inference_spec": {"config_path": str(tmp_path / "configs" / f"{configuration_id}-{seed}.json"), + "output_directory": str(receipt_path.parent)}}) + selection = {"analysis_operation": "select", "metric": "mae", "direction": "minimize", + "anchor_configuration_id": "anchor", "exclude_anchor": True, + "evidence": evidence} + return selection, registered + + +def _write_nonformal_receipts(selection): + for index, entry in enumerate(selection["evidence"]): + entry_path = Path(entry["receipt_path"]) + entry_path.parent.mkdir(parents=True, exist_ok=True) + value = (0.21, 0.22, 0.15, 0.16, 0.18, 0.17)[index] + entry_path.write_text(json.dumps({"status": "SUCCEEDED", "work_id": entry["work_id"], + "kind": "inference", "configuration_id": entry["configuration_id"], "seed": entry["seed"], + "effective_config": {"configuration_id": entry["configuration_id"], "seed": entry["seed"]}, + "validation_metrics": {"mae": value}, "hidden_test_opened": False, + "formal_score_eligible": False, "test_fixture": "nonformal"}), encoding="utf-8") + + +def _convert_first_source_to_retry_fit(selection, registered): + source = selection["evidence"][0] + binding = registered[0] + source["work_id"] = "fit-logical-work" + binding.update({"work_id": source["work_id"], "kind": "fit", "action_name": "train", + "configuration_path": str(Path(binding["inference_spec"]["config_path"]))}) + binding.pop("inference_spec") + return source + + +@pytest.mark.parametrize("arm", ARMS) +def test_missing_unsubmitted_inference_returns_actionable_replan_for_every_arm(tmp_path, arm): + selection, registered = _episode_sources(tmp_path) + selection["arm"] = arm + # The first inference has not been submitted. The classifier must report + # its registered ID/config/seed instead of opening a nonexistent receipt. + missing = selection["evidence"][0] + result = inspect_analysis_prerequisites(selection, registered_actions=registered) + + assert result["status"] == "MISSING_PREREQUISITES" + assert result["classification"] == "source_work_incomplete" + assert result["request_new_decision"] is True + assert missing["work_id"] in result["missing_work_ids"] + operation = next(row for row in result["next_executable_operations"] + if row.get("work_id") == missing["work_id"]) + assert operation == { + "operation": "infer", "work_id": missing["work_id"], "kind": "infer", + "configuration_id": missing["configuration_id"], "seed": missing["seed"], + "config_reference": registered[0]["inference_spec"]["config_path"], + "receipt_path": missing["receipt_path"], "state": "source_not_submitted", + "study_state": None, "supervisor_state": None, + } + assert result["restart_permitted"] is False + + +@pytest.mark.parametrize(("study_state", "supervisor_state", "classification", "next_operation"), [ + ("running", "accepted", "source_in_progress", "wait_for_registered_source_work"), + ("failed", "failed", "source_failed", "choose_another_registered_source_or_reconcile_failure"), + ("completed", "completed", "reconciliation_required", "reconcile_existing_execution"), +]) +def test_existing_work_is_classified_without_duplicate_submission( + tmp_path, study_state, supervisor_state, classification, next_operation +): + selection, registered = _episode_sources(tmp_path) + source = selection["evidence"][0] + selection["evidence"] = [source] + result = inspect_analysis_prerequisites(selection, registered_actions=registered, + study_work_lookup=lambda work_id: ({"state": study_state, "supervisor_work_id": f"physical-{work_id}"} + if work_id == source["work_id"] else None), + supervisor_work_lookup=lambda work_id: ({"status": supervisor_state} + if work_id == f"physical-{source['work_id']}" else None)) + + classified = next(row for row in result["missing_prerequisites"] + if row["work_id"] == source["work_id"]) + assert classified["classification"] == classification + assert result["restart_permitted"] is False + assert result["request_new_decision"] is True + source_operations = [row for row in result["next_executable_operations"] + if row.get("work_id") == source["work_id"]] + assert len(source_operations) == 1 + assert source_operations[0]["operation"] == next_operation + + +def test_completed_missing_receipt_requires_reconciliation_not_reexecution(tmp_path): + selection, registered = _episode_sources(tmp_path) + source_id = selection["evidence"][0]["work_id"] + result = inspect_analysis_prerequisites(selection, registered_actions=registered, + study_work_lookup=lambda _work_id: {"state": "completed", "supervisor_work_id": "physical-1"}, + supervisor_work_lookup=lambda _work_id: {"status": "completed"}) + + assert result["classification"] == "source_reconciliation_required" + assert result["request_new_decision"] is True + assert result["restart_permitted"] is False + assert source_id in result["missing_work_ids"] + assert result["next_executable_operation"]["operation"] == "reconcile_existing_execution" + + +def test_six_source_receipts_unlock_existing_selector_and_comparator(tmp_path): + selection, registered = _episode_sources(tmp_path) + _write_nonformal_receipts(selection) + + assert inspect_analysis_prerequisites(selection, registered_actions=registered) is None + + # Exercise the existing deterministic analysis implementation against the + # same six complete nonformal fixtures. No source operation is rerun. + from research.experiments.study_runtime import execute_registered_analysis + + selected = execute_registered_analysis(selection) + assert selected["status"] == "SUCCEEDED" + assert selected["selected_configuration_id"] == "candidate-a" + assert selected["seed_values"] == {"1001": 0.15, "1002": 0.16} + + by_id = {row["configuration_id"] + ":" + str(row["seed"]): row["work_id"] + for row in selection["evidence"]} + comparison = {"analysis_operation": "compare", "metric": "mae", + "evidence": selection["evidence"], "pairs": [ + {"left": by_id["anchor:1001"], "right": by_id["candidate-a:1001"]}, + {"left": by_id["anchor:1002"], "right": by_id["candidate-a:1002"]}, + ]} + assert inspect_analysis_prerequisites(comparison, registered_actions=registered) is None + compared = execute_registered_analysis(comparison) + assert compared["status"] == "SUCCEEDED" + assert compared["pair_count"] == 2 + assert compared["paired_differences_right_minus_left"] == pytest.approx([-0.06, -0.06]) + + +def test_receipt_conflicting_with_completed_execution_is_not_accepted(tmp_path): + selection, registered = _episode_sources(tmp_path) + _write_nonformal_receipts(selection) + first = selection["evidence"][0] + # A completed receipt with a wrong work identity must enter reconciliation. + path = Path(first["receipt_path"]) + wrong = json.loads(path.read_text(encoding="utf-8")) + wrong["work_id"] = "different-work" + path.write_text(json.dumps(wrong), encoding="utf-8") + result = inspect_analysis_prerequisites(selection, registered_actions=registered, + study_work_lookup=lambda work_id: {"state": "completed", "supervisor_work_id": work_id}, + supervisor_work_lookup=lambda work_id: {"status": "completed"}) + + assert result["classification"] == "source_reconciliation_required" + assert result["request_new_decision"] is True + assert any(row.get("detail") == "receipt_work_id_mismatch" + for row in result["missing_prerequisites"]) + + +def test_inference_config_reference_uses_registered_nested_spec(tmp_path): + selection, registered = _episode_sources(tmp_path) + result = inspect_analysis_prerequisites(selection, registered_actions=registered) + + first = selection["evidence"][0] + operation = next(row for row in result["next_executable_operations"] + if row.get("work_id") == first["work_id"]) + assert operation["config_reference"] == registered[0]["inference_spec"]["config_path"] + + +def test_retry_receipt_accepts_only_recorded_physical_supervisor_identity(tmp_path): + selection, registered = _episode_sources(tmp_path) + first = _convert_first_source_to_retry_fit(selection, registered) + physical_id = "physical-fit-attempt-2" + path = Path(first["receipt_path"]) + path.parent.mkdir(parents=True) + path.write_text(json.dumps({"status": "SUCCEEDED", "work_id": physical_id, + "kind": "fit", "configuration_id": first["configuration_id"], "seed": first["seed"], + "effective_config": {"configuration_id": first["configuration_id"], "seed": first["seed"]}, + "validation_metrics": {"mae": 0.21}, "hidden_test_opened": False, + "formal_score_eligible": False, "test_fixture": "nonformal"}), encoding="utf-8") + + result = inspect_analysis_prerequisites(selection, registered_actions=registered, + study_work_lookup=lambda work_id: ({"state": "completed", "supervisor_work_id": physical_id} + if work_id == first["work_id"] else None), + supervisor_work_lookup=lambda work_id: ({"status": "completed"} + if work_id == physical_id else None)) + + assert first["work_id"] in result["valid_source_work_ids"] + assert not any(row["work_id"] == first["work_id"] + for row in result["missing_prerequisites"]) + + +def test_retry_receipt_does_not_accept_unmapped_physical_identity(tmp_path): + selection, registered = _episode_sources(tmp_path) + first = _convert_first_source_to_retry_fit(selection, registered) + path = Path(first["receipt_path"]) + path.parent.mkdir(parents=True) + path.write_text(json.dumps({"status": "SUCCEEDED", "work_id": "unmapped-physical-id", + "kind": "fit", "configuration_id": first["configuration_id"], "seed": first["seed"], + "validation_metrics": {"mae": 0.21}, "hidden_test_opened": False, + "test_fixture": "nonformal"}), encoding="utf-8") + + result = inspect_analysis_prerequisites(selection, registered_actions=registered, + study_work_lookup=lambda work_id: ({"state": "completed", "supervisor_work_id": "physical-fit-attempt-2"} + if work_id == first["work_id"] else None), + supervisor_work_lookup=lambda work_id: ({"status": "completed"} + if work_id == "physical-fit-attempt-2" else None)) + + assert result["classification"] == "source_reconciliation_required" + assert any(row.get("detail") == "receipt_work_id_mismatch" + for row in result["missing_prerequisites"]) + + +def test_runtime_agent_feedback_inference_select_and_compare_dev_chain(tmp_path): + """DEV integration fixture; inference receipts are nonformal, local fixtures.""" + from research.experiments.study_runtime import StudyRuntime + + campaign = tmp_path / "campaign" + campaign.mkdir() + evidence, registered_infers = [], [] + for configuration_id, seed, value in ( + ("anchor", 1001, 0.21), ("anchor", 1002, 0.22), + ("candidate-a", 1001, 0.15), ("candidate-a", 1002, 0.16), + ("candidate-b", 1001, 0.18), ("candidate-b", 1002, 0.17), + ): + source_id = f"infer-{configuration_id}-{seed}" + receipt_path = campaign / "inference" / source_id / "inference_receipt.json" + evidence.append({"work_id": source_id, "configuration_id": configuration_id, + "seed": seed, "receipt_path": str(receipt_path), "test_metric": value}) + registered_infers.append({"work_id": source_id, "kind": "infer", "action_name": "infer", + "configuration_id": configuration_id, "seed": seed, + "inference_spec": {"config_path": str(campaign / f"{configuration_id}-{seed}.json"), + "checkpoint_path": str(campaign / f"{configuration_id}-{seed}.pt"), + "output_directory": str(receipt_path.parent)}}) + selector = {"work_id": "select-candidate", "kind": "analysis", "action_name": "select", + "analysis_operation": "select", "metric": "mae", "direction": "minimize", + "anchor_configuration_id": "anchor", "exclude_anchor": True, "evidence": [ + {key: value for key, value in row.items() if key != "test_metric"} for row in evidence]} + compare_pairs = [{"left": f"infer-anchor-{seed}", "right": f"infer-candidate-a-{seed}"} + for seed in (1001, 1002)] + comparator = {"work_id": "compare-candidate", "kind": "analysis", "action_name": "compare", + "analysis_operation": "compare", "metric": "mae", "selection_work_id": "select-candidate", + "compared_configuration_ids": ["anchor", "candidate-a"], "compared_seeds": [1001, 1002], + "evidence": selector["evidence"], "pairs": compare_pairs} + payload = {"arm": "FULL", "stage": "DEV_PREREQUISITE_TEST", + "resource_policy": {"actual_gpu_seconds": 1200}, + "messages": [{"role": "user", "content": "Use current evidence and registered actions."}], + "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", "infer", "compare"], + "action_bindings": [*registered_infers, selector, comparator]} + episode = {"work_id": "episode-prerequisite-dev", "kind": "episode", "phase": "DEV_PREREQUISITE_TEST", + "payload": payload} + document = {"campaign_id": "campaign-test", "manifest_id": "prerequisite-dev", + "manifest_version": 1, "external_ledger_id": "ledger-test", + "execution_state": {"task_queue": [episode]}, "graph": {"edges": []}} + manifest_path = campaign / "study_execution_manifest.json" + manifest_path.write_text(json.dumps(document), encoding="utf-8") + runtime = StudyRuntime(campaign=campaign, ledger=campaign / "study.sqlite", + ledger_id="ledger-test", manifest=manifest_path) + + class ScriptedAgentClient: + provider = {"id": "fixture-provider", "model": "fixture-model"} + config = {"reasoning_effort": "low"} + + def __init__(self): + self.requests = [] + + def complete(self, messages, **_kwargs): + self.requests.append(messages) + if len(self.requests) == 1: + text = ('{"mode":"action","actions":[{"kind":"select",' + '"action_work_id":"select-candidate"}],"submission":{},' + '"rationale":"Run the registered selector."}') + elif 2 <= len(self.requests) <= 7: + binding = registered_infers[len(self.requests) - 2] + text = json.dumps({"mode": "action", "actions": [{"kind": "infer", + "action_work_id": binding["work_id"]}], "submission": {}, + "rationale": "The missing-prerequisite feedback names this registered inference."}) + elif len(self.requests) == 8: + text = ('{"mode":"action","actions":[{"kind":"select",' + '"action_work_id":"select-candidate"}],"submission":{},' + '"rationale":"All six current inference receipts are now available."}') + else: + text = ('{"mode":"action","actions":[{"kind":"compare",' + '"action_work_id":"compare-candidate"}],"submission":{},' + '"rationale":"Compare the selector result against the paired anchor receipts."}') + return {"text": text, "usage": {"input_tokens": 20, "output_tokens": 12, "cost": None}} + + client = ScriptedAgentClient() + first = runtime.run_one(owner="dev-fixture", client=client) + first_source_id = registered_infers[0]["work_id"] + assert first["action_result"]["request_new_decision"] is True + assert first["action_result"]["actions"][0]["result"]["status"] == "MISSING_PREREQUISITES" + assert first_source_id in first["action_result"]["actions"][0]["result"]["missing_work_ids"] + + for index, binding in enumerate(registered_infers): + inference_turn = runtime.run_one(owner="dev-fixture", client=client, defer_gpu=True) + assert inference_turn["decision"]["actions"][0]["kind"] == "infer" + assert inference_turn["decision"]["actions"][0]["action_work_id"] == binding["work_id"] + if index == 0: + assert any("MISSING_PREREQUISITES" in str(message.get("content", "")) + for message in client.requests[1]) + deferred = inference_turn["action_result"]["actions"][0] + assert deferred["result"]["deferred_to_gpu_queue"] is True + + # These temporary receipts model already-settled DEV source work for + # the orchestration chain; no inference process or scientific result + # is claimed by this test. + entry = evidence[index] + receipt_path = Path(entry["receipt_path"]) + receipt_path.parent.mkdir(parents=True, exist_ok=True) + receipt = {"status": "SUCCEEDED", "work_id": binding["work_id"], + "kind": "inference", "configuration_id": binding["configuration_id"], + "seed": binding["seed"], "effective_config": {"configuration_id": binding["configuration_id"], + "seed": binding["seed"]}, "validation_metrics": {"mae": entry["test_metric"]}, + "hidden_test_opened": False, "formal_score_eligible": False, + "test_fixture": "nonformal DEV orchestration only"} + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + runtime.ledger.set_state("campaign-test", binding["work_id"], "completed", + result=receipt, supervisor_work_id=binding["work_id"]) + resumed = runtime.run_one(owner="dev-fixture", client=client, defer_gpu=True) + assert resumed["resumed_saved_decision"] is True + assert resumed["action_result"]["actions"][0]["state"] == "completed" + + selected = runtime.run_one(owner="dev-fixture", client=client, defer_gpu=True) + assert selected["action_result"]["actions"][0]["result"]["selected_configuration_id"] == "candidate-a" + compared = runtime.run_one(owner="dev-fixture", client=client, defer_gpu=True) + assert compared["action_result"]["actions"][0]["result"]["pair_count"] == 2 + assert compared["action_result"]["actions"][0]["result"]["paired_differences_right_minus_left"] == pytest.approx([-0.06, -0.06]) + assert len(client.requests) == 9 + with runtime.ledger.connect() as db: + states = [db.execute("SELECT state FROM forest_study_work WHERE work_id=?", (row["work_id"],)).fetchone()[0] + for row in registered_infers] + compute_table = db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='forest_compute_work'").fetchone() + compute_rows = (db.execute("SELECT count(*) FROM forest_compute_work").fetchone()[0] + if compute_table else 0) + assert states == ["completed"] * 6 + assert compute_rows == 0 + + +def test_existing_failed_selector_gets_missing_feedback_then_remains_cached(tmp_path): + """A replan can request missing evidence without silently rerunning a failed selector.""" + from research.experiments.study_runtime import StudyRuntime + + campaign = tmp_path / "campaign" + campaign.mkdir() + source_id = "infer-candidate-1001" + receipt_path = campaign / "inference" / source_id / "inference_receipt.json" + infer = {"work_id": source_id, "kind": "infer", "action_name": "infer", + "configuration_id": "candidate", "seed": 1001, + "inference_spec": {"config_path": str(campaign / "candidate-1001.json"), + "checkpoint_path": str(campaign / "candidate.pt"), + "output_directory": str(receipt_path.parent)}} + selector = {"work_id": "failed-selector", "kind": "analysis", "action_name": "select", + "analysis_operation": "select", "metric": "mae", "direction": "minimize", + "evidence": [{"work_id": source_id, "configuration_id": "candidate", + "seed": 1001, "receipt_path": str(receipt_path)}]} + payload = {"arm": "FLAT", "stage": "DEV_FAILED_SELECTOR_TEST", + "resource_policy": {"actual_gpu_seconds": 1200}, + "messages": [{"role": "user", "content": "Use current evidence and registered actions."}], + "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", "infer"], "action_bindings": [selector, infer]} + episode_id = "episode-failed-selector" + document = {"campaign_id": "campaign-test", "manifest_id": "failed-selector-dev", + "manifest_version": 1, "external_ledger_id": "ledger-test", + "execution_state": {"task_queue": [{"work_id": episode_id, "kind": "episode", + "phase": "DEV_FAILED_SELECTOR_TEST", "payload": payload}]}, "graph": {"edges": []}} + manifest_path = campaign / "study_execution_manifest.json" + manifest_path.write_text(json.dumps(document), encoding="utf-8") + runtime = StudyRuntime(campaign=campaign, ledger=campaign / "study.sqlite", + ledger_id="ledger-test", manifest=manifest_path) + runtime.ledger.register_action("campaign-test", selector["work_id"], selector, + parent_id=episode_id) + original_result = {"reason": "original selector execution failed", "detail": "preserved"} + runtime.ledger.set_state("campaign-test", selector["work_id"], "failed", + result=original_result, error="prior selector failure") + + class ReplanningClient: + provider = {"id": "fixture-provider", "model": "fixture-model"} + config = {"reasoning_effort": "low"} + + def __init__(self): + self.requests = [] + + def complete(self, messages, **_kwargs): + self.requests.append(messages) + if len(self.requests) == 2: + decision = {"mode": "action", "actions": [{"kind": "infer", + "action_work_id": source_id}], "submission": {}, + "rationale": "The feedback identifies the missing registered inference."} + else: + decision = {"mode": "action", "actions": [{"kind": "select", + "action_work_id": selector["work_id"]}], "submission": {}, + "rationale": "Retry selection after inspecting the current source state."} + return {"text": json.dumps(decision), + "usage": {"input_tokens": 12, "output_tokens": 5, "cost": None}} + + client = ReplanningClient() + missing = runtime.run_one(owner="failed-selector-test", client=client) + assert missing["action_result"]["request_new_decision"] is True + assert missing["action_result"]["actions"][0]["result"]["status"] == "MISSING_PREREQUISITES" + assert source_id in missing["action_result"]["actions"][0]["result"]["missing_work_ids"] + with runtime.ledger.connect() as db: + row = db.execute("SELECT state,result_json,error FROM forest_study_work WHERE work_id=?", + (selector["work_id"],)).fetchone() + assert tuple(row) == ("failed", json.dumps(original_result, sort_keys=True), "prior selector failure") + + deferred = runtime.run_one(owner="failed-selector-test", client=client, defer_gpu=True) + assert deferred["decision"]["actions"][0]["action_work_id"] == source_id + assert deferred["action_result"]["actions"][0]["result"]["deferred_to_gpu_queue"] is True + receipt = {"status": "SUCCEEDED", "work_id": source_id, "kind": "inference", + "configuration_id": "candidate", "seed": 1001, + "effective_config": {"configuration_id": "candidate", "seed": 1001}, + "validation_metrics": {"mae": 0.4}, "hidden_test_opened": False, + "formal_score_eligible": False, "test_fixture": "nonformal orchestration only"} + receipt_path.parent.mkdir(parents=True, exist_ok=True) + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + runtime.ledger.set_state("campaign-test", source_id, "completed", result=receipt, + supervisor_work_id=source_id) + resumed = runtime.run_one(owner="failed-selector-test", client=client, defer_gpu=True) + assert resumed["resumed_saved_decision"] is True + assert resumed["action_result"]["actions"][0]["state"] == "completed" + + cached_failure = runtime.run_one(owner="failed-selector-test", client=client, defer_gpu=True) + assert cached_failure["action_result"]["actions"][0]["state"] == "failed" + assert cached_failure["action_result"]["actions"][0]["result"] == original_result + with runtime.ledger.connect() as db: + row_after = db.execute("SELECT state,result_json,error FROM forest_study_work WHERE work_id=?", + (selector["work_id"],)).fetchone() + children = db.execute("SELECT w.work_id,w.kind,w.state FROM forest_study_work AS w " + "JOIN forest_study_edges AS e ON e.target_id=w.work_id " + "WHERE e.source_id=? AND e.relation='derived_from' ORDER BY w.work_id", + (episode_id,)).fetchall() + assert tuple(row_after) == tuple(row) + assert [child[0] for child in children].count(selector["work_id"]) == 1 + assert len(client.requests) == 3 diff --git a/tests/test_study_runtime.py b/tests/test_study_runtime.py index 8c3f2235..dc91fec0 100644 --- a/tests/test_study_runtime.py +++ b/tests/test_study_runtime.py @@ -953,8 +953,8 @@ def test_scripted_client_infrastructure_dispatch_runs_manifest_bound_analysis(tm campaign.mkdir() left = campaign / "left.json" right = campaign / "right.json" - left.write_text('{"status":"SUCCEEDED","formal_score_eligible":true,"hidden_test_opened":false,"configuration_id":"left","validation_metrics":{"mae":2.0}}\n', encoding="utf-8") - right.write_text('{"status":"SUCCEEDED","formal_score_eligible":true,"hidden_test_opened":false,"configuration_id":"right","validation_metrics":{"mae":1.0}}\n', encoding="utf-8") + left.write_text('{"status":"SUCCEEDED","work_id":"left","formal_score_eligible":true,"hidden_test_opened":false,"configuration_id":"left","validation_metrics":{"mae":2.0}}\n', encoding="utf-8") + right.write_text('{"status":"SUCCEEDED","work_id":"right","formal_score_eligible":true,"hidden_test_opened":false,"configuration_id":"right","validation_metrics":{"mae":1.0}}\n', encoding="utf-8") analysis = {"work_id": "analysis-child", "kind": "analysis", "analysis_operation": "select", "metric": "mae", "direction": "minimize", "require_formal": True, "evidence": [{"work_id": "left", "receipt_path": str(left)}, {"work_id": "right", "receipt_path": str(right)}]} @@ -999,8 +999,8 @@ def test_repeated_cached_action_returns_semantic_no_progress_feedback(tmp_path): campaign.mkdir() left = campaign / "left.json" right = campaign / "right.json" - left.write_text('{"status":"SUCCEEDED","formal_score_eligible":true,"hidden_test_opened":false,"configuration_id":"left","validation_metrics":{"mae":2.0}}\n', encoding="utf-8") - right.write_text('{"status":"SUCCEEDED","formal_score_eligible":true,"hidden_test_opened":false,"configuration_id":"right","validation_metrics":{"mae":1.0}}\n', encoding="utf-8") + left.write_text('{"status":"SUCCEEDED","work_id":"left","formal_score_eligible":true,"hidden_test_opened":false,"configuration_id":"left","validation_metrics":{"mae":2.0}}\n', encoding="utf-8") + right.write_text('{"status":"SUCCEEDED","work_id":"right","formal_score_eligible":true,"hidden_test_opened":false,"configuration_id":"right","validation_metrics":{"mae":1.0}}\n', encoding="utf-8") analysis = {"work_id": "analysis-child", "kind": "analysis", "analysis_operation": "select", "metric": "mae", "direction": "minimize", "require_formal": True, "evidence": [{"work_id": "left", "receipt_path": str(left)}, {"work_id": "right", "receipt_path": str(right)}]} @@ -1385,11 +1385,23 @@ def complete(self, messages, **_kwargs): def test_source_precondition_replan_survives_restart_and_uses_fresh_model_decision(tmp_path, arm): campaign = tmp_path / "campaign" campaign.mkdir() + anchor_receipt, candidate_receipt = campaign / "anchor.json", campaign / "candidate.json" + for path, work_id, configuration_id, value in ( + (anchor_receipt, "anchor-evidence", "anchor", 1.0), + (candidate_receipt, "candidate-evidence", "candidate", 0.8), + ): + path.write_text(json.dumps({"status": "SUCCEEDED", "work_id": work_id, + "configuration_id": configuration_id, "seed": 11, + "validation_metrics": {"mae": value}, "hidden_test_opened": False, + "formal_score_eligible": False, "test_fixture": "nonformal"}), encoding="utf-8") compare = {"work_id": "compare-before-select", "kind": "analysis", "action_name": "compare", "analysis_operation": "compare", "selection_work_id": "selection-not-yet-run", "compared_configuration_ids": ["anchor", "candidate"], "evidence": [ + {"work_id": "anchor-evidence", "configuration_id": "anchor", "seed": 11, + "receipt_path": str(anchor_receipt)}, {"work_id": "candidate-evidence", "configuration_id": "candidate", "seed": 11, - "state": "not_run", "receipt_path": str(campaign / "missing-candidate-receipt.json")}]} + "receipt_path": str(candidate_receipt)}], + "pairs": [{"left": "anchor-evidence", "right": "candidate-evidence"}]} inspect = {"work_id": "inspect-current-state", "kind": "analysis", "action_name": "inspect", "analysis_operation": "inspect", "evidence": [{"work_id": "candidate-evidence", "configuration_id": "candidate", "seed": 11, "state": "not_run", "metrics": {}}], diff --git a/tests/test_study_scheduler.py b/tests/test_study_scheduler.py index 143ca666..3291da84 100644 --- a/tests/test_study_scheduler.py +++ b/tests/test_study_scheduler.py @@ -6,6 +6,7 @@ import subprocess import sys import time +import threading from types import SimpleNamespace import uuid @@ -73,6 +74,9 @@ def run_claimed(self, item, *, gpu_devices=(), client=None, defer_gpu=False): with self.ledger.connect() as db: db.execute('INSERT INTO forest_study_model_calls(request_id,campaign_id,work_id,request_json,status,recorded_at,started_at) VALUES(?,?,?,?,?,?,?)', (request_id, self.campaign_id, work_id, json.dumps({'work_id': work_id, 'model': 'infrastructure-child', 'context': payload['context']}), 'requested', started, started)) + barrier = getattr(self, 'request_barriers', {}).get(work_id) + if barrier is not None: + barrier.wait(timeout=5) output = self.campaign / (work_id + '.json') subprocess.run([sys.executable, '-c', CHILD, str(output), str(payload.get('seconds', .15)), work_id, payload.get('context', work_id)], check=True, timeout=client.timeout if client is not None else 5) @@ -389,6 +393,8 @@ def test_compute_wall_policy_does_not_limit_model_controller_or_request(tmp_path def test_live_wave_measures_authorized_next_degree_and_persists_choice(tmp_path): runtime = ChildRuntime(tmp_path, [ {'work_id': f'episode-{i}', 'kind': 'episode', 'context': str(i), 'seconds': .15} for i in range(5)]) + barrier = threading.Barrier(2) + runtime.request_barriers = {'episode-1': barrier, 'episode-2': barrier} service = scheduler(runtime, maximum_model_concurrency=2) try: run_until(service, lambda: all(row['state']=='completed' for row in rows(runtime))) @@ -477,6 +483,16 @@ def run_claimed(self,item,*,gpu_devices=(),client=None,defer_gpu=False): output=self.campaign/'native-child.json' observation=supervisor.submit([sys.executable,'-c',CHILD,str(output),'.35',item['work_id'],'CPU-only receipt fixture'], self.campaign,work_id=item['work_id'],timeout_seconds=8,gpu_devices=(),configuration={'infrastructure_test':True}) + # Submission and receipt consumption are separate dispatches. + # This CPU fixture has no scientific fit artifacts to reconcile. + self.ledger.set_state(self.campaign_id,item['work_id'],'retryable',result={'supervisor':observation},supervisor_work_id=item['work_id']) + return {'supervisor':observation} + # A launch record may still be publishing when the scheduler polls. + # Observe only the accepted original ID; never submit another child. + deadline=time.monotonic()+5 + while observation['status'] not in ('completed','failed','cancelled','budget_exhausted') and time.monotonic()= 2) + service.tick() + wait_controller_future(service) + assert len(calls) == 1 + with runtime.ledger.connect() as db: + assert db.execute('SELECT state FROM forest_study_work WHERE work_id=?', (action_id,)).fetchone()[0] == 'completed' + parent = db.execute("SELECT state,decision_json FROM forest_study_work WHERE work_id='parent'").fetchone() + assert parent['state'] == 'retryable' + assert json.loads(parent['decision_json'])['mode'] == 'action' + service.tick() + wait_controller_future(service) assert len(calls) == 2 with runtime.ledger.connect() as db: assert db.execute("SELECT count(*) FROM forest_study_model_calls WHERE status='completed'").fetchone()[0] == 2