From 37a6289bf4b4a1e3793eaabb7569b88519c8063f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 25 Oct 2026 02:57:29 +0800 Subject: [PATCH] fix(context-provider): support OpenViking content writes without body timeout Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../context_providers/openviking.py | 47 +++++- .../test_reward_memory_provider_retrieval.py | 158 ++++++++++++++++++ 2 files changed, 199 insertions(+), 6 deletions(-) diff --git a/loopx/capabilities/context_providers/openviking.py b/loopx/capabilities/context_providers/openviking.py index d97b2ad34d..1ff769f935 100644 --- a/loopx/capabilities/context_providers/openviking.py +++ b/loopx/capabilities/context_providers/openviking.py @@ -315,6 +315,7 @@ def _run( args: Sequence[str], *, timeout_seconds: float, + deadline: float | None = None, ) -> subprocess.CompletedProcess[str]: command = [self.executable] if ( @@ -329,13 +330,18 @@ def _run( ): command.extend(["--actor-peer-id", self.actor_peer_id]) command.extend(args) + timeout = max(1.0, timeout_seconds) + if deadline is not None: + timeout = min(timeout, deadline - time.monotonic()) + if timeout <= 0: + raise subprocess.TimeoutExpired(command, timeout_seconds) return self.runner( command, text=True, encoding="utf-8", stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - timeout=max(1.0, timeout_seconds), + timeout=timeout, env=self.env, check=False, ) @@ -345,9 +351,12 @@ def _preflight( *, timeout_seconds: float, peer_scope_required: bool = False, + deadline: float | None = None, ) -> tuple[str | None, str | None]: try: - version_result = self._run(["--version"], timeout_seconds=timeout_seconds) + version_result = self._run( + ["--version"], timeout_seconds=timeout_seconds, deadline=deadline + ) except (FileNotFoundError, subprocess.TimeoutExpired): return None, "provider_cli_unavailable" if version_result.returncode != 0: @@ -367,7 +376,7 @@ def _preflight( if peer_scope_required: try: server_version_result = self._run( - ["version"], timeout_seconds=timeout_seconds + ["version"], timeout_seconds=timeout_seconds, deadline=deadline ) except subprocess.TimeoutExpired: return version, "provider_server_version_timeout" @@ -390,6 +399,7 @@ def _preflight( status_result = self._run( ["status", "-o", "json", "-c", "true"], timeout_seconds=timeout_seconds, + deadline=deadline, ) except subprocess.TimeoutExpired: return version, "provider_service_timeout" @@ -620,6 +630,7 @@ def _reconcile_uncertain_sync_write( target: str, source_content: str, timeout_seconds: float, + deadline: float | None = None, ) -> str: """Classify an uncertain write without retrying it. @@ -628,10 +639,17 @@ def _reconcile_uncertain_sync_write( the immutable target before deciding whether another write is safe. """ + reconciliation_deadline = time.monotonic() + max(0.0, timeout_seconds) + deadline = ( + min(deadline, reconciliation_deadline) + if deadline is not None + else reconciliation_deadline + ) try: exact = self._run( ["read", target, "-o", "json", "-c", "true"], timeout_seconds=timeout_seconds, + deadline=deadline, ) except subprocess.TimeoutExpired: return "reconciliation_unavailable" @@ -648,6 +666,7 @@ def _reconcile_uncertain_sync_write( tree = self._run( ["tree", target, "-L", "3", "-o", "json", "-c", "true"], timeout_seconds=timeout_seconds, + deadline=deadline, ) except subprocess.TimeoutExpired: return "reconciliation_unavailable" @@ -667,6 +686,7 @@ def _reconcile_uncertain_sync_write( read = self._run( ["read", ref, "-o", "json", "-c", "true"], timeout_seconds=timeout_seconds, + deadline=deadline, ) except subprocess.TimeoutExpired: continue @@ -702,6 +722,7 @@ def _reconcile_uncertain_sync_write( listing = self._run( ["ls", parent, "-n", "100", "-o", "json", "-c", "true"], timeout_seconds=timeout_seconds, + deadline=deadline, ) except subprocess.TimeoutExpired: return "reconciliation_unavailable" @@ -731,6 +752,7 @@ def sync( execute: bool, ) -> ContextProviderSync: started = time.monotonic() + deadline = started + max(0.0, timeout_seconds) namespace = _compact(namespace, limit=120) if len(resources) > MAX_OPENVIKING_SYNC_RESOURCES: raise ValueError( @@ -754,6 +776,7 @@ def sync( version, blocker = self._preflight( timeout_seconds=timeout_seconds, peer_scope_required=any(scope.actor_binding_required for scope in scopes), + deadline=deadline, ) if blocker: return ContextProviderSync( @@ -780,6 +803,7 @@ def sync( existing = self._run( ["read", target, "-o", "json", "-c", "true"], timeout_seconds=remaining, + deadline=deadline, ) except subprocess.TimeoutExpired: preflight_reason = "provider_sync_target_preflight_timeout" @@ -792,6 +816,7 @@ def sync( parent_probe = self._run( ["ls", parent, "-n", "1", "-o", "json", "-c", "true"], timeout_seconds=remaining, + deadline=deadline, ) except subprocess.TimeoutExpired: preflight_reason = "provider_sync_target_preflight_timeout" @@ -860,6 +885,7 @@ def sync( existing = self._run( ["read", target, "-o", "json", "-c", "true"], timeout_seconds=remaining, + deadline=deadline, ) except subprocess.TimeoutExpired: sync_reason = "provider_sync_timeout" @@ -880,6 +906,7 @@ def sync( tree_result = self._run( ["tree", target, "-L", "3", "-o", "json", "-c", "true"], timeout_seconds=remaining, + deadline=deadline, ) except subprocess.TimeoutExpired: sync_reason = "provider_sync_timeout" @@ -903,6 +930,7 @@ def sync( tree_read = self._run( ["read", tree_ref, "-o", "json", "-c", "true"], timeout_seconds=remaining, + deadline=deadline, ) except subprocess.TimeoutExpired: sync_reason = "provider_sync_timeout" @@ -941,6 +969,7 @@ def sync( parent_probe = self._run( ["ls", parent, "-n", "1", "-o", "json", "-c", "true"], timeout_seconds=remaining, + deadline=deadline, ) except subprocess.TimeoutExpired: sync_reason = "provider_sync_timeout" @@ -950,6 +979,7 @@ def sync( mkdir_result = self._run( ["mkdir", parent, "-o", "json", "-c", "true"], timeout_seconds=remaining, + deadline=deadline, ) except subprocess.TimeoutExpired: sync_reason = "provider_sync_timeout" @@ -957,6 +987,7 @@ def sync( if mkdir_result.returncode != 0: sync_reason = "provider_sync_parent_create_failed" break + remaining = max(0.0, deadline - time.monotonic()) reserve = min(15.0, max(1.0, remaining * 0.2)) write_timeout = max(1.0, remaining - reserve) write_args = ( @@ -982,8 +1013,6 @@ def sync( "--mode", "create", "--wait", - "--timeout", - str(max(1, int(write_timeout))), "-o", "json", "-c", @@ -991,7 +1020,12 @@ def sync( ] ) try: - result = self._run(write_args, timeout_seconds=write_timeout) + # Content-write 0.5.2 rejects body.timeout. Omit that optional + # field on all supported servers; the process deadline still + # bounds --wait, with time reserved for uncertain-write readback. + result = self._run( + write_args, timeout_seconds=write_timeout, deadline=deadline + ) except subprocess.TimeoutExpired: result = None reconciliation_performed = True @@ -999,6 +1033,7 @@ def sync( target=target, source_content=source_content, timeout_seconds=reserve, + deadline=deadline, ) if reconciliation == "verified_success": completed_refs.append(target) diff --git a/tests/capabilities/test_reward_memory_provider_retrieval.py b/tests/capabilities/test_reward_memory_provider_retrieval.py index dd53367b91..99afe354db 100644 --- a/tests/capabilities/test_reward_memory_provider_retrieval.py +++ b/tests/capabilities/test_reward_memory_provider_retrieval.py @@ -478,3 +478,161 @@ def runner(command, **kwargs): timeout_seconds=5, observed_at="2026-01-01T00:00:00Z", ) + + +@pytest.mark.parametrize("server_version", ["0.5.0", "0.5.2"]) +@pytest.mark.parametrize( + "root", + [ + "viking://~/memories", + "viking://user/default/memories", + "viking://user/default/peers/goal-one-pilot/memories", + ], +) +@pytest.mark.parametrize("lost_response", [False, True]) +def test_content_sync_compatible_write_reconciles_without_repeating( + tmp_path: Path, + server_version, + root, + lost_response, +): + source = tmp_path / "record.json" + content = '{"lesson":"核对来源后再采用"}\n' + source.write_text(content, encoding="utf-8") + target = root + "/record.json" + stored = {} + writes = [] + + def runner(command, **kwargs): + operation = command[3] if command[1] == "--actor-peer-id" else command[1] + if operation == "--version": + output = "openviking 0.4.23" + elif operation == "version": + output = f"CLI: 0.4.23\nServer: {server_version}" + elif operation == "status": + output = "{}" + elif operation == "ls": + output = '{"items":[]}' + elif operation == "read": + return subprocess.CompletedProcess( + command, + 0 if stored else 1, + json.dumps({"content": stored[target]}) if stored else "not found", + ) + else: + assert operation == "write" + # Both servers accept omission. 0.5.2 forbids body.timeout; + # the CLI version alone does not describe the server protocol. + assert "--timeout" not in command + assert command[command.index("--mode") + 1] == "create" + assert "--wait" in command + assert 0 < kwargs["timeout"] < 30 + writes.append(command) + stored[target] = content + if lost_response: + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + output = '{"status":"ok"}' + return subprocess.CompletedProcess(command, 0, output) + + provider = OpenVikingContextProvider(actor_peer_id="goal-one-pilot", runner=runner) + arguments = dict( + namespace="reward_memory", + resources=[(str(source), target)], + timeout_seconds=30, + observed_at="2026-01-01T00:00:00Z", + ) + preview = provider.sync(**arguments, execute=False) + assert preview.status == "preflight_ready" and not writes + first = provider.sync(**arguments, execute=True) + replay = provider.sync(**arguments, execute=True) + assert first.status == replay.status == "completed" + assert first.write_count == 1 and first.reconciliation_performed + assert replay.write_count == 0 and len(writes) == 1 + stored[target] = "different revision" + conflict = provider.sync(**arguments, execute=True) + assert conflict.reason_code == "provider_sync_revision_conflict" + assert len(writes) == 1 and stored[target] == "different revision" + + +@pytest.mark.parametrize("stall_at", ["--version", "read", "write"]) +def test_sync_deadline_is_shared_by_preflight_write_and_reconciliation( + tmp_path: Path, + monkeypatch, + stall_at, +): + from loopx.capabilities.context_providers import openviking + + clock = [100.0] + monkeypatch.setattr(openviking.time, "monotonic", lambda: clock[0]) + source = tmp_path / "record.json" + source.write_text("{}\n", encoding="utf-8") + calls = [] + stalled = False + + def runner(command, **kwargs): + nonlocal stalled + operation = command[1] + calls.append(operation) + available = 105.0 - clock[0] + assert 0 < kwargs["timeout"] <= available + if operation == stall_at or stalled: + stalled = True + clock[0] += kwargs["timeout"] + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + clock[0] += min(0.4, kwargs["timeout"]) + if operation == "--version": + return subprocess.CompletedProcess(command, 0, "openviking 0.4.23") + if operation == "status": + return subprocess.CompletedProcess(command, 0, "{}") + return subprocess.CompletedProcess(command, 1, "not found") + + result = OpenVikingContextProvider(runner=runner).sync( + namespace="reward_memory", + resources=[(str(source), "viking://user/default/memories/record.json")], + timeout_seconds=5, + observed_at="2026-01-01T00:00:00Z", + execute=True, + ) + assert clock[0] <= 105.0 + assert calls.count("write") <= 1 + if stall_at != "write": + assert "write" not in calls + else: + assert result.retry_disposition == "manual_reconcile" + + +def test_resource_ingestion_keeps_server_timeout(tmp_path: Path): + source = tmp_path / "record.json" + source.write_text("{}\n", encoding="utf-8") + written = False + + def runner(command, **kwargs): + nonlocal written + operation = command[1] + if operation == "--version": + output = "openviking 0.4.23" + elif operation == "status": + output = "{}" + elif operation == "read": + return subprocess.CompletedProcess( + command, + 0 if written else 1, + '{"content":"{}"}' if written else "not found", + ) + elif operation == "add-resource": + assert "--timeout" in command and "--wait" in command + written = True + output = "{}" + else: + assert operation in {"ls", "tree"} + output = '{"items":[]}' + return subprocess.CompletedProcess(command, 0, output) + + result = OpenVikingContextProvider(runner=runner).sync( + namespace="decision_context", + resources=[(str(source), "viking://resources/example/record.json")], + timeout_seconds=30, + observed_at="2026-01-01T00:00:00Z", + execute=True, + ) + assert result.status == "completed" and result.write_count == 1