v71: execute resource tuning, distribute owned CPU work and reject false chat completion

Route optimization commands and explicit follow-ups to bounded host policy receipts. Respect affinity/quota/RAM, assign distinct CPU slots to parallel Python experiments and reserve executor capacity for observation. Retain Windows adaptive guards and accepted GPU settings. Report missing result archives explicitly and store failed/empty work as failure. Add targeted profile-preserving updater applying the policy.

Validation: full suite 1252 passed, 65 skipped with one existing PEFT warning; 107 focused tests passed after final receipt display change and added test. Ruff rlm/tests/tools excluding notebooks, bash syntax and git diff checks passed. Hardware throughput not measured.

Author: Mantiee

README.md

@@ -1,8 +1,10 @@
 
 ---
 
-Synta v70 avoids repeated full artifact reads within the same process, detects changed file
-identities, and bounds native startup verification. See [SYNTA_V70.md](SYNTA_V70.md).
+Synta v71 executes resource optimization directly, keeps follow-ups on the operator task,
+and rejects empty/failed work as successful chat completion. CPU concurrency adapts to
+affinity, quota and RAM. See [SYNTA_V71.md](SYNTA_V71.md).
+The startup verification fix from [v70](SYNTA_V70.md) is retained.
 
 > **V100 fork, v100.14:** the local workflow uses your own llama.cpp server, no paid API.
 > v100.14 fixes CPU research request budgets and one-slot job scheduling, trims

SYNTA_V71.md

@@ -0,0 +1,43 @@
+# Synta v71: resource optimization and truthful execution status
+
+An operator request to optimize CPU/RAM/resources now executes a host-owned operation,
+rather than asking the model to select a backtest. This includes mixed questions and
+commands such as “CPU Debian or Windows? Increase usage, do it.” Explicit follow-ups
+such as “do all of it” or “error, continue” resolve prior operator messages, not assistant
+claims. An intervening unrelated request ends that binding.
+
+The controller writes `research/state/resource-policy.json`, archives the previous policy
+and receipts under `research/resource-optimization`, and reports the measured current
+admission. Debian admits up to four CPU/network jobs, bounded by affinity and cgroup v2
+CPU quotas, two threads per slot and a CPU reserved when possible. Concurrent owned Python experiments receive distinct CPU affinity slots instead of
+all being pinned to the first two CPUs. Each result and live drone status records its
+assigned CPUs. The tuned policy
+reserves 4 GiB available RAM plus 3 GiB per slot. Host CPU at 65% reduces concurrency
+and 85% stops new admission. Existing bounded jobs can finish. One model job remains
+allowed; six executor workers accommodate four CPU tasks, the model and the observer.
+
+Windows affinity/RAM and adaptive RTX guards are retained. No board power cap, GPU
+clock or quality gate is changed. With a fresh idle owned CPU worker, the controller can
+queue the existing verified goal trial under the shared dispatch lock. Its existing label,
+source split and independent evaluation requirements still apply. Missing labels report
+a blocker instead of launching synthetic busywork. Existing owned jobs retain priority.
+No measured speedup, profit, continuous training or crash immunity is claimed.
+
+Missing archived tool results return `archive_not_found` with the requested digest and
+recovery guidance. No missing content is invented. A chat whose tools all failed is
+stored as `failed` with its response and receipts; an empty reply is rejected. Successful
+partial work retains its receipts. No automatic rerun of mutations is introduced.
+
+## Install
+
+Use `tools/update-synta-resources.sh` with this release's pinned commit on Debian.
+It accepts v67-v71, requires the owned user service, retains the current accepted/live
+profile and goals, checks for exclusive experiments, stops the owned mission, installs
+v71, restarts it through the existing supervisor and applies the resource policy. It does
+not reinstall Windows, reset the ledger or restart Tailscale routing. Inspect the printed
+resource receipt for actual slot counts and Windows blockers. Windows does not need
+another installer for this change.
+
+Hardware throughput and responsiveness must be measured on the deployment. This
+release does not promise that every arbitrary optimization is implemented, nor does
+CPU concurrency alone improve GPU inference. Accepted model settings are unchanged.

pyproject.toml

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

rlm/v100/chat_resources.py

@@ -28,14 +28,16 @@
 
 def instruction(message: str) -> bool:
     text = normalized(message)
-    if "?" in text or re.match(r"^(?:czy|jak|why|how|is|are)\b", text):
+    action = re.search(r"uzyw|wykorzyst|przydz|deleg|udostep|wspoldziel|shared", text)
+    if not action or (
+        re.match(r"^(?:czy|jak|why|how|is|are)\b", text)
+        and not re.search(r"zrob|wykonaj|allocate|use", text)
+    ):
         return False
     if re.search(r"dashboard|\bhtml\b|pulpit|(?:ustaw|zmien)\w*\s+(?:cel|plan)", text):
         return False
     resource = re.findall(r"\b(?:rtx|windows\w*|cpu|ram\w*|dysk\w*|worker\w*|dron\w*)\b", text)
