v62: resume income analysis after protected serving startup, bound research retries and scope live evidence

Author: Mantiee

SYNTA_RECOVERY.md

@@ -115,3 +115,42 @@ Inspect `research/income-work/status.json`, its source-linked dossiers,
 `research/paper/fee-source-status.json` for measured execution. Feasibility
 estimates are neither real income nor training labels. Actual master training
 and promotion still need independently verified outcomes and all quality gates.
+
+## v62: recover the supplied startup race without discarding evidence
+
+The uploaded v61 report shows an income analysis hitting `/proc/515286/stat`
+after that old server exited. Source acquisition had succeeded. This is a
+serving readiness failure, not a failed income experiment.
+
+- `income_work.accepted_master` waits for current-run startup readiness and
+  requires the protected process/model receipt before generation. Missing process,
+  receipt or endpoint becomes an explicit deferral. An identity mismatch remains
+  a rejection. No candidate or disabled Windows RTX is substituted.
+- `income_work.execute` resumes a pending dossier under the same goal for up to
+  24 hours without refetching or counting its source twice. Startup retries use
+  30-120-second backoff for six attempts, then the normal ten-minute period.
+  Analysis attempts are separately limited to three; failures stay archived.
+  Goal changes during generation block registration under the new goal.
+- `drones.finish` changes only the next due time for these income retries;
+  the configured periodic interval is retained. The waiting job releases its
+  worker slot rather than sleeping inside it.
+- `research_policy.apply` cannot expand the bounded resident client's 1024-token
+  ceiling or enable its thinking mode. Ordinary research policy remains separate.
+- The research audit rejects actual-labor income comparisons against simulated
+  paper returns, recognizes the supplied Prolific/microtask variants, and flags
+  the same advisory test under paraphrased hypotheses. Lexical checking is not
+  semantic proof. Researchers receive the last four rejected tests and their
+  reasons so they can revise them. Historical migration cannot replace the
+  latest live audit.
+- New worker notes carry the originating mission run. Prior-run and migrated
+  historical notes remain archived and are excluded from the live ideas list.
+- Cached official baselines publish finished progress for the current run.
+  Complete current-run reports take precedence over stale global counters;
+  previous-run counters are presented as history, never current computation.
+- `mission-report` includes owned CPU dispatch gates and the archived fee-source
+  blocker. Dispatch is goal-attributed and rechecks promptly after a goal change.
+  No training is forced with unverified labels or inadequate independent splits.
+
+This release does not certify fees unavailable from the actual retrieved page,
+income, demand, novelty, CUDA utilization or browser behavior on the operator's
+machines. The existing Windows CPU limit and disabled RTX preference are retained.

SYNTA_REQUIREMENTS_AUDIT.md

@@ -1,4 +1,4 @@
-# Synta: requirement and verification audit, v61
+# Synta: requirement and verification audit, v62
 
 This audits the visible conversation. "Implemented" means code exists and the
 listed checks pass, not that it has run successfully on the operator's computers.
@@ -100,3 +100,19 @@ and registered proposal when analysis succeeds. A blocked source/analysis has
 its own receipt. This is not autonomous financial execution, a profitable strategy,
 a security scan, a sale or universal task learning. See SYNTA_RECOVERY.md for
 exact modules and runtime artifact paths.
+
+## v62 supplied-report regressions
+
+Protected native startup defers income analysis without discarding the source;
+pending analysis resumes without counting another fetch. Missing PID, startup
+backoff, unchanged scheduling interval, bounded policy, invalid-analysis rotation,
+paper-vs-earned-income rejection, advisory-test repeats and benchmark run scoping
+have deterministic regression tests. Historical audits and notes are separated
+from current work. CPU training admission and fee-page blockers are visible.
+These are code/transport verifications, not a claim of verified income or hardware
+execution. See the v62 section of SYNTA_RECOVERY.md for exact limits.
+
+Final v62 verification: 1098 tests passed, 65 skipped, one existing PEFT warning;
+no deselections. Ruff for Python/scripts/tests excluding the unchanged notebook,
+upgrade shell syntax and diff whitespace checks pass. No Windows, CUDA or
+operator-browser runtime acceptance is asserted by this result.

pyproject.toml

@@ -3,7 +3,7 @@ name = "rlms"
 authors = [
     {name = "Alex Zhang", email = "altzhang@mit.edu"},
 ]
-version = "0.1.3+v100.61"
+version = "0.1.3+v100.62"
 description = "Recursive Language Models."
 readme = "README.md"
 license = "MIT"

