Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
318 changes: 317 additions & 1 deletion docs/architecture/rfcs/monorepo-distribution-split-v0.md

Large diffs are not rendered by default.

251 changes: 250 additions & 1 deletion docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md

Large diffs are not rendered by default.

4 changes: 4 additions & 0 deletions examples/control_plane/status-projection-cache-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,10 @@ def run_cli_result(
) -> tuple[dict[str, Any], int, str]:
env = os.environ.copy()
env["PYTHONPATH"] = str(REPO_ROOT)
# This smoke reads machine JSON and asserts the inline decision/cache fields;
# the model-facing capture transport replaces that body with a reader
# reference, so keep the same opt-out the canary harness uses in argv.
env["LOOPX_CONTEXT_DELIVERY"] = "off"
result = subprocess.run(
[
sys.executable,
Expand Down
4 changes: 4 additions & 0 deletions examples/control_plane/status-quota-perf-budget-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,10 @@ def run_cli_result(
) -> tuple[dict[str, Any], float, int, str]:
env = os.environ.copy()
env["PYTHONPATH"] = str(REPO_ROOT)
# This smoke measures status/quota decision latency and reads the inline
# `should_run` field; the model-facing capture transport replaces that body
# with a reader reference, so keep the same opt-out the canary harness uses.
env["LOOPX_CONTEXT_DELIVERY"] = "off"
started = time.perf_counter()
result = subprocess.run(
[
Expand Down
8 changes: 7 additions & 1 deletion examples/control_plane/todo_lifecycle_fixtures.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,11 @@
import contextlib
import io
import json
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch


REPO_ROOT = Path(__file__).resolve().parents[2]
Expand Down Expand Up @@ -126,7 +128,11 @@ def run_fixture_cli(
]
stdout = io.StringIO()
stderr = io.StringIO()
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
# These fixtures consume machine JSON and read the inline decision fields;
# the model-facing capture transport replaces that body with a reader
# reference, so opt out in-process the way the canary harness does in argv.
with patch.dict(os.environ, {"LOOPX_CONTEXT_DELIVERY": "off"}), \
contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
try:
returncode = loopx_cli_main(argv)
except SystemExit as exc:
Expand Down
7 changes: 6 additions & 1 deletion examples/custom-runtime-minimal-cli-turn-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,10 +13,12 @@
import contextlib
import io
import json
import os
import sys
import tempfile
from pathlib import Path
from typing import Any
from unittest.mock import patch

REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
Expand Down Expand Up @@ -62,7 +64,10 @@ def run_cli(registry_path: Path, *args: str) -> dict[str, Any]:
*args,
]
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
# This smoke reads machine JSON and asserts the inline decision fields; the
# model-facing capture transport replaces that body with a reader reference,
# so opt out in-process the way the canary harness does in argv.
with patch.dict(os.environ, {"LOOPX_CONTEXT_DELIVERY": "off"}), contextlib.redirect_stdout(buf):
code = loopx_cli_main(argv)
raw = buf.getvalue().strip()
payload: dict[str, Any] = json.loads(raw) if raw else {}
Expand Down
33 changes: 30 additions & 3 deletions loopx/cli_commands/turn_run_once.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
from __future__ import annotations

import argparse
import sys as _sys
from collections.abc import Callable, Mapping
from pathlib import Path
from typing import TYPE_CHECKING as _TYPE_CHECKING
from typing import Any

from ..capabilities.explore.composition_frontier import (
Expand Down Expand Up @@ -42,15 +44,37 @@
from .turn_cadence import managed_cadence_start
from .turn_host_request import build_codex_cli_host_runner, prepare_turn_host_request
from .turn_decision import collect_turn_status_payload, current_turn_host_launch_decision
from .turn_dsh_host import build_dsh_host_runner, build_dsh_operation_host_runner
from .turn_rendering import build_turn_error_payload
from .turn_post_writeback import drain_committed_turn_post_writeback_hooks
from .turn_todo_writeback import captured_todo_text_sha256, write_turn_validated_completion

if _TYPE_CHECKING:
from .turn_dsh_host import (
build_dsh_host_runner as build_dsh_host_runner,
build_dsh_operation_host_runner as build_dsh_operation_host_runner,
)

EXACT_SETTLEMENT_READBACK_NOT_FOUND = (
"exact settlement readback unexpectedly returned not-found"
)

# Keep the existing import and patch seams without loading an unselected host.
_DSH_HOST_EXPORTS = ("build_dsh_host_runner", "build_dsh_operation_host_runner")


def __getattr__(name: str) -> Any:
if name not in _DSH_HOST_EXPORTS:
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
from . import turn_dsh_host

value = getattr(turn_dsh_host, name)
globals()[name] = value
return value


def __dir__() -> list[str]:
return sorted(set(globals()) | set(_DSH_HOST_EXPORTS))


def execute_turn_run_once(
args: argparse.Namespace,
Expand Down Expand Up @@ -816,7 +840,7 @@ def resolve_built_in_session_binding(
session_binding_resolver = resolve_built_in_session_binding
elif args.host == "dsh":
if getattr(args, "dsh_operation_module", None):
host_runner = build_dsh_operation_host_runner(
host_runner = getattr(_sys.modules[__name__], "build_dsh_operation_host_runner")(
args, runtime_root=runtime_root, registry_path=registry_path,
workspace=project, environ=operator_environ,
goal_admission=goal_admission,
Expand All @@ -829,7 +853,7 @@ def resolve_dsh_operation_session(envelope: Mapping[str, Any]) -> dict[str, str]

session_binding_resolver = resolve_dsh_operation_session
else:
host_runner = build_dsh_host_runner(
host_runner = getattr(_sys.modules[__name__], "build_dsh_host_runner")(
args, workspace=project, environ=operator_environ,
)

Expand Down Expand Up @@ -937,3 +961,6 @@ def on_managed_start_admitted() -> None:
execution_started=execution_started,
journal_readback=journal_readback,
)


__all__ = [name for name in globals() if not name.startswith("_")] + list(_DSH_HOST_EXPORTS)
39 changes: 34 additions & 5 deletions loopx/control_plane/turn_driver/host_bridge_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,35 @@
import queue
import signal
import subprocess
import sys as _sys
import threading
import time
from typing import Any
from typing import TYPE_CHECKING as _TYPE_CHECKING

from ...extensions.process_runtime import force_terminate_process, write_process_stdin_before_deadline
if _TYPE_CHECKING:
from ...extensions.process_runtime import (
force_terminate_process as force_terminate_process,
write_process_stdin_before_deadline as write_process_stdin_before_deadline,
)

# Preview needs the bridge type; process helpers belong to execution or an
# explicit caller import. Keep their existing identity and patch seams.
_PROCESS_EXPORTS = ("force_terminate_process", "write_process_stdin_before_deadline")


def __getattr__(name: str) -> Any:
if name not in _PROCESS_EXPORTS:
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
from ...extensions import process_runtime

value = getattr(process_runtime, name)
globals()[name] = value
return value


def __dir__() -> list[str]:
return sorted(set(globals()) | set(_PROCESS_EXPORTS))


class GuardedHostBridge:
Expand Down Expand Up @@ -61,8 +85,8 @@ def send(self, frame: str) -> None:
if self.next(deadline=self.deadline) != {"kind": "ready"}:
raise RuntimeError("managed Host bridge did not become ready")
self.requested = True
write_process_stdin_before_deadline(self.process, frame, deadline=self.deadline,
cancel_and_reap=self.cancel)
write_stdin = getattr(_sys.modules[__name__], "write_process_stdin_before_deadline")
write_stdin(self.process, frame, deadline=self.deadline, cancel_and_reap=self.cancel)

def cancel(self) -> None:
if self.cancelled:
Expand All @@ -73,7 +97,8 @@ def cancel(self) -> None:
if not self.requested:
# No request bytes have been sent, so this bridge cannot own a
# Host. This also handles suspension before JS handlers exist.
force_terminate_process(self.process, process_tree_owned=True)
getattr(_sys.modules[__name__], "force_terminate_process")(
self.process, process_tree_owned=True)
self.not_launched = True
elif self.process.poll() is None:
self.process.terminate()
Expand All @@ -98,7 +123,8 @@ def cancel(self) -> None:
# Killing the bridge alone cannot certify an unobserved detached
# Host. Retain the launching record and fail instead of claiming it.
if self.process.poll() is None:
force_terminate_process(self.process, process_tree_owned=True)
getattr(_sys.modules[__name__], "force_terminate_process")(
self.process, process_tree_owned=True)
raise RuntimeError("managed Host bridge cleanup unconfirmed") from error

def close_output(self) -> None:
Expand All @@ -112,3 +138,6 @@ def close_output(self) -> None:
self.reader.join(timeout=self.cleanup_timeout)
if not self.reader.is_alive():
self.output.close()


__all__ = [name for name in globals() if not name.startswith("_")] + list(_PROCESS_EXPORTS)
4 changes: 2 additions & 2 deletions loopx/control_plane/work_items/execution_preparation.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import {EXECUTION_HANDOFF_MODES} from "../coordination/handoff_mode_vocabulary.ts";
import type {JsonObject} from "../effect_program.ts";
import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
import {
Expand All @@ -13,7 +14,6 @@ export const EXECUTION_PREPARATION_FACTS_SCHEMA_VERSION =
export const EXECUTION_PREPARATION_SCHEMA_VERSION =
"execution_preparation_v0";

const HANDOFF_MODES = ["soft_claim", "hard_lease"] as const;
const LEASE_STATUSES = ["active", "released", "unknown"] as const;

function leaseFact(value: unknown, label: string): JsonObject {
Expand Down Expand Up @@ -55,7 +55,7 @@ export function projectExecutionPreparation(value: unknown): JsonObject {
}
const goalId = requireNonEmptyString(request.goal_id, "goal_id");
const agentId = requireNonEmptyString(request.agent_id, "agent_id");
const handoffMode = requireStringLiteral(request.handoff_mode, HANDOFF_MODES, "handoff_mode");
const handoffMode = requireStringLiteral(request.handoff_mode, EXECUTION_HANDOFF_MODES, "handoff_mode");
const authorityAvailable = requireBoolean(request.authority_available, "authority_available");
const selected = requireJsonObject(request.selected_todo, "selected_todo");
const todoId = requireNonEmptyString(selected.todo_id, "selected_todo.todo_id");
Expand Down
25 changes: 23 additions & 2 deletions loopx/doctor_git.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,31 @@
import os
import re
import subprocess
import sys as _sys
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Any
from typing import TYPE_CHECKING as _TYPE_CHECKING
from urllib.parse import urlsplit

from .extensions.process_runtime import terminate_process_tree
if _TYPE_CHECKING:
from .extensions.process_runtime import terminate_process_tree as terminate_process_tree


# Successful read-only Git probes do not need process cleanup. Keep the former
# helper export and patch point, resolving its owner only when requested.
def __getattr__(name: str) -> Any:
if name != "terminate_process_tree":
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
from .extensions.process_runtime import terminate_process_tree

globals()[name] = terminate_process_tree
return terminate_process_tree


def __dir__() -> list[str]:
return sorted(set(globals()) | {"terminate_process_tree"})


# `doctor` inspects a checkout the operator may not control, so every Git read
Expand Down Expand Up @@ -64,7 +82,7 @@ def _retire_owned_probe(process: subprocess.Popen[str]) -> None:
"""

try:
terminate_process_tree(process, grace_seconds=0.2)
getattr(_sys.modules[__name__], "terminate_process_tree")(process, grace_seconds=0.2)
except (OSError, RuntimeError):
# The answer is already unknown; an unconfirmed cleanup must not turn a
# read-only diagnostic into a traceback.
Expand Down Expand Up @@ -324,3 +342,6 @@ def trusted_release_ref_for_root(
"git_ref": f"{remote}/{expected_ref}",
}
return None


__all__ = [name for name in globals() if not name.startswith("_")] + ["terminate_process_tree"]
1 change: 1 addition & 0 deletions skills/loopx-self-repair/references/repair-patterns.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ teaches a reusable control-plane lesson.

| Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair |
| --- | --- | --- | --- | --- |
| `marked_todo_archive_pointer_reader_drift` | An installed `quota should-run` reports `quota_unexpected_collection_error`; read-only `loopx --format json quota status --goal-id <goal-id> --verbose` confirms `ValueError: non-generated content inside marked Todo region`, while current source reads succeed. Thin Codex App output may expose only the error code. | Use `python3 skills/loopx-self-repair/scripts/find_pattern.py --query 'quota_unexpected_collection_error non-generated content inside marked Todo region'`, then read this exact pattern. Inspect the launcher, Python/import route, component hashes, exact emitted line and installed `machine_region.py`; compare synthetic File/SQLite source/wheel fixtures. Keep verbose diagnostics and snapshots private. | The observed reader matching `8d314af49` rejects an emitted archive-summary comment without recognized Todo metadata. It already accepts the emitted per-Todo active-field omission row through valid `todo_id` metadata. Their shared `loopx:todo-archive-pointer` prefix does not establish the cause; attribute an omission-row failure only when the exact installed reader actually rejects that exact generated line. Version or declared commit alone cannot establish dirty installed bytes. | Verify projection ownership and compare archive-summary, per-Todo omission and foreign-narrative controls under the same interpreter and real backend; preserve foreign-content rejection, original-Turn recovery/replay, unchanged canonical work and zero quota debit. If main already includes the archive-summary reader repair, as in #7834, do not duplicate it, strip active metadata or relax the grammar. Qualify the affected source/wheel composition, then follow existing upgrade authority and original-binding readback. Source recovery alone does not qualify the installed runtime or prove long-horizon benefit. |
| `answer_question_replaced_by_delivery_status` | A user asks why a result occurred or whether it is expected, but receives measurements, tests or PR progress even though the reply was delivered completely. | The actual question and relevant corrections, current receiver brief, shared answer guidance, and the complete message on its original return route. | The author answered implementation progress instead of the causal question; successful transport or correct counters were mistaken for a useful answer. | Retain the correction in the existing Task and outcome, then repair the existing shared generation guidance: explain meaning, supported mechanism, expectations and the specific unknown before supporting data. Keep hypotheses separate from confirmed causes and preserve the qualified investigation owner. Verify guidance at native callers and read back the actual explanatory answer. Do not add keyword-based answer scoring, a second model rewrite, another display or a duplicate investigation; a prompt test or delivery receipt does not prove model compliance or user adoption. |
| `queue_action_type_coercion` | A queued input action such as `["edit"]` is accepted but applies cancellation, or a stored operation accepts a non-string action. | Original JSON types, the typed planner and receipt decoder, real HTTP responses, canonical Turn/transcript/events/Session snapshots and native IO before/after refusal. | Membership checks stringify the action while effect selection and replay compare its original literal value. | Validate the original literal action at the existing owner before branching and when decoding pending or committed receipts. Cover arrays, nested arrays, objects, nulls, numbers, booleans and unknown strings; require rejection without state or native effects, then verify legal edit/cancel, original-operation replay and queue races. Preserve malformed historical evidence rather than coercing or rewriting receipts. |
| `collaboration_return_format_projection_gap` | A complete answer that fits the result budget is rejected, or a published versioned return becomes unreadable or cannot replay through the real CLI. | Selected renderer output and Unicode count; persisted reply, compact publication/Inbox observations, original-route delivery and legacy replay at the same boundary. | Validation counted a different renderer before choosing the immutable format, or a compact codec omitted the persisted format marker and silently treated it as legacy. | Apply the unchanged budget to the actual selected format, carry its existing marker through every compact observation, and retain old bytes on replay. Validate BMP/non-BMP exact-limit and one-character-over cases, unsupported versions, real CLI publication/retry, original-audience return and File/SQLite packaged readback. Do not raise the budget, truncate the answer, rewrite historical receipts or add a second model. |
Expand Down
Loading
Loading