-    return len(set(resource)) >= 2 and bool(
-        re.search(r"uzyw|wykorzyst|przydz|deleg|udostep|wspoldziel|shared", text)
-    )
+    return len(set(resource)) >= 2
 
 
 def filter_actions(actions: list[dict]) -> tuple[list[dict], list[dict]]:
@@ -50,8 +52,10 @@ def status(root: Path) -> dict:
     from rlm.v100.distributed_compute import inspect
     from rlm.v100.drones import inspect as drone_status
     from rlm.v100.remote_helper import remote_profile, selected_helper
+    from rlm.v100.resource_budget import local_budget
 
     value = {
+        "local": local_budget(root),
         "external_compute": inspect(root),
         "drones": drone_status(root),
         "scope": "Owned job execution only. Windows RAM/disk are worker-local resources, not pooled Debian RAM or shared VRAM. No arbitrary Windows shell is exposed.",

rlm/v100/drones.py

@@ -155,7 +155,7 @@ def execute(root: Path, job: dict) -> dict:
     if job["kind"] == "python":
         from rlm.v100.research_sandbox import run
 
-        return run(root, job["branch"], job["payload"])
+        return run(root, job["branch"], job["payload"], cpu_affinity=job.get("cpu_affinity"))
     if job["kind"] == "source":
         from rlm.v100.research_tools import ResearchTools
 
@@ -277,10 +277,11 @@ def finish(root: Path, identity: str, value: dict) -> None:
 
 
 def service(root: Path, stop: threading.Event) -> None:
-    # A single RTX job plus two CPU/network jobs. RTX requests share the global
+    # One model job plus up to four CPU/network jobs and one observer. RTX requests share the global
     # existing helper pacing lock with foreground researchers and benchmark clients.
-    with ThreadPoolExecutor(max_workers=4) as workers:
+    with ThreadPoolExecutor(max_workers=6) as workers:
         active = {}
+        affinities = {}
         goal_future = None
         goal_due = 0
         while not stop.is_set():
@@ -317,10 +318,11 @@ def service(root: Path, stop: threading.Event) -> None:
                         }
                     finish(root, identity, result)
                     del active[identity]
+                    affinities.pop(identity, None)
             with connect(root) as db:
                 from rlm.v100.resource_budget import local_budget
 
-                budget = local_budget()
+                budget = local_budget(root)
                 jobs = db.execute(
                     "SELECT * FROM jobs WHERE state='queued' AND due<=? ORDER BY due LIMIT 16",
                     (time.time(),),
@@ -333,12 +335,22 @@ def service(root: Path, stop: threading.Event) -> None:
                     )
                     if used >= (min(1, budget["cpu_slots"]) if llm else budget["cpu_slots"]):
                         continue
+                    job = dict(row)
+                    if job["kind"] == "python":
+                        from rlm.v100.resource_budget import cpu_assignment
+
+                        assignment = cpu_assignment(list(affinities.values()))
+                        if not assignment:
+                            continue
+                        job["cpu_affinity"] = assignment
                     updated = db.execute(
                         "UPDATE jobs SET state='running',updated=? WHERE id=? AND state='queued'",
                         (time.time(), row["id"]),
                     )
                     if updated.rowcount:
-                        active[row["id"]] = (workers.submit(execute, root, dict(row)), row["kind"])
+                        if job["kind"] == "python":
+                            affinities[row["id"]] = assignment
+                        active[row["id"]] = (workers.submit(execute, root, job), row["kind"])
             atomic_json(
                 root / "research/drones-status.json",
                 {
@@ -347,6 +359,7 @@ def service(root: Path, stop: threading.Event) -> None:
                     "active": list(active),
                     "cpu_slots": budget["cpu_slots"],
                     "budget": budget,
+                    "cpu_affinities": affinities,
                     "rtx_slots": 0 if gpu_disabled(root) else 1,
                     "local_model_slots": 1 if gpu_disabled(root) else 0,
                 },

rlm/v100/mission_chat.py

@@ -145,6 +145,8 @@ def save_direct_reply(root: Path, message: str, response: dict) -> dict:
         raise ValueError("Chat needs 1-8000 characters")
     if response.get("actions") or response.get("applied"):
         raise ValueError("Direct replies must be read-only facts")
+    if not isinstance(response.get("answer"), str) or not response["answer"].strip():
+        raise ValueError("Empty direct chat reply rejected")
     identity = uuid.uuid4().hex
     with connect(root) as db:
         db.execute(
@@ -565,6 +567,11 @@ def respond(root: Path, directory: Path, request: dict, accepted_cpu=None) -> di
     # Authorization comes only from the current authenticated local user message,
     # never from a model-generated action or text retrieved from the internet.
     message = request["message"].strip()
+    from rlm.v100 import resource_optimization
+
+    resource_context = resource_optimization.context(root, request)
+    if resource_context is not None:
+        return resource_optimization.apply(root, message, resource_context)
     from rlm.v100 import chat_progress
 
     if chat_progress.requested(message):
@@ -1009,6 +1016,7 @@ def respond(root: Path, directory: Path, request: dict, accepted_cpu=None) -> di
     result["execution_receipts"] = execution_receipts(tool_receipts)
     if tool_receipts and all(receipt_failed(row) for row in tool_receipts):
         result["actions"] = []
+        result["execution_status"] = "failed"
         result["answer"] = "Requested work failed: " + "; ".join(
             str(row.get("result")) for row in tool_receipts
         )
@@ -1084,13 +1092,24 @@ def service_loop(root: Path, directory: Path, stop: threading.Event, accepted_cp
                 if accepted_cpu is not None
                 else respond(root, directory, request)
             )
+            if not isinstance(response.get("answer"), str) or not response["answer"].strip():
+                raise ValueError("Empty chat reply rejected; request not completed")
+            failed = response.get("execution_status") == "failed"
             with connect(root) as db:
                 db.execute(
-                    "UPDATE requests SET state='completed',response=?,error=NULL,updated=? WHERE id=? AND state='queued'",
-                    (json.dumps(response, ensure_ascii=False), time.time(), request["id"]),
+                    "UPDATE requests SET state=?,response=?,error=?,updated=? WHERE id=? AND state='queued'",
+                    (
+                        "failed" if failed else "completed",
+                        json.dumps(response, ensure_ascii=False),
+                        response["answer"][:500] if failed else None,
+                        time.time(),
+                        request["id"],
+                    ),
                 )
             ActivityLog(root, "controller", "chat").write(
-                "decisions", "user-command-completed", {"id": request["id"], **response}
+                "errors" if failed else "decisions",
+                "user-command-failed" if failed else "user-command-completed",
+                {"id": request["id"], **response},
             )
         except BackendNotReady as error:
             started = request.get("wait_started") or time.time()

rlm/v100/research_sandbox.py

@@ -95,12 +95,22 @@ def command(root: Path, source: Path, work: Path, broker: Path) -> list[str]:
     ]
 
 
-def run(root: Path, branch: str, code: str, seconds: int = 60) -> dict:
+def run(
+    root: Path, branch: str, code: str, seconds: int = 60, cpu_affinity: list[int] | None = None
+) -> dict:
     if branch not in ("A", "B") or not isinstance(code, str) or not 1 <= len(code) <= 12000:
         raise ValueError("CPU experiment needs A/B and 1-12000 code characters")
     if type(seconds) is not int or not 1 <= seconds <= 60:
         raise ValueError("CPU experiment timeout must be 1-60 seconds")
     compile(code, "experiment.py", "exec")
+    available = os.sched_getaffinity(0)
+    affinity = sorted(available)[:2] if cpu_affinity is None else cpu_affinity
+    if (
+        not 1 <= len(affinity) <= 2
+        or any(type(cpu) is not int or cpu not in available for cpu in affinity)
+        or len(set(affinity)) != len(affinity)
+    ):
+        raise ValueError("CPU experiment affinity must select 1-2 distinct allowed CPUs")
     # Verify the namespace runtime BEFORE executing any model-generated code.
     runtime(root)
     import shutil
@@ -155,7 +165,7 @@ def handle(self):
                         start_new_session=True,
                     )
                     try:
-                        os.sched_setaffinity(process.pid, sorted(os.sched_getaffinity(0))[:2])
+                        os.sched_setaffinity(process.pid, affinity)
                         deadline = time.monotonic() + seconds
                         while process.poll() is None:
                             used = sum(
@@ -186,6 +196,7 @@ def handle(self):
         "directory": str(folder),
         "exit_code": process.returncode,
         "timed_out": timed_out,
+        "cpu_affinity": affinity,
         "seconds": time.monotonic() - started,
         "output": (folder / "output.log").read_text(errors="replace")[-12000:],
         "sources": observations,

rlm/v100/research_tools.py

@@ -609,6 +609,14 @@ def execute(self, name: str, arguments: dict) -> dict:
             ):
                 raise ValueError("Invalid archive identity or byte offset")
             path = self.root / "research/tool-results" / f"{digest}.json"