rlm/v100/agent.py

@@ -55,6 +55,9 @@ def native_turn(
     )
     prompt_tokens = client.count_text(template["prompt"], parse_special=True) + 1
     max_tokens = client.sampling_args.get("max_tokens", 512)
+    host_ceiling = getattr(client, "research_token_ceiling", None)
+    if host_ceiling is not None:
+        max_tokens = min(max_tokens, host_ceiling)
     sampling = {
         key: client.sampling_args[key]
         for key in ("temperature", "seed", "top_p", "top_k")
@@ -82,6 +85,8 @@ def native_turn(
     ):
         raise ValueError("Research retry output limit must be 1-8192 tokens")
     ceiling = min(retry_output_limit or max_tokens, client.context_window - prompt_tokens)
+    if host_ceiling is not None:
+        ceiling = min(ceiling, host_ceiling)
     attempts, completion_tokens = 0, 0
     while True:
         attempts += 1

rlm/v100/dashboard_snapshot.py

@@ -12,6 +12,11 @@ def supplement(root: Path, mission: dict, report: dict, errors: list) -> dict:
     from rlm.v100.planning import read
 
     result = dict(report)
+    from rlm.v100.public_benchmarks import progress_for_run
+
+    benchmark = progress_for_run(root, Path(mission["run"]) if mission.get("run") else None)
+    if benchmark.get("report") or benchmark.get("historical") or "official_benchmark" in result:
+        result["official_benchmark"] = benchmark
     for key, filename in (
         ("drones", "research/drones-status.json"),
         ("external_compute", "research/compute-status.json"),

rlm/v100/drones.py

@@ -220,6 +220,10 @@ def finish(root: Path, identity: str, value: dict) -> None:
     with connect(root) as db:
         row = db.execute("SELECT * FROM jobs WHERE id=?", (identity,)).fetchone()
         interval = row["interval"]
+        delay = interval
+        retry = value.get("retry_after_seconds")
+        if row["kind"] == "income" and interval and type(retry) is int and 15 <= retry <= 600:
+            delay = retry
         state = (
             "cancelled" if row["state"] == "cancelled" else "queued" if interval else "completed"
         )
@@ -254,12 +258,13 @@ def finish(root: Path, identity: str, value: dict) -> None:
             job_state["validation_error"] = value.get("detail", "Local validation failed")
             if retries < 3:
                 state, interval = "queued", 60
+                delay = interval
             else:
                 job_state["state"] = "validation-deferred"
             atomic_json(source, job_state)
         db.execute(
             "UPDATE jobs SET state=?,due=?,updated=?,result=? WHERE id=?",
-            (state, time.time() + interval, time.time(), json.dumps(value), identity),
+            (state, time.time() + delay, time.time(), json.dumps(value), identity),
         )
     ActivityLog(root, row["branch"], "drone").write(
         "steps", "drone-result", {"id": identity, **value}

rlm/v100/goal_compute.py

@@ -113,6 +113,9 @@ def propose(root: Path, branch: str = "A") -> dict:
 
 def tick(root: Path) -> None:
     from rlm.v100.distributed_compute import inspect
+    from rlm.v100.goals import load_goal
+
+    goal_id = (load_goal(root) or {}).get("id")
 
     preferences = root / "research/user-preferences.json"
     if not preferences.exists() or not json.loads(preferences.read_text()).get(
@@ -128,10 +131,16 @@ def tick(root: Path) -> None:
         except BlockingIOError:
             return
         previous = json.loads(status.read_text()) if status.exists() else {}
-        if time.time() < previous.get("next_poll", 0):
+        if previous.get("goal_id") == goal_id and time.time() < previous.get("next_poll", 0):
             return
         atomic_json(
-            status, {"state": "checking", "updated": time.time(), "next_poll": time.time() + 300}
+            status,
+            {
+                "state": "checking",
+                "goal_id": goal_id,
+                "updated": time.time(),
+                "next_poll": time.time() + 300,
+            },
         )
         try:
             resources = inspect(root)
@@ -157,7 +166,10 @@ def tick(root: Path) -> None:
                     result["state"] = "no-new-data"
         except (OSError, ValueError, RuntimeError, KeyError) as error:
             result = {"state": "blocked", "reason": str(error)[:400]}
-        atomic_json(status, {**result, "updated": time.time(), "next_poll": time.time() + 300})
+        atomic_json(
+            status,
+            {**result, "goal_id": goal_id, "updated": time.time(), "next_poll": time.time() + 300},
+        )
 
 
 def verify_job(root: Path, job: dict) -> None:

rlm/v100/income_work.py

@@ -27,6 +27,10 @@
 )
 
 
+class MasterNotReady(RuntimeError):
+    """Transient serving startup, shutdown or unavailable protected receipt."""
+
+
 def accepted_master(root: Path, branch: str):
     """Reuse a protected serving master, never launch a model or use a candidate."""
     from rlm.v100.competition import helper_client
@@ -36,22 +40,37 @@ def accepted_master(root: Path, branch: str):
     run = Path(active["run"]).resolve()
     if not run.is_relative_to((root / "research/mission").resolve()):
         raise ValueError("Mission escaped its owned directory")
-    state = json.loads((run / "status.json").read_text())
+    status_path = run / "status.json"
+    state = json.loads(status_path.read_text()) if status_path.exists() else {}
     paths = [run / "serving-active.json", run / "learning/live.json"]
     paths.append(run / f"profile-{state.get('context_window')}.json")
     path = next((candidate for candidate in paths if candidate.is_file()), None)
     if path is None:
-        raise RuntimeError("Accepted master profile not ready; collected evidence retained")
+        raise MasterNotReady("Accepted master profile not ready; collected evidence retained")
+    startup = run / f"server-{state.get('context_window')}.startup.json"
+    # The initial server is stopped when continuous learning starts its own
+    # protected server. Its old startup file must not block the live profile.
+    if (
+        path.name.startswith("profile-")
+        and startup.exists()
+        and json.loads(startup.read_text()).get("stage") != "ready"
+    ):
+        raise MasterNotReady("Accepted V100 serving startup not ready; evidence retained")
     profile = load_profile(path, root)
     client = helper_client(profile, root, branch)
     client.timeout = min(10, client.timeout)
     # Full protected receipt and endpoint check precedes any generation. A
     # candidate temporarily occupying the same endpoint is never used.
-    assert_served_expert(client, profile, root)
+    try:
+        assert_served_expert(client, profile, root)
+    except (FileNotFoundError, ProcessLookupError, requests.RequestException) as error:
+        raise MasterNotReady("Accepted serving process or endpoint not ready") from error
     client.timeout = 40
     client.request_deadline = time.monotonic() + 60
     client.enable_thinking = False
     client.sampling_args = dict(client.sampling_args) | {"max_tokens": 1024}
+    client.research_token_ceiling = 1024
+    client.research_thinking_allowed = False
     client.activity_actor = "income-research"
     return client
 
@@ -86,52 +105,90 @@ def execute(root: Path, job: dict) -> dict:
     path = directory / "status.json"
     previous = json.loads(path.read_text()) if path.exists() else {}
     same_goal = previous.get("goal_id") == goal["id"]
+    pending = None
+    if same_goal and previous.get("dossier"):
+        candidate = Path(previous["dossier"]).resolve()
+        if candidate.is_relative_to((directory / "dossiers").resolve()) and candidate.is_file():
+            saved = json.loads(candidate.read_text())
+            if (
+                saved.get("goal_id") == goal["id"]
+                and saved.get("analysis") is None
+                and previous.get("state")
+                in (
+                    "waiting for accepted master",
+                    "evidence collected; analysis blocked",
+                )
+                and saved.get("analysis_attempts", 0) < 3
+                and 0 <= time.time() - saved["source"]["acquired"] <= 86400
+            ):
+                pending = saved
     index = previous.get("next_source", 0) if same_goal else 0
     domain, url = STARTING_SOURCES[index % len(STARTING_SOURCES)]
+    if pending:
+        domain, url = previous["domain"], previous["url"]
     value = {
         "goal_id": goal["id"],
         "job_id": job["id"],
         "updated": time.time(),
         "state": "collecting primary terms",
         "domain": domain,
         "url": url,
-        "next_source": index + 1,
+        "next_source": index if pending else index + 1,
         "sources_collected": previous.get("sources_collected", 0) if same_goal else 0,
         "proposals_registered": previous.get("proposals_registered", 0) if same_goal else 0,
         "actual_income_pln": None,
         "scope": "Public-terms research and preparation only; no account, application, sale, security scan or payment executed",
     }
+    if pending:
+        value["readiness_attempts"] = previous.get("readiness_attempts", 0)
     atomic_json(path, value)
     journal = ActivityLog(root, job["branch"], "income-research")
     step = journal.write(
         "tools",
         "tool-start",
-        {"tool": "observe_goal_source", "arguments": {"url": url, "field": ""}},
+        {
+            "tool": "read_income_dossier" if pending else "observe_goal_source",
+            "arguments": {"path": previous["dossier"]} if pending else {"url": url, "field": ""},
+        },
     )
     try:
-        source = observe(root, url)
+        source = pending["source"] if pending else observe(root, url)
     except (ValueError, OSError, RuntimeError, requests.RequestException) as error:
         value.update(state="source unavailable", error=str(error)[:400], updated=time.time())
         atomic_json(path, value)
         journal.write("errors", "source-unavailable", value, step_id=step)
         return {**value, "status": "failed"}
     journal.write(
-        "tools", "tool-result", {"tool": "observe_goal_source", "result": source}, step_id=step
+        "tools",
+        "tool-result",
+        {
+            "tool": "read_income_dossier" if pending else "observe_goal_source",
+            "result": source,
+            "scope": "Archived evidence reused; no new fetch" if pending else "New source fetch",
+        },
+        step_id=step,
     )
     value.update(
         state="evidence collected; analysis pending",
         evidence_id=source["id"],
-        sources_collected=value["sources_collected"] + 1,
+        sources_collected=value["sources_collected"] + (0 if pending else 1),
     )
     dossier = directory / "dossiers" / (source["id"] + ".json")
-    atomic_json(
-        dossier,
-        {"goal_id": goal["id"], "source": source, "analysis": None, "scope": value["scope"]},
-    )
+    saved = pending or {
+        "goal_id": goal["id"],
+        "source": source,
+        "analysis": None,
+        "scope": value["scope"],
+        "analysis_attempts": 0,
+    }
+    atomic_json(dossier, saved)
     value["dossier"] = str(dossier)
     atomic_json(path, value)
+    attempts_before = saved.get("analysis_attempts", 0)
     try:
         client = accepted_master(root, job["branch"])
+        saved["analysis_attempts"] = saved.get("analysis_attempts", 0) + 1
+        atomic_json(dossier, saved)
         response = native_turn(
             client,
             [
@@ -165,6 +222,8 @@ def execute(root: Path, job: dict) -> dict:
             raise ValueError(
                 "No accepted work or payment receipt; conservative income lower bound must be zero"
             )
+        if (load_goal(root) or {}).get("id") != goal["id"]:
+            raise ValueError("Goal changed during analysis; proposal retained but not registered")
         row = register(root, job["branch"], specification)
         atomic_json(
             dossier,
@@ -175,8 +234,22 @@ def execute(root: Path, job: dict) -> dict:
             candidate_id=row["id"],
             proposals_registered=value["proposals_registered"] + 1,
         )
+    except MasterNotReady as error:
+        attempts = value.get("readiness_attempts", 0) + 1
+        value.update(
+            state="waiting for accepted master",
+            error=str(error)[:400],
+            readiness_attempts=attempts,
+            retry_after_seconds=min(120, 15 * 2 ** min(attempts, 3)) if attempts <= 6 else 600,
+        )
+        journal.write("steps", "income-analysis-deferred", value)
     except (ValueError, OSError, RuntimeError, requests.RequestException) as error:
+        saved["analysis_attempts"] = max(attempts_before + 1, saved.get("analysis_attempts", 0))
+        saved["analysis_error"] = str(error)[:400]
+        atomic_json(dossier, saved)
         value.update(state="evidence collected; analysis blocked", error=str(error)[:400])
+        if saved.get("analysis_attempts", 0) < 3:
+            value["retry_after_seconds"] = 120
         journal.write("errors", "income-analysis-blocked", value)
     value["updated"] = time.time()
     atomic_json(path, value)

rlm/v100/mission.py

@@ -363,7 +363,13 @@ def run(root: Path, profile: dict, directory: Path) -> None:
                     try:
                         cached_public = reusable_public(root, serving)
                         if cached_public:
-                            atomic_json(public_path, json.loads(cached_public.read_text()))
+                            from rlm.v100.public_benchmarks import write_progress
+
+                            reused = json.loads(cached_public.read_text())
+                            atomic_json(public_path, reused)
+                            write_progress(
+                                root, public_path, reused, len(reused["cases"]), "finished"
+                            )
                             atomic_json(
                                 directory / "public-baseline-reused.json",
                                 {"source": str(cached_public), "sha256": file_hash(cached_public)},