From b8993871f3dd634319964f4134775431155a81ed Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 12:58:03 +0800 Subject: [PATCH 1/4] fix(runtime): keep heartbeat launch public scan scope consistent Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/cli_commands/turn_decision.py | 5 ++- .../test_benchmark_current_admission.py | 32 ++++++++++++++++++- 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py index 9afe869ff2..9cedae2030 100644 --- a/loopx/cli_commands/turn_decision.py +++ b/loopx/cli_commands/turn_decision.py @@ -38,6 +38,7 @@ scheduler_execution_context_for_turn, ) from ..status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, collect_status +from ..paths import default_public_scan_root from ..control_plane.effect_runtime import effect_runtime_result from .lark_inbox import build_lark_operator_inbox_urgency_projector from .turn_selection import turn_controller_advisory_primary @@ -337,7 +338,9 @@ def current_heartbeat_host_launch_decision( args = argparse.Namespace( goal_id=goal_id, agent_id=agent_id, turn_instance_id=original_turn, todo_id=None, codex_app=False, runtime_profile="generic_cli", - scan_path=[str(project)], scan_root=str(project), limit=0, + # Match heartbeat-prompt's default public scan. The execution workspace + # may be a private task with auth fixtures, rather than public LoopX code. + scan_path=[], scan_root=default_public_scan_root(), limit=0, available_capabilities=["shell", "filesystem_write"], ) selection = load_requested_quota_action_selection( diff --git a/tests/control_plane/test_benchmark_current_admission.py b/tests/control_plane/test_benchmark_current_admission.py index 479238c8d6..508a7a889b 100644 --- a/tests/control_plane/test_benchmark_current_admission.py +++ b/tests/control_plane/test_benchmark_current_admission.py @@ -2,6 +2,7 @@ from __future__ import annotations import json +from ipaddress import IPv4Address from pathlib import Path import sys import time @@ -11,9 +12,11 @@ from benchmark.runtime import worker from benchmark.tests.test_shared_codex_runtime import worker_env from loopx.configure_goal import configure_goal +from loopx.cli_commands import turn_decision from loopx.control_plane.goals import activation_service from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected from loopx.global_registry import sync_project_registry_to_global +from loopx.paths import default_public_scan_root from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime from tests.control_plane.host_process_fixture import write_python_host_fixture from tests.control_plane import test_quota_settlement_cli as fixture @@ -39,6 +42,11 @@ def _environment(tmp_path, monkeypatch, provider, *, replan=False): elif replan: fixture._configure_autonomous_replan_fixture(project, runtime, source) fixture._initialize_fixture_authority(project, runtime, provider=provider) + # A private task can contain synthetic private-network fixtures. It is not + # the public package tree selected by the heartbeat CLI's default scan. + (project / "task_network_fixture.py").write_text( + "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n" + ) synced = sync_project_registry_to_global( registry_path=source, runtime_root_override=str(runtime), goal_id=fixture.GOAL_ID, ) @@ -70,7 +78,7 @@ def _environment(tmp_path, monkeypatch, provider, *, replan=False): def _quota(registry, runtime, project): code, result = fixture._run_cli(registry, runtime, "quota", "should-run", "--goal-id", fixture.GOAL_ID, "--agent-id", fixture.AGENT_ID, - "--runtime-profile", "generic_cli", cwd=project) + "--runtime-profile", "generic_cli", "--scan-root", default_public_scan_root(), cwd=project) assert code == 0 or code == 1 and result.get("status_health_ok") is False, result return result @@ -112,6 +120,7 @@ def after_session_selection(*arguments): monkeypatch.setattr(worker, "native_command", after_session_selection) bindings_before = _bindings(runtime) rejected = None + result = None try: result = worker.run_once(env) except FirstPartyHostRuntimeRejected as error: @@ -166,3 +175,24 @@ def test_initial_benchmark_planner_can_launch_without_a_prior_heartbeat_receipt( assert len(planned["todo_ids"]) == 1 and result["planning"]["todo_ids"] == planned["todo_ids"] assert not fixture._heartbeat_receipt_events(runtime, result["turn_id"]) assert fixture._spend_run_count(runtime) == 0 + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_heartbeat_host_launch_retains_public_scan_refusal(tmp_path, monkeypatch, provider): + env, project, runtime, source, _state = _environment(tmp_path, monkeypatch, provider) + public = tmp_path / "selected-public" + public.mkdir() + (public / "unsafe_fixture.py").write_text( + "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n" + ) + code, original = fixture._run_cli(source, runtime, + "quota", "should-run", "--goal-id", fixture.GOAL_ID, "--agent-id", fixture.AGENT_ID, + "--scan-path", str(public), cwd=project) + assert code == 1 and original["status_health_ok"] is False and original["should_run"] is False + monkeypatch.setattr(turn_decision, "default_public_scan_root", lambda: str(public)) + current = turn_decision.current_heartbeat_host_launch_decision( + registry_path=Path(env["LOOPX_REGISTRY"]), runtime_root=runtime, project=project, + goal_id=fixture.GOAL_ID, agent_id=fixture.AGENT_ID, turn_instance_id="public-scan-refusal", + )["current"] + assert current["status_health_ok"] is False and current["should_run"] is False + assert not (project / "native-starts").exists() and not (project / "model-starts").exists() From 7086073e92abbb0f28451f548c51db0fbec583d5 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 13:53:14 +0800 Subject: [PATCH 2/4] fix(runtime): preserve held checkpoint scan scope and isolate fixtures Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/cli_commands/turn_decision.py | 7 ++++--- .../test_benchmark_checkpoint_admission.py | 11 ++++++++++- .../test_benchmark_current_admission.py | 17 ++++++++++------- 3 files changed, 24 insertions(+), 11 deletions(-) diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py index 9cedae2030..2cf6275c59 100644 --- a/loopx/cli_commands/turn_decision.py +++ b/loopx/cli_commands/turn_decision.py @@ -338,9 +338,10 @@ def current_heartbeat_host_launch_decision( args = argparse.Namespace( goal_id=goal_id, agent_id=agent_id, turn_instance_id=original_turn, todo_id=None, codex_app=False, runtime_profile="generic_cli", - # Match heartbeat-prompt's default public scan. The execution workspace - # may be a private task with auth fixtures, rather than public LoopX code. - scan_path=[], scan_root=default_public_scan_root(), limit=0, + # Uncaptured wakes use heartbeat-prompt's public-package default. Keep + # the existing workspace recheck for an already-held checkpoint packet. + scan_path=[str(project)] if original_quota is not None else [], + scan_root=default_public_scan_root(), limit=0, available_capabilities=["shell", "filesystem_write"], ) selection = load_requested_quota_action_selection( diff --git a/tests/control_plane/test_benchmark_checkpoint_admission.py b/tests/control_plane/test_benchmark_checkpoint_admission.py index e7c9d1a1d4..bb3ff3e06d 100644 --- a/tests/control_plane/test_benchmark_checkpoint_admission.py +++ b/tests/control_plane/test_benchmark_checkpoint_admission.py @@ -2,6 +2,7 @@ from __future__ import annotations import json +from ipaddress import IPv4Address import os from pathlib import Path import sys @@ -115,7 +116,7 @@ def rendered(command): @pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("refusal", ["source_compute_zero", "missing_state", "stop", "original_event_changed", None]) +@pytest.mark.parametrize("refusal", ["source_compute_zero", "missing_state", "stop", "original_event_changed", "public_scope_violation", None]) def test_checkpoint_resume_rechecks_current_source_with_original_guard(tmp_path, monkeypatch, provider, refusal): _run_checkpoint(tmp_path, monkeypatch, provider, refusal=refusal, todo_bound=False) @@ -186,6 +187,13 @@ def captured_then_changed(held, deadline): elif refusal == "missing_state": state.rename(state.with_suffix(".saved")) assert _quota(source, runtime, project)["status_health_ok"] is False + elif refusal == "public_scope_violation": + # The held guard explicitly scanned this workspace. A default + # package scan must not replace that check on continuation. + (project / "changed_public_fixture.py").write_text( + "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n" + ) + assert _quota(source, runtime, project)["status_health_ok"] is False elif refusal == "stop": with monkeypatch.context() as fault: fault.setattr(activation_service, "sync_project_registry_to_global", lambda **kwargs: { @@ -237,6 +245,7 @@ def captured_then_changed(held, deadline): assert error is not None and { "source_compute_zero":"goal_launch_not_allowed", "missing_state":"goal_launch_not_allowed", "stop":"goal_stopped", "original_event_changed":"heartbeat_launch_original_binding_changed", + "public_scope_violation":"goal_launch_not_allowed", }[refusal] in str(error) assert not (project / "checkpoint-business.json").exists() assert fixture._heartbeat_receipt_events(runtime, HELD_TURN) == observations["events"] diff --git a/tests/control_plane/test_benchmark_current_admission.py b/tests/control_plane/test_benchmark_current_admission.py index 508a7a889b..0384f8bd39 100644 --- a/tests/control_plane/test_benchmark_current_admission.py +++ b/tests/control_plane/test_benchmark_current_admission.py @@ -16,7 +16,6 @@ from loopx.control_plane.goals import activation_service from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected from loopx.global_registry import sync_project_registry_to_global -from loopx.paths import default_public_scan_root from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime from tests.control_plane.host_process_fixture import write_python_host_fixture from tests.control_plane import test_quota_settlement_cli as fixture @@ -42,11 +41,6 @@ def _environment(tmp_path, monkeypatch, provider, *, replan=False): elif replan: fixture._configure_autonomous_replan_fixture(project, runtime, source) fixture._initialize_fixture_authority(project, runtime, provider=provider) - # A private task can contain synthetic private-network fixtures. It is not - # the public package tree selected by the heartbeat CLI's default scan. - (project / "task_network_fixture.py").write_text( - "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n" - ) synced = sync_project_registry_to_global( registry_path=source, runtime_root_override=str(runtime), goal_id=fixture.GOAL_ID, ) @@ -78,7 +72,7 @@ def _environment(tmp_path, monkeypatch, provider, *, replan=False): def _quota(registry, runtime, project): code, result = fixture._run_cli(registry, runtime, "quota", "should-run", "--goal-id", fixture.GOAL_ID, "--agent-id", fixture.AGENT_ID, - "--runtime-profile", "generic_cli", "--scan-root", default_public_scan_root(), cwd=project) + "--runtime-profile", "generic_cli", cwd=project) assert code == 0 or code == 1 and result.get("status_health_ok") is False, result return result @@ -87,6 +81,12 @@ def _quota(registry, runtime, project): @pytest.mark.parametrize("refusal", ["source_compute_zero", "missing_state", "stop", None]) def test_heartbeat_worker_qualifies_current_source_at_actual_spawn(tmp_path, monkeypatch, provider, refusal): env, project, runtime, source, state = _environment(tmp_path, monkeypatch, provider) + if refusal is None: + # This default-scan caller may work on a private task. Keep that fixture + # out of the shared environment used by explicit project-scan callers. + (project / "task_network_fixture.py").write_text( + "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n" + ) mirror = Path(env["LOOPX_REGISTRY"]) mirror_before, state_before = mirror.read_bytes(), state.read_bytes() observations = [] @@ -149,6 +149,9 @@ def test_initial_benchmark_planner_can_launch_without_a_prior_heartbeat_receipt( env, project, runtime, source, _state = _environment(tmp_path, monkeypatch, provider, replan="empty") packet = _quota(source, runtime, project) assert packet["should_run"] is True and packet.get("selected_todo") is None + (project / "task_network_fixture.py").write_text( + "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n" + ) env.update(LOOPX_TASK_ENTRY="loopx-planned", LOOPX_TASK_STAGE="plan", LOOPX_EXECUTION_MODE=mode, LOOPX_PHASE_DEADLINE_EPOCH=str(time.time()+600), LOOPX_PLANNING_RESULT=str(project / "planning.json")) if mode == "turn": From ac3e760d919858a55c9b2f520207fb3368605fdc Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 18:46:20 +0800 Subject: [PATCH 3/4] fix(runtime): separate execution admission from public disclosure checks Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/public-boundary-read-cost.md | 110 +++++++++++++----- loopx/canary/premerge.py | 23 ++-- loopx/cli_commands/quota_registration.py | 4 +- loopx/cli_commands/status_registration.py | 21 ++-- loopx/cli_commands/turn_decision.py | 9 +- loopx/cli_commands/turn_registration.py | 10 +- loopx/contract.py | 66 ++++++++--- loopx/control_plane/status/collection.py | 2 +- loopx/status.py | 2 +- .../test_premerge_public_boundary_run.py | 20 ++++ .../test_benchmark_checkpoint_admission.py | 23 ++-- .../test_benchmark_current_admission.py | 89 +++++++++----- .../test_public_boundary_parallel_reads.py | 26 +++-- ...st_contract_scan_unreadable_directories.py | 6 +- tests/test_contract_scan_unreadable_files.py | 2 +- tests/test_public_check_disclosure.py | 100 ++++++++++++++++ 16 files changed, 380 insertions(+), 133 deletions(-) create mode 100644 tests/test_public_check_disclosure.py diff --git a/docs/reference/public-boundary-read-cost.md b/docs/reference/public-boundary-read-cost.md index 4ab8962b16..db1a9063e5 100644 --- a/docs/reference/public-boundary-read-cost.md +++ b/docs/reference/public-boundary-read-cost.md @@ -2,52 +2,96 @@ ## Contract / 契约 -`scan_public_boundary` still enumerates the full requested tree, including Git-owned -files inside otherwise pruned directories. Its existing owner decides which -local-private files to exclude **before** submitting any reads. Eligible files are -reopened on every request. The filesystem adapter overlaps at most eight reads and -returns them in the original sorted order; it does not cache scan results or -change leak rules, policy, Git ownership, counts, or error handling. - -`scan_public_boundary` 仍扫描完整的指定目录，包括通常会被剪枝、但属于 Git -版本控制的文件。既有扫描 owner 在提交读取之前排除未跟踪的本地私有文件。 -每次请求重新打开所有符合条件的文件；文件系统适配器最多并发读取八个文件， -按原来的排序返回结果，不缓存扫描结果，不改变泄漏规则、策略、Git 所有权、 -计数及错误处理。 +Ordinary `status`, quota decisions, initial planning and native Turn/heartbeat +launch or checkpoint continuation **defer** repository public-boundary scanning. +They still re-read current execution authority: Stop, pause, compute, Goal/Agent, +source, claim/lease and the original receipt/action/settlement binding. Deferral +does not certify public safety. The existing status checks explicitly say +`public boundary scan deferred`; no execution setting or permission bypass is +introduced. + +普通 `status`、quota 决策、首次规划、原生 Turn/heartbeat 启动及 checkpoint 接续 +默认**推迟**仓库公共边界扫描，仍复查 Stop、暂停、compute、Goal/Agent、source、 +claim/lease 和原 receipt/action/settlement 绑定。推迟不等于公开安全；既有状态检查 +明确显示 `public boundary scan deferred`，没有新增执行配置或权限绕过。 + +Explicit `loopx check --scan-path ` and the existing +premerge check scan selected outgoing content. Repeat `--scan-path` for multiple +files, including a saved diff, attachment or draft body. Premerge selects existing +changed files. This is not a universal outbound interceptor: generated-text +presentation guards retain their separate existing owner and policy; arbitrary +uploads/messages are not automatically covered by a successful file check. + +显式 `loopx check --scan-path <拟外发文件或目录>` 与既有 premerge 检查选定外发内容。 +多个文件可重复传 `--scan-path`，也可选择已保存的 diff、附件或正文草稿；premerge +选择当前存在的变更文件。这不是统一外发拦截器：生成文本的 presentation guard +保留既有独立 owner 和策略，任意上传或消息不因文件检查成功而自动获得覆盖。 + +The existing scanner retains all rule locations in `hits`. Literal credentials, +private source/path content and non-public registries remain blockers in +`blocking_hits`. A private IP alone is ambiguous: it remains a located warning, +as do existing credential environment/example references. These warnings do not +block the check and are labelled **completed with warnings**, never clean. This +is heuristic assistance, not proof that arbitrary content is safe or secret. + +既有扫描器在 `hits` 保留全部规则定位，`blocking_hits` 保留字面凭据、私有来源/ +路径内容及非公开 registry 等阻断项。单独内网 IP 属于含糊信号，仍显示带位置的 +提示；原有凭据环境变量/示例引用同样提示。提示不阻断检查，显示为 +**completed with warnings**，不称 clean。规则是辅助检查，不能证明任意内容安全 +或含有秘密。 + +Missing roots, unreadable directories/files and undecodable UTF-8 fail the check. +Explicit private-file/directory selection cannot hide behind Git ignore or +directory pruning; tracked private state is also blocked. The existing opt-in +tracked-document-URL policy still applies to otherwise public files, and is +re-read on each check; it does not authorize private-state export. Untargeted, +untracked private state stays excluded **before** file reads. + +缺失根目录、不可列目录、不可读文件及无法解码的 UTF-8 均使检查失败。显式选择 +私有文件/目录不能靠 Git ignore 或目录剪枝得到空扫描成功；已跟踪私有状态也阻断。 +原有显式 tracked-document-URL 策略仍适用于其他公开文件，每次检查重新读取；它 +不授权私有状态导出。未被选作外发目标的未跟踪私有状态仍在读取**之前**排除。 + +`scan_public_boundary` enumerates the full selected tree, including Git-owned +files inside otherwise pruned directories. Eligible files are reopened on each +explicit check. The filesystem adapter overlaps at most eight reads and returns +them in sorted order; it caches no content or result. + +`scan_public_boundary` 遍历完整指定树，包括通常剪枝目录中的 Git 文件。显式检查 +每次重新打开符合条件的文件；文件系统适配器最多并发读取八个文件，按排序返回， +不缓存内容或结果。 The bound is a file-count bound, not a byte-budget or atomic filesystem snapshot. -Unreadable-file observations and Unicode decoding behavior retain their existing -semantics. An unexpected reader failure is not downgraded to a clean scan. Workers +An unreadable or undecodable input is not downgraded to a clean scan. Workers only read UTF-8 bytes: traversal, policy, classification and state writes are not delegated to them. 限制针对待处理文件数量，不是字节预算，也不承诺原子文件系统快照。 -不可读文件和 Unicode 解码错误保留既有语义；意外读取异常不会被当成扫描通过。 +不可读文件、Unicode 解码错误及意外读取异常不会被当成扫描通过。 工作线程只读取 UTF-8 内容，不承担目录遍历、策略判断、分类或状态写入。 ## Placement / 实现边界 -The change is a Python filesystem adapter beneath the existing Python scan owner. -It does not add a second control-plane decision source. Moving the scan policy or -completion/deadline authority to a new owner is out of scope; the existing -TypeScript transaction and settlement owners remain unchanged. This bounded I/O -repair need not wait for the larger TypeScript migration. +Filesystem classification and explicit check reporting remain in the existing +Python scan/check owner and bounded reader. There is no second policy store or +control-plane decision source. Existing TypeScript admission, transaction, +completion/deadline and settlement owners remain unchanged. -这次改动是既有 Python 扫描 owner 下面的文件系统适配器，不新增平行控制面决策源。 -不迁移扫描策略，也不改变完成、时限或结算权威；既有 TypeScript 事务及结算 -owner 保持不变。这个有界 I/O 修复不以前置完成整个 TS 重构为条件。 +文件系统分类及显式检查报告保留在既有 Python scan/check owner 和有界 reader， +没有第二份策略存储或平行控制面决策源；既有 TypeScript 准入、事务、完成、 +时限和结算 owner 不变。 ## User paths / 用户入口 -CLI checks and quota preparation reuse the same scan and response projection. -There is no new setting, CLI flag, frontend control or Lark command. The existing -frontend status hot path still explicitly defers the repository scan; it must not -be reported as a completed public-boundary check. Managed/Lark consumers keep the -same quota gate rather than a separate optimization-specific result. +CLI checks run the scanner; operational CLI, managed hosts and Lark use the same +status owner with scanning deferred by default. The frontend status path already +defers it. There is no new setting, CLI flag, frontend control or Lark command. +The compact status projection keeps the deferred check text rather than adding +repository scan bodies to every ordinary response. -CLI 检查及 quota 准备继续复用同一扫描和响应投影，没有新增配置、CLI 参数、 -前端控件或 Lark 命令。既有前端状态热路径仍明确推迟仓库扫描，不能冒充已完成 -公共边界检查。Managed/Lark 消费者使用原有 quota 门禁，不另建优化专用结果。 +CLI 显式检查执行扫描，运行用 CLI、managed host 与 Lark 复用既有状态 owner， +默认推迟扫描；前端状态路径此前已推迟。没有新增配置、CLI 参数、前端控件或 +Lark 命令。紧凑状态投影保留 deferred 检查文字，不向每次普通响应加入仓库扫描正文。 Therefore no packaged frontend rebuild or layout change is required. Existing status-hot-path regression checks must continue to prove the explicit deferred @@ -68,6 +112,10 @@ uv run --extra test python -m pytest \ tests/test_contract_public_boundary_prefilter.py \ tests/test_contract_scan_unreadable_files.py \ tests/test_contract_credential_pattern_precision.py \ + tests/test_public_check_disclosure.py \ + tests/canary/test_premerge_public_boundary_run.py \ + tests/control_plane/test_benchmark_current_admission.py \ + tests/control_plane/test_benchmark_checkpoint_admission.py \ tests/test_public_package_lock_boundary.py -q ``` diff --git a/loopx/canary/premerge.py b/loopx/canary/premerge.py index 44106003ba..51189f479e 100644 --- a/loopx/canary/premerge.py +++ b/loopx/canary/premerge.py @@ -205,21 +205,26 @@ def _public_boundary_changed_files_run( unreadable_directories = [ str(item) for item in boundary.get("unreadable_directories") or [] ] + unreadable_files = [str(item) for item in boundary.get("unreadable_files") or []] + missing_roots = [str(item) for item in boundary.get("missing_scan_roots") or []] + warnings = [str(item) for item in boundary.get("warnings") or []] + complete_ok = bool(boundary.get("ok")) and not ( + unreadable_directories or unreadable_files or missing_roots + ) + label = "completed with warnings" if warnings else "clean" check.update( { - "status": ( - "passed" - if boundary.get("ok") and not unreadable_directories - else "failed" - ), - "ok": bool(boundary.get("ok")) and not unreadable_directories, + "status": "passed" if complete_ok else "failed", + "ok": complete_ok, "duration_seconds": round(time.monotonic() - started, 3), "stdout_tail": ( - f"public boundary scan clean: {boundary.get('scanned_files')} files" - if boundary.get("ok") and not unreadable_directories + f"public boundary scan {label}: {boundary.get('scanned_files')} files" + if complete_ok else "" ), - "stderr_tail": "\n".join(hits + unreadable_directories)[-800:], + "stderr_tail": "\n".join(dict.fromkeys( + hits + warnings + unreadable_files + unreadable_directories + missing_roots + ))[-800:], "boundary": { "scanned_files": boundary.get("scanned_files"), "skipped_private_state_files": boundary.get("skipped_private_state_files"), diff --git a/loopx/cli_commands/quota_registration.py b/loopx/cli_commands/quota_registration.py index ac456517c0..b4b22de429 100644 --- a/loopx/cli_commands/quota_registration.py +++ b/loopx/cli_commands/quota_registration.py @@ -325,13 +325,13 @@ def register_quota_command( quota_parser.add_argument( "--scan-root", default=default_public_scan_root(), - help="Public files to scan for obvious private material. Defaults to the LoopX install root.", + help="Compatibility scope metadata; quota defers public scans. Use loopx check for outgoing content.", ) quota_parser.add_argument( "--scan-path", action="append", default=[], - help="Specific public file or directory to scan. Repeatable. Overrides --scan-root when set.", + help="Compatibility scope metadata. Repeatable; does not request a public scan.", ) quota_parser.add_argument( "--use-projection-cache", diff --git a/loopx/cli_commands/status_registration.py b/loopx/cli_commands/status_registration.py index 42982a56d8..d230efd315 100644 --- a/loopx/cli_commands/status_registration.py +++ b/loopx/cli_commands/status_registration.py @@ -39,8 +39,8 @@ def register_status_commands( "--scan-root", default=default_public_scan_root(), help=( - "Public files to scan for obvious private material. " - "Defaults to the LoopX install root." + "Compatibility scope metadata; operational status defers public scans. " + "Use loopx check for outgoing content." ), ) status_parser.add_argument( @@ -48,8 +48,7 @@ def register_status_commands( action="append", default=[], help=( - "Specific public file or directory to scan. Repeatable. " - "Overrides --scan-root when set." + "Compatibility scope metadata. Repeatable; does not request a public scan." ), ) status_parser.add_argument("--limit", type=int, default=5) @@ -143,8 +142,8 @@ def register_status_commands( "--scan-root", default=default_public_scan_root(), help=( - "Public files to scan for obvious private material. " - "Defaults to the LoopX install root." + "Compatibility scope metadata; operational status defers public scans. " + "Use loopx check for outgoing content." ), ) diagnose_parser.add_argument( @@ -152,8 +151,7 @@ def register_status_commands( action="append", default=[], help=( - "Specific public file or directory to scan. Repeatable. " - "Overrides --scan-root when set." + "Compatibility scope metadata. Repeatable; does not request a public scan." ), ) diagnose_parser.add_argument("--limit", type=int, default=5) @@ -186,8 +184,8 @@ def register_status_commands( "--scan-root", default=default_public_scan_root(), help=( - "Public files to scan for obvious private material. " - "Defaults to the LoopX install root." + "Compatibility scope metadata; operational status defers public scans. " + "Use loopx check for outgoing content." ), ) review_packet_parser.add_argument( @@ -195,8 +193,7 @@ def register_status_commands( action="append", default=[], help=( - "Specific public file or directory to scan. Repeatable. " - "Overrides --scan-root when set." + "Compatibility scope metadata. Repeatable; does not request a public scan." ), ) review_packet_parser.add_argument( diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py index 2cf6275c59..bb75d975fc 100644 --- a/loopx/cli_commands/turn_decision.py +++ b/loopx/cli_commands/turn_decision.py @@ -38,7 +38,6 @@ scheduler_execution_context_for_turn, ) from ..status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, collect_status -from ..paths import default_public_scan_root from ..control_plane.effect_runtime import effect_runtime_result from .lark_inbox import build_lark_operator_inbox_urgency_projector from .turn_selection import turn_controller_advisory_primary @@ -79,6 +78,7 @@ def collect_turn_status_payload( goal_id=args.goal_id, available_capabilities=args.available_capabilities, agent_lane_id=args.agent_id, + include_public_boundary_scan=False, ) @@ -338,10 +338,9 @@ def current_heartbeat_host_launch_decision( args = argparse.Namespace( goal_id=goal_id, agent_id=agent_id, turn_instance_id=original_turn, todo_id=None, codex_app=False, runtime_profile="generic_cli", - # Uncaptured wakes use heartbeat-prompt's public-package default. Keep - # the existing workspace recheck for an already-held checkpoint packet. - scan_path=[str(project)] if original_quota is not None else [], - scan_root=default_public_scan_root(), limit=0, + # Re-read execution authority, including held receipt identity. Public + # disclosure is independently checked by `loopx check`/premerge. + scan_path=[], scan_root=str(project), limit=0, available_capabilities=["shell", "filesystem_write"], ) selection = load_requested_quota_action_selection( diff --git a/loopx/cli_commands/turn_registration.py b/loopx/cli_commands/turn_registration.py index e538a426ea..4fafe05e80 100644 --- a/loopx/cli_commands/turn_registration.py +++ b/loopx/cli_commands/turn_registration.py @@ -93,13 +93,13 @@ def register_turn_commands( plan.add_argument( "--scan-root", default=default_public_scan_root(), - help="Public files to scan for obvious private material.", + help="Compatibility scope metadata; Turn execution defers public scans. Use loopx check before publishing.", ) plan.add_argument( "--scan-path", action="append", default=[], - help="Specific public file or directory to scan. Repeatable.", + help="Compatibility scope metadata. Repeatable; does not request a public scan.", ) plan.add_argument("--limit", type=int, default=5) @@ -153,13 +153,13 @@ def register_turn_commands( managed_step.add_argument( "--scan-root", default=default_public_scan_root(), - help="Public files to scan for obvious private material.", + help="Compatibility scope metadata; Turn execution defers public scans. Use loopx check before publishing.", ) managed_step.add_argument( "--scan-path", action="append", default=[], - help="Specific public file or directory to scan. Repeatable.", + help="Compatibility scope metadata. Repeatable; does not request a public scan.", ) managed_step.add_argument("--limit", type=int, default=5) @@ -366,7 +366,7 @@ def register_turn_commands( run_once.add_argument( "--scan-root", default=default_public_scan_root(), - help="Public files to scan for obvious private material.", + help="Compatibility scope metadata; Turn execution defers public scans. Use loopx check before publishing.", ) run_once.add_argument("--scan-path", action="append", default=[]) run_once.add_argument("--limit", type=int, default=5) diff --git a/loopx/contract.py b/loopx/contract.py index 19b0c0ad43..c0d5e2ce38 100644 --- a/loopx/contract.py +++ b/loopx/contract.py @@ -66,6 +66,9 @@ class LeakRule: pattern: re.Pattern[str] required_literals: tuple[str, ...] + # A rule match is evidence for review, not necessarily a secret. Keep + # severity with the existing classification rule, not the runtime gate. + blocking: bool = True def is_candidate(self, folded_text: str) -> bool: return any(literal in folded_text for literal in self.required_literals) @@ -151,6 +154,7 @@ def _prefilter_fold(text: str) -> str: r"|\b192\.168\.\d+\.\d+\b" ), required_literals=("10.", "172.", "192.168."), + blocking=False, ), } LEAK_PATTERNS = {name: rule.pattern for name, rule in LEAK_RULES.items()} @@ -999,14 +1003,15 @@ def iter_scan_files( def scan_public_boundary( scan_roots: list[Path], *, registry: dict[str, Any] | None = None ) -> dict[str, Any]: - """Scan the requested trees and report what was found, and what was not read. + """Check selected outgoing content, preserving findings and unread inputs. - ``ok`` answers only "no private material was found in the files this scan - read", so a caller that must fail closed has to act on ``missing_scan_roots`` - and ``unreadable_directories`` as well; ``check_contract`` does that for the - shipped CLI entry points. + ``hits`` retains rule locations; ``blocking_hits`` excludes ambiguous IP + advisories. ``ok`` requires no blocker or unread input. It + never certifies an unrequested file, generated text or future publication. """ hits: list[str] = [] + blocking_hits: list[str] = [] + advisory_hits: list[str] = [] allowed_hits: list[str] = [] private_state_git_warnings: list[str] = [] skipped_private_state_files: list[str] = [] @@ -1018,6 +1023,11 @@ def scan_public_boundary( file_roots: dict[Path, Path] = {} directory_roots: dict[Path, Path] = {} private_files: set[Path] = set() + + def record_hit(hit: str, *, blocking: bool = True) -> None: + hits.append(hit) + (blocking_hits if blocking else advisory_hits).append(hit) + for scan_root in scan_roots: resolved_scan_root = scan_root.resolve() if not resolved_scan_root.exists(): @@ -1038,6 +1048,11 @@ def scan_public_boundary( classification_root = ( _git_worktree_root(resolved_scan_root) or Path(resolved_scan_root.anchor) ) + if _is_local_private_state_path(resolved_scan_root, classification_root): + # Explicit outgoing selection cannot turn an ignored private file + # or a pruned private directory into a successful empty scan. + record_hit(f"{rel_or_abs(resolved_scan_root, classification_root)}: local_private_state") + continue for file_path in iter_scan_files( resolved_scan_root, unreadable_directories=unreadable_directories ): @@ -1069,16 +1084,19 @@ def public_files() -> Iterator[Path]: ) continue private_file_git[path] = git + record_hit(f"{rel_or_abs(path, root)}: local_private_state") yield path for read in iter_utf8_file_reads(public_files()): path = read.path root = file_roots.get(path, path) git: dict[str, Any] | None = private_file_git.pop(path, None) - if isinstance(read.error, UnicodeDecodeError): - continue if read.error is not None: - unreadable_files.append(f"{rel_or_abs(path, root)}: {read.error.strerror or read.error}") + detail = ( + read.error.strerror or type(read.error).__name__ + if isinstance(read.error, OSError) else "invalid UTF-8" + ) + unreadable_files.append(f"{rel_or_abs(path, root)}: {detail}") continue text = read.text assert text is not None @@ -1100,7 +1118,7 @@ def public_files() -> Iterator[Path]: and not resolved.startswith(_PUBLIC_NPM_REGISTRY_PREFIX) ): display_package = package_path or "" - hits.append( + record_hit( f"{rel_or_abs(path, root)}: " "non_public_package_registry " f"({display_package})" @@ -1134,9 +1152,9 @@ def public_files() -> Iterator[Path]: if git is not None and _hit_allowed_by_policy(name, git, policy): allowed_hits.append(hit) else: - hits.append(hit) + record_hit(hit, blocking=rule.blocking) return { - "ok": not hits, + "ok": not (blocking_hits or missing_scan_roots or unreadable_files or unreadable_directories), "scan_roots": [str(path) for path in scan_roots], "files": len(files), "scanned_files": len(files) - len(skipped_private_state_files), @@ -1153,6 +1171,8 @@ def public_files() -> Iterator[Path]: "private_state_git_warnings": private_state_git_warnings, "policy": policy, "hits": hits, + "blocking_hits": blocking_hits, + "warnings": advisory_hits + credential_reference_hits, } @@ -1319,12 +1339,18 @@ def add_global_error(code: str, message: str) -> None: unreadable_directories = [ str(item) for item in boundary.get("unreadable_directories") or [] ] + unreadable_files = [str(item) for item in boundary.get("unreadable_files") or []] + boundary_warnings = [str(item) for item in boundary.get("warnings") or []] public_boundary_scan = { "state": "completed", - "ok": bool(boundary.get("ok")) and not missing_scan_roots, + "ok": bool(boundary.get("ok")) and not ( + missing_scan_roots or unreadable_directories or unreadable_files + ), "scanned_files": int(boundary.get("scanned_files") or 0), "missing_scan_roots": missing_scan_roots, "unreadable_directories": unreadable_directories, + "unreadable_files": unreadable_files, + "warnings": boundary_warnings, } for missing_root in missing_scan_roots: add_global_error( @@ -1336,11 +1362,18 @@ def add_global_error(code: str, message: str) -> None: "public_boundary_scan_directory_unreadable", f"scan directory could not be listed: {unreadable_directory}", ) - if boundary.get("ok") and not missing_scan_roots and not unreadable_directories: - checks.append(f"public boundary scan clean: {boundary.get('scanned_files')} files") + for unreadable_file in unreadable_files: + add_global_error( + "public_boundary_scan_file_unreadable", + f"scan file could not be read: {unreadable_file}", + ) + if public_boundary_scan["ok"]: + label = "completed with warnings" if boundary_warnings else "clean" + checks.append(f"public boundary scan {label}: {boundary.get('scanned_files')} files") else: - for hit in boundary.get("hits") or []: + for hit in boundary.get("blocking_hits", boundary.get("hits")) or []: add_global_error("public_boundary_violation", str(hit)) + warnings.extend(boundary_warnings) if boundary.get("skipped_private_state_files"): checks.append( "private state scan skipped: " @@ -1357,9 +1390,6 @@ def add_global_error(code: str, message: str) -> None: "credential references downgraded: " f"{len(boundary.get('credential_reference_hits') or [])} non-literal hits" ) - warnings.extend( - f"unreadable file skipped: {item}" for item in boundary.get("unreadable_files") or [] - ) else: public_boundary_scan = { "state": "deferred", diff --git a/loopx/control_plane/status/collection.py b/loopx/control_plane/status/collection.py index 89df438f2d..181ec9ecbb 100644 --- a/loopx/control_plane/status/collection.py +++ b/loopx/control_plane/status/collection.py @@ -79,7 +79,7 @@ def collect_status( include_task_graph: bool = False, goal_id: str | None = None, available_capabilities: Any = None, - include_public_boundary_scan: bool = True, + include_public_boundary_scan: bool = False, recent_run_limit: int | None = None, include_goal_subagent_configuration: bool = False, include_zcode_goal_eligibility: bool = False, diff --git a/loopx/status.py b/loopx/status.py index c505bbc574..f66731ed97 100644 --- a/loopx/status.py +++ b/loopx/status.py @@ -1356,7 +1356,7 @@ def collect_status( include_task_graph: bool = False, goal_id: str | None = None, available_capabilities: Any = None, - include_public_boundary_scan: bool = True, + include_public_boundary_scan: bool = False, recent_run_limit: int | None = None, include_goal_subagent_configuration: bool = False, include_zcode_goal_eligibility: bool = False, diff --git a/tests/canary/test_premerge_public_boundary_run.py b/tests/canary/test_premerge_public_boundary_run.py index 37804881ae..f2a5fc60d9 100644 --- a/tests/canary/test_premerge_public_boundary_run.py +++ b/tests/canary/test_premerge_public_boundary_run.py @@ -78,3 +78,23 @@ def test_clean_boundary_passes_the_premerge_boundary_check( assert check["stdout_tail"] == "public boundary scan clean: 3 files" assert check["boundary"]["unreadable_directory_count"] == 0 assert run["ok"] is True + + +def test_actual_premerge_keeps_example_warnings_visible(tmp_path: Path) -> None: + (tmp_path / "example.md").write_text("Synthetic network: http://192.168.10.10\n") + run = _public_boundary_changed_files_run(changed_files=["example.md"], + execute=True, timeout_seconds=30.0, repo_root=tmp_path) + check = run["selected_checks"][0] + assert run["ok"] is True and check["ok"] is True + assert "completed with warnings" in check["stdout_tail"] + assert "example.md:1: private_ip" in check["stderr_tail"] + + +def test_actual_premerge_rejects_unread_content(tmp_path: Path) -> None: + (tmp_path / "unknown.md").write_bytes(b"\xff\xfe") + run = _public_boundary_changed_files_run(changed_files=["unknown.md"], + execute=True, timeout_seconds=30.0, repo_root=tmp_path) + assert run["ok"] is False + check = run["selected_checks"][0] + assert check["stdout_tail"] == "" + assert "invalid UTF-8" in check["stderr_tail"] diff --git a/tests/control_plane/test_benchmark_checkpoint_admission.py b/tests/control_plane/test_benchmark_checkpoint_admission.py index bb3ff3e06d..3e697dd185 100644 --- a/tests/control_plane/test_benchmark_checkpoint_admission.py +++ b/tests/control_plane/test_benchmark_checkpoint_admission.py @@ -2,7 +2,6 @@ from __future__ import annotations import json -from ipaddress import IPv4Address import os from pathlib import Path import sys @@ -19,7 +18,7 @@ from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected from tests.control_plane import test_quota_settlement_cli as fixture from tests.control_plane.host_process_fixture import write_python_host_fixture -from tests.control_plane.test_benchmark_current_admission import _environment, _quota +from tests.control_plane.test_benchmark_current_admission import _environment, _private_workspace, _quota from tests.control_plane.test_codex_operation_launch import _bindings @@ -116,7 +115,7 @@ def rendered(command): @pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("refusal", ["source_compute_zero", "missing_state", "stop", "original_event_changed", "public_scope_violation", None]) +@pytest.mark.parametrize("refusal", ["source_compute_zero", "missing_state", "stop", "original_event_changed", None]) def test_checkpoint_resume_rechecks_current_source_with_original_guard(tmp_path, monkeypatch, provider, refusal): _run_checkpoint(tmp_path, monkeypatch, provider, refusal=refusal, todo_bound=False) @@ -126,12 +125,18 @@ def test_checkpoint_todo_bound_closeout_keeps_original_host(tmp_path, monkeypatc _run_checkpoint(tmp_path, monkeypatch, provider, refusal=None, todo_bound=True) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_checkpoint_resume_keeps_private_workspace_and_original_receipt(tmp_path, monkeypatch, provider): + _run_checkpoint(tmp_path, monkeypatch, provider, + refusal=None, todo_bound=False, workspace_change=True) + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_checkpoint_unbound_guard_selects_existing_work_in_original_turn(tmp_path, monkeypatch, provider): _run_checkpoint(tmp_path, monkeypatch, provider, refusal=None, todo_bound=False, unbound=True) -def _run_checkpoint(tmp_path, monkeypatch, provider, *, refusal, todo_bound, unbound=False): +def _run_checkpoint(tmp_path, monkeypatch, provider, *, refusal, todo_bound, unbound=False, workspace_change=False): unbound = unbound or refusal == "original_event_changed" env, project, runtime, source, state = _environment( tmp_path, monkeypatch, provider, replan="todo" if todo_bound or unbound else True) @@ -179,6 +184,9 @@ def captured_then_changed(held, deadline): if todo_bound: assert held["quota"]["heartbeat_receipt"]["settlement_identity"]["todo_id"] == fixture.SELECTED_REPLAN_TODO_ID env["FIXTURE_HELD_QUOTA"] = str(held["path"]) + if workspace_change: + _private_workspace(project) + assert _quota(source, runtime, project)["status_health_ok"] is True if refusal == "source_compute_zero": changed = configure_goal(registry_path=source, goal_id=fixture.GOAL_ID, quota_compute=0, execute=True) assert changed["written"], changed @@ -187,13 +195,6 @@ def captured_then_changed(held, deadline): elif refusal == "missing_state": state.rename(state.with_suffix(".saved")) assert _quota(source, runtime, project)["status_health_ok"] is False - elif refusal == "public_scope_violation": - # The held guard explicitly scanned this workspace. A default - # package scan must not replace that check on continuation. - (project / "changed_public_fixture.py").write_text( - "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n" - ) - assert _quota(source, runtime, project)["status_health_ok"] is False elif refusal == "stop": with monkeypatch.context() as fault: fault.setattr(activation_service, "sync_project_registry_to_global", lambda **kwargs: { diff --git a/tests/control_plane/test_benchmark_current_admission.py b/tests/control_plane/test_benchmark_current_admission.py index 0384f8bd39..dbd5dd3ed3 100644 --- a/tests/control_plane/test_benchmark_current_admission.py +++ b/tests/control_plane/test_benchmark_current_admission.py @@ -2,7 +2,6 @@ from __future__ import annotations import json -from ipaddress import IPv4Address from pathlib import Path import sys import time @@ -15,16 +14,29 @@ from loopx.cli_commands import turn_decision from loopx.control_plane.goals import activation_service from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected +from loopx import contract, status as status_owner +from loopx.cli import main from loopx.global_registry import sync_project_registry_to_global from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime from tests.control_plane.host_process_fixture import write_python_host_fixture from tests.control_plane import test_quota_settlement_cli as fixture from tests.control_plane.test_codex_operation_launch import _bindings +from tests.test_contract_credential_pattern_precision import PRIVATE_LARK_TENANT_HOST ROOT = Path(__file__).resolve().parents[2] +def _private_workspace(project): + path = project / "task_network_fixture.py" + path.write_text( + 'endpoint = "http://192.168.10.10"\n' + 'auth = "Authorization: Bearer ${SERVICE_TOKEN}"\n' + f'link = "https://{PRIVATE_LARK_TENANT_HOST}/wiki/example"\n' + ) + return path + + def _environment(tmp_path, monkeypatch, provider, *, replan=False): isolate_sqlite_runtime(tmp_path, monkeypatch) monkeypatch.setenv("LOOPX_USAGE_PING", "0") @@ -82,11 +94,7 @@ def _quota(registry, runtime, project): def test_heartbeat_worker_qualifies_current_source_at_actual_spawn(tmp_path, monkeypatch, provider, refusal): env, project, runtime, source, state = _environment(tmp_path, monkeypatch, provider) if refusal is None: - # This default-scan caller may work on a private task. Keep that fixture - # out of the shared environment used by explicit project-scan callers. - (project / "task_network_fixture.py").write_text( - "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n" - ) + _private_workspace(project) mirror = Path(env["LOOPX_REGISTRY"]) mirror_before, state_before = mirror.read_bytes(), state.read_bytes() observations = [] @@ -149,9 +157,7 @@ def test_initial_benchmark_planner_can_launch_without_a_prior_heartbeat_receipt( env, project, runtime, source, _state = _environment(tmp_path, monkeypatch, provider, replan="empty") packet = _quota(source, runtime, project) assert packet["should_run"] is True and packet.get("selected_todo") is None - (project / "task_network_fixture.py").write_text( - "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n" - ) + _private_workspace(project) env.update(LOOPX_TASK_ENTRY="loopx-planned", LOOPX_TASK_STAGE="plan", LOOPX_EXECUTION_MODE=mode, LOOPX_PHASE_DEADLINE_EPOCH=str(time.time()+600), LOOPX_PLANNING_RESULT=str(project / "planning.json")) if mode == "turn": @@ -181,21 +187,52 @@ def test_initial_benchmark_planner_can_launch_without_a_prior_heartbeat_receipt( @pytest.mark.parametrize("provider", ["file", "sqlite"]) -def test_heartbeat_host_launch_retains_public_scan_refusal(tmp_path, monkeypatch, provider): - env, project, runtime, source, _state = _environment(tmp_path, monkeypatch, provider) - public = tmp_path / "selected-public" - public.mkdir() - (public / "unsafe_fixture.py").write_text( - "endpoint = " + repr("http://" + str(IPv4Address(0x0A000001))) + "\n" - ) - code, original = fixture._run_cli(source, runtime, - "quota", "should-run", "--goal-id", fixture.GOAL_ID, "--agent-id", fixture.AGENT_ID, - "--scan-path", str(public), cwd=project) - assert code == 1 and original["status_health_ok"] is False and original["should_run"] is False - monkeypatch.setattr(turn_decision, "default_public_scan_root", lambda: str(public)) - current = turn_decision.current_heartbeat_host_launch_decision( - registry_path=Path(env["LOOPX_REGISTRY"]), runtime_root=runtime, project=project, - goal_id=fixture.GOAL_ID, agent_id=fixture.AGENT_ID, turn_instance_id="public-scan-refusal", - )["current"] - assert current["status_health_ok"] is False and current["should_run"] is False +def test_operational_cli_and_launch_defer_scans_but_explicit_check_blocks_private_content( + tmp_path, monkeypatch, provider, capsys, +): + _env, project, runtime, source, state = _environment(tmp_path, monkeypatch, provider) + outgoing = _private_workspace(project) + before = (source.read_bytes(), state.read_bytes()) + monkeypatch.chdir(project) + base = ["--format", "json", "--registry", str(source), "--runtime-root", str(runtime)] + observed = [] + real_check = status_owner.check_contract + + def check_execution(*args, **kwargs): + result = real_check(*args, **kwargs) + observed.append(result["public_boundary_scan"]) + return result + + def unexpected_scan(*args, **kwargs): + pytest.fail("ordinary execution traversed or read public-scan inputs") + + with monkeypatch.context() as instrumentation: + instrumentation.setattr(status_owner, "check_contract", check_execution) + instrumentation.setattr(contract, "scan_public_boundary", unexpected_scan) + instrumentation.setattr(contract, "iter_scan_files", unexpected_scan) + instrumentation.setattr(contract, "iter_utf8_file_reads", unexpected_scan) + assert main([*base, "status", "--goal-id", fixture.GOAL_ID, + "--scan-path", str(project)]) == 0 + assert json.loads(capsys.readouterr().out)["contract"]["ok"] is True + args = [*base, "quota", "should-run", "--goal-id", fixture.GOAL_ID, + "--agent-id", fixture.AGENT_ID, "--runtime-profile", "generic_cli", + "--turn-instance-id", "private-workspace-turn", "--scan-path", str(project)] + assert main(args) == 0 + original = json.loads(capsys.readouterr().out) + events = fixture._heartbeat_receipt_events(runtime, "private-workspace-turn") + for held in (None, original): + current = turn_decision.current_heartbeat_host_launch_decision( + registry_path=source, runtime_root=runtime, project=project, + goal_id=fixture.GOAL_ID, agent_id=fixture.AGENT_ID, + turn_instance_id="private-workspace-turn", original_quota=held, + )["current"] + assert current["status_health_ok"] and current["should_run"] + assert observed and all(row["state"] == "deferred" for row in observed) + assert fixture._heartbeat_receipt_events(runtime, "private-workspace-turn") == events + assert main([*base, "check", "--scan-path", str(outgoing)]) == 1 + checked = json.loads(capsys.readouterr().out) + assert not checked["ok"] and checked["public_boundary_scan"]["state"] == "completed" + assert any("private_doc_url" in str(error) for error in checked["errors"]) + assert (source.read_bytes(), state.read_bytes()) == before assert not (project / "native-starts").exists() and not (project / "model-starts").exists() + assert fixture._spend_run_count(runtime) == 0 diff --git a/tests/control_plane/test_public_boundary_parallel_reads.py b/tests/control_plane/test_public_boundary_parallel_reads.py index 0a5973b6e1..4653dc1944 100644 --- a/tests/control_plane/test_public_boundary_parallel_reads.py +++ b/tests/control_plane/test_public_boundary_parallel_reads.py @@ -185,7 +185,7 @@ def test_full_scan_projection_matches_serial_reads_and_observes_later_changes( assert fresh["files"] == 3 -def test_tracked_private_policy_and_warning_order_are_preserved( +def test_tracked_url_policy_does_not_authorize_private_state_export( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state = tmp_path / ".codex" / "ACTIVE_GOAL_STATE.md" @@ -200,7 +200,8 @@ def test_tracked_private_policy_and_warning_order_are_preserved( payload = contract.scan_public_boundary([tmp_path], registry={ "public_boundary": {"tracked_private_doc_urls": "allow"}, }) - assert payload["ok"] is True + assert payload["ok"] is False + assert payload["blocking_hits"] == [".codex/ACTIVE_GOAL_STATE.md: local_private_state"] assert payload["allowed_hits"] == [".codex/ACTIVE_GOAL_STATE.md:1: private_doc_url"] assert payload["skipped_private_state_files"] == [".codex/untracked.md"] assert payload["private_state_git_warnings"] == [ @@ -239,9 +240,10 @@ def read(path: Path, *args, **kwargs) -> str: monkeypatch.setattr(Path, "read_text", read) payload = contract.scan_public_boundary(scan_roots) - assert payload["ok"] is True + assert payload["ok"] is False + assert any(hit.endswith(": local_private_state") for hit in payload["blocking_hits"]) assert payload["scanned_files"] == (0 if roots == "alone" else 1) - assert len(payload["skipped_private_state_files"]) == 1 + assert len(payload["skipped_private_state_files"]) == (0 if roots == "alone" else 1) assert opened == ([] if roots == "alone" else [public.resolve()]) @@ -279,13 +281,18 @@ def test_overlapping_roots_preserve_real_tracked_private_policy( if reverse: roots.reverse() blocked = contract.scan_public_boundary(roots) - assert blocked["hits"] == [".codex/notes/tracked.md:1: private_doc_url"] + assert blocked["hits"] == [ + ".codex/notes: local_private_state", + ".codex/notes/tracked.md: local_private_state", + ".codex/notes/tracked.md:1: private_doc_url", + ] assert blocked["skipped_private_state_files"] == [".codex/notes/untracked.md"] allowed = contract.scan_public_boundary(roots, registry={ "public_boundary": {"tracked_private_doc_urls": "allow"}, }) - assert allowed["ok"] is True - assert allowed["allowed_hits"] == blocked["hits"] + assert allowed["ok"] is False + assert allowed["allowed_hits"] == [".codex/notes/tracked.md:1: private_doc_url"] + assert allowed["blocking_hits"] == blocked["blocking_hits"][:2] assert allowed["skipped_private_state_files"] == blocked["skipped_private_state_files"] @@ -298,10 +305,12 @@ def test_private_alias_retarget_is_classified_again_on_the_next_scan(tmp_path: P alias = tmp_path / "alias.md" alias.symlink_to(private) first = contract.scan_public_boundary([alias]) + assert first["ok"] is False and first["blocking_hits"] assert first["scanned_files"] == 0 alias.unlink() alias.symlink_to(public) second = contract.scan_public_boundary([alias]) + assert second["ok"] is True assert second["scanned_files"] == 1 assert second["skipped_private_state_files"] == [] @@ -312,5 +321,6 @@ def test_git_marker_alone_cannot_hide_private_scan_ancestry(tmp_path: Path) -> N (private.parent / ".git").mkdir() private.write_text("synthetic private content", encoding="utf-8") payload = contract.scan_public_boundary([private]) + assert payload["ok"] is False and payload["blocking_hits"] assert payload["scanned_files"] == 0 - assert len(payload["skipped_private_state_files"]) == 1 + assert payload["skipped_private_state_files"] == [] diff --git a/tests/test_contract_scan_unreadable_directories.py b/tests/test_contract_scan_unreadable_directories.py index 2f01a21dea..17ad2ace1e 100644 --- a/tests/test_contract_scan_unreadable_directories.py +++ b/tests/test_contract_scan_unreadable_directories.py @@ -65,7 +65,7 @@ def test_scan_records_directory_it_cannot_list(tmp_path: Path) -> None: assert boundary["unreadable_directories"] == ["notes: Permission denied"] assert boundary["unreadable_files"] == [] assert boundary["hits"] == [] - assert boundary["ok"] is True + assert boundary["ok"] is False assert boundary["scanned_files"] == 1 @@ -129,7 +129,7 @@ def test_check_contract_fails_closed_on_unreadable_directory(tmp_path: Path) -> assert contract["public_boundary_scan"]["unreadable_directories"] == [ "notes: Permission denied" ] - assert contract["public_boundary_scan"]["ok"] is True + assert contract["public_boundary_scan"]["ok"] is False def test_readable_tree_records_no_unreadable_directory(tmp_path: Path) -> None: @@ -259,7 +259,7 @@ def walk_with_listing_failure(root: str, *, onerror: object = None): assert boundary["unreadable_directories"] == ["notes: Permission denied"] assert boundary["unreadable_files"] == [] - assert boundary["ok"] is True + assert boundary["ok"] is False def test_check_contract_fails_closed_on_a_walk_listing_failure( diff --git a/tests/test_contract_scan_unreadable_files.py b/tests/test_contract_scan_unreadable_files.py index fa22eabf0e..f38716d1ca 100644 --- a/tests/test_contract_scan_unreadable_files.py +++ b/tests/test_contract_scan_unreadable_files.py @@ -81,7 +81,7 @@ def test_scan_public_boundary_reports_unreadable_file(tmp_path: Path) -> None: finally: locked.chmod(0o644) - assert payload["ok"] is True + assert payload["ok"] is False assert payload["unreadable_files"] == ["locked.md: Permission denied"] diff --git a/tests/test_public_check_disclosure.py b/tests/test_public_check_disclosure.py new file mode 100644 index 0000000000..5960644b06 --- /dev/null +++ b/tests/test_public_check_disclosure.py @@ -0,0 +1,100 @@ +"""Explicit outgoing checks distinguish blockers, advisories and unread scope.""" +from __future__ import annotations + +import subprocess + +import pytest + +from loopx import contract +from tests.test_contract_scan_missing_roots import _registry +from tests.test_contract_credential_pattern_precision import ( + AUTH_HEADER, BEARER, PRIVATE_LARK_TENANT_HOST, REAL_BEARER_VALUE, +) + + +def test_selected_examples_keep_locations_without_claiming_a_clean_scan(tmp_path): + outgoing = tmp_path / "request.txt" + outgoing.write_text( + 'endpoint = "http://192.168.10.10"\n' + 'auth = "Authorization: Bearer ${SERVICE_TOKEN}"\n' + ) + checked = contract.check_contract(registry_path=_registry(tmp_path), + runtime_root_override=None, scan_roots=[outgoing], limit=0) + assert checked["ok"] is True + assert checked["public_boundary_scan"]["state"] == "completed" + assert checked["public_boundary_scan"]["ok"] is True + assert set(checked["public_boundary_scan"]["warnings"]) == { + "request.txt:1: private_ip", "request.txt:2: credential", + } + assert all(hit in checked["warnings"] for hit in checked["public_boundary_scan"]["warnings"]) + assert not any("scan clean" in text for text in checked["checks"]) + + +@pytest.mark.parametrize("text", [ + f"{AUTH_HEADER} {BEARER} {REAL_BEARER_VALUE}", + f"https://{PRIVATE_LARK_TENANT_HOST}/wiki/example", +]) +def test_selected_literal_credentials_and_private_links_still_block(tmp_path, text): + outgoing = tmp_path / "request.txt" + outgoing.write_text(text) + checked = contract.check_contract(registry_path=_registry(tmp_path), + runtime_root_override=None, scan_roots=[outgoing], limit=0) + assert checked["ok"] is False + assert checked["public_boundary_scan"]["ok"] is False + assert any(row["code"] == "public_boundary_violation" for row in checked["error_diagnostics"]) + + +@pytest.mark.parametrize("tracked", [False, True]) +@pytest.mark.parametrize("directory", [False, True]) +def test_explicit_private_selection_cannot_hide_behind_git_exclusion( + tmp_path, monkeypatch, tracked, directory, +): + subprocess.run(["git", "init", "-q", str(tmp_path)], check=True) + (tmp_path / ".gitignore").write_text(".local/\n") + private = tmp_path / ".local" / "notes.md" + private.parent.mkdir() + private.write_text("Owner-only synthetic notes.\n") + if tracked: + subprocess.run(["git", "-C", str(tmp_path), "add", "-f", ".local/notes.md"], check=True) + + def must_not_read(paths): + assert list(paths) == [], "explicit private scope was opened before rejection" + return iter(()) + + monkeypatch.setattr(contract, "iter_utf8_file_reads", must_not_read) + checked = contract.scan_public_boundary([private.parent if directory else private]) + assert checked["ok"] is False + assert checked["blocking_hits"] and all("local_private_state" in hit for hit in checked["hits"]) + assert checked["scanned_files"] == 0 + + +def test_tracked_private_state_is_not_public_even_without_content_matches(tmp_path): + subprocess.run(["git", "init", "-q", str(tmp_path)], check=True) + private = tmp_path / "ACTIVE_GOAL_STATE.md" + private.write_text("Owner-only synthetic notes.\n") + subprocess.run(["git", "-C", str(tmp_path), "add", private.name], check=True) + checked = contract.scan_public_boundary([tmp_path]) + assert checked["ok"] is False + assert checked["blocking_hits"] == ["ACTIVE_GOAL_STATE.md: local_private_state"] + + +def test_public_check_reobserves_withdrawn_tracked_source_permission(tmp_path): + subprocess.run(["git", "init", "-q", str(tmp_path)], check=True) + outgoing = tmp_path / "reference.md" + outgoing.write_text(f"https://{PRIVATE_LARK_TENANT_HOST}/wiki/example") + subprocess.run(["git", "-C", str(tmp_path), "add", outgoing.name], check=True) + allowed = contract.scan_public_boundary([outgoing], + registry={"public_boundary": {"tracked_private_doc_urls": "allow"}}) + assert allowed["ok"] and allowed["allowed_hits"] + withdrawn = contract.scan_public_boundary([outgoing], registry={}) + assert not withdrawn["ok"] and withdrawn["blocking_hits"] + + +def test_invalid_utf8_is_unexamined_and_never_clean(tmp_path): + outgoing = tmp_path / "request.md" + outgoing.write_bytes(b"\xff\xfe") + checked = contract.check_contract(registry_path=_registry(tmp_path), + runtime_root_override=None, scan_roots=[outgoing], limit=0) + assert checked["ok"] is False + assert checked["public_boundary_scan"]["unreadable_files"] == ["request.md: invalid UTF-8"] + assert not any("scan clean" in text for text in checked["checks"]) From 7c106a2affe7c0e6a9f03c5e1c847441c6188849 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 21 Oct 2026 19:15:31 +0800 Subject: [PATCH 4/4] test(runtime): preserve original held receipt and action fences Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_benchmark_checkpoint_admission.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/tests/control_plane/test_benchmark_checkpoint_admission.py b/tests/control_plane/test_benchmark_checkpoint_admission.py index 3e697dd185..17f016c9f4 100644 --- a/tests/control_plane/test_benchmark_checkpoint_admission.py +++ b/tests/control_plane/test_benchmark_checkpoint_admission.py @@ -115,7 +115,8 @@ def rendered(command): @pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("refusal", ["source_compute_zero", "missing_state", "stop", "original_event_changed", None]) +@pytest.mark.parametrize("refusal", ["source_compute_zero", "missing_state", "stop", + "original_event_changed", "original_settlement_changed", "original_action_changed", None]) def test_checkpoint_resume_rechecks_current_source_with_original_guard(tmp_path, monkeypatch, provider, refusal): _run_checkpoint(tmp_path, monkeypatch, provider, refusal=refusal, todo_bound=False) @@ -210,6 +211,14 @@ def captured_then_changed(held, deadline): # unbound receipt merely because both have no selected Todo. held["quota"] = {**held["quota"], "heartbeat_receipt": { **original_receipt, "event_id": "different-committed-guard-event"}} + elif refusal == "original_settlement_changed": + original_receipt = held["quota"]["heartbeat_receipt"] + assert original_receipt["settlement_identity"] + held["quota"] = {**held["quota"], "heartbeat_receipt": { + **original_receipt, "settlement_identity": { + **original_receipt["settlement_identity"], "effect_id": "different-original-effect"}}} + elif refusal == "original_action_changed": + held["quota"] = {**held["quota"], "effective_action": "skip"} return captured monkeypatch.setattr(hook, "_capture", captured_then_changed) @@ -246,7 +255,8 @@ def captured_then_changed(held, deadline): assert error is not None and { "source_compute_zero":"goal_launch_not_allowed", "missing_state":"goal_launch_not_allowed", "stop":"goal_stopped", "original_event_changed":"heartbeat_launch_original_binding_changed", - "public_scope_violation":"goal_launch_not_allowed", + "original_settlement_changed":"heartbeat_launch_original_binding_changed", + "original_action_changed":"goal_launch_not_allowed", }[refusal] in str(error) assert not (project / "checkpoint-business.json").exists() assert fixture._heartbeat_receipt_events(runtime, HELD_TURN) == observations["events"]