+            if not path.exists():
+                return {
+                    "status": "failed",
+                    "code": "archive_not_found",
+                    "sha256": digest,
+                    "error": "Tool archive unavailable. Use only full_result_sha256 returned by a tool in this task. Do not invent an ID or infer the missing content; rerun the original read-only tool if known.",
+                    "retryable": False,
+                }
             if path.is_symlink() or path.stat().st_size > 8 * 2**20:
                 raise ValueError("Archive exceeds its read budget or is a symlink")
             raw = path.read_bytes()

rlm/v100/resource_budget.py

@@ -1,18 +1,86 @@
 """Headroom admission, not synthetic activity or a board power limit."""
 
+import json
+import math
+import os
+from pathlib import Path
 
-def local_budget() -> dict:
+
+def available_cpus(
+    cgroup_root: Path = Path("/sys/fs/cgroup"), membership: Path = Path("/proc/self/cgroup")
+) -> int:
+    """Count process affinity and Linux cgroup v2 quota; admission reserves CPU separately."""
+    count = (
+        len(os.sched_getaffinity(0)) if hasattr(os, "sched_getaffinity") else os.cpu_count() or 1
+    )
+    if membership.exists():
+        for line in membership.read_text().splitlines():
+            if line.startswith("0::"):
+                base = cgroup_root.resolve()
+                folder = (base / line[3:].lstrip("/")).resolve()
+                if not folder.is_relative_to(base):
+                    # A cgroup namespace can report a path outside the mounted subtree.
+                    # Its mounted root still exposes the namespace's effective root quota.
+                    folder = base
+                while folder.is_relative_to(base):
+                    quota = folder / "cpu.max"
+                    if quota.exists():
+                        limit, period = quota.read_text().split()
+                        if limit != "max":
+                            count = min(count, max(1, math.floor(int(limit) / int(period))))
+                    if folder == base:
+                        break
+                    folder = folder.parent
+    return max(1, count)
+
+
+def local_budget(root: Path | None = None) -> dict:
     import psutil
 
     cpu = psutil.cpu_percent(interval=0.05)
     free = psutil.virtual_memory().available / 2**30
-    slots = 0 if cpu >= 85 or free < 2 else 1 if cpu >= 65 or free < 4 else 2
+    maximum = 2
+    reserve_ram, ram_per_slot = 0, 2
+    path = root / "research/state/resource-policy.json" if root is not None else None
+    if path is not None and path.exists():
+        policy = json.loads(path.read_text())
+        maximum = policy["max_cpu_slots"]
+        if type(maximum) is not int or not 1 <= maximum <= 4:
+            raise ValueError("Local CPU policy must allow 1-4 slots")
+        reserve_ram, ram_per_slot = (
+            policy.get("reserve_ram_gib", 0),
+            policy.get("ram_per_slot_gib", 2),
+        )
+        if (
+            type(reserve_ram) is not int
+            or not 0 <= reserve_ram <= 8
+            or type(ram_per_slot) is not int
+            or not 2 <= ram_per_slot <= 4
+        ):
+            raise ValueError("Invalid local RAM reservation")
+    cpus = available_cpus()
+    maximum = min(maximum, max(1, (cpus - 1) // 2))
+    ram_slots = max(0, int((free - reserve_ram) // ram_per_slot))
+    slots = 0 if cpu >= 85 else min(1 if cpu >= 65 else maximum, ram_slots)
     return {
         "cpu_slots": slots,
+        "max_cpu_slots": maximum,
+        "available_cpus": cpus,
+        "reserve_ram_gib": reserve_ram,
+        "ram_per_slot_gib": ram_per_slot,
         "host_cpu_percent": cpu,
         "free_ram_gib": round(free, 2),
         "reason": "Host pressure; queued work retained"
         if not slots
         else "Measured CPU/RAM headroom",
-        "scope": "At most two local CPU/network drones. GPU serving/training use existing exclusive admission. Windows uses its own lower limits; RTX remains operator-disabled.",
+        "scope": "Bounded local CPU/network admission; one model job. GPU serving/training keep exclusive gates. Windows keeps its own pressure, RAM and game guards. Slots are concurrency limits, not measured speedup.",
     }
+
+
+def cpu_assignment(active: list[list[int]]) -> list[int]:
+    """Allocate distinct affinity slots for concurrent owned Python experiments."""
+    cpus = sorted(os.sched_getaffinity(0))[: available_cpus()]
+    if len(cpus) > 2:
+        cpus = cpus[1:]
+    used = {cpu for assignment in active for cpu in assignment}
+    return [cpu for cpu in cpus if cpu not in used][:2]