diff --git a/.github/workflows/crewai-native-retry.yml b/.github/workflows/crewai-native-retry.yml new file mode 100644 index 00000000..01680031 --- /dev/null +++ b/.github/workflows/crewai-native-retry.yml @@ -0,0 +1,51 @@ +name: CrewAI Native Retry Evidence + +on: + pull_request: + paths: + - "examples/crewai-retry-lab/**" + - "sdk/python/**" + - ".github/workflows/crewai-native-retry.yml" + workflow_dispatch: + +permissions: + contents: read + +jobs: + verify: + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest] + sdk: [published, checkout-wheel] + runs-on: ${{ matrix.os }} + timeout-minutes: 15 + env: + PYTHONUTF8: "1" + OTEL_SDK_DISABLED: "true" + CREWAI_TELEMETRY_DISABLED: "true" + CREWAI_TRACING_ENABLED: "false" + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Install pinned public dependencies + run: python -m pip install -r examples/crewai-retry-lab/requirements-native.txt + - name: Install checkout as a built wheel + if: matrix.sdk == 'checkout-wheel' + run: python -m pip install --no-deps --force-reinstall ./sdk/python + - name: Run native retry and adversarial regressions + run: python examples/crewai-retry-lab/native_retry.py --evidence native-evidence.json + - name: Record dependency resolution + if: always() + run: python -m pip freeze --all > resolved-requirements.txt + - uses: actions/upload-artifact@v4 + if: always() + with: + name: crewai-${{ matrix.os }}-${{ matrix.sdk }} + path: | + native-evidence.json + resolved-requirements.txt diff --git a/examples/crewai-retry-lab/.gitignore b/examples/crewai-retry-lab/.gitignore index d7c410cb..e7193ff6 100644 --- a/examples/crewai-retry-lab/.gitignore +++ b/examples/crewai-retry-lab/.gitignore @@ -1 +1,3 @@ .venv-crewai-v4/ +.venv-native/ +native-evidence.json diff --git a/examples/crewai-retry-lab/NATIVE_RETRY.md b/examples/crewai-retry-lab/NATIVE_RETRY.md new file mode 100644 index 00000000..015acb21 --- /dev/null +++ b/examples/crewai-retry-lab/NATIVE_RETRY.md @@ -0,0 +1,145 @@ +# CrewAI #5802: native task retry and bounded recovery evidence + +The existing lab proves structured-tool invocation plus externally controlled +redispatch. `native_retry.py` adds an actual `Agent.execute_task` failure/retry +without patching CrewAI's agent, executor, tool invocation, cache, or retry code. +It also checks process death separately. No SDK production code changes. + +## Reproduce with the published Python SDK + +From this directory, using Python 3.12: + +```powershell +python -m venv .venv-native +.venv-native/Scripts/python -m pip install -r requirements-native.txt +.venv-native/Scripts/python native_retry.py --evidence native-evidence.json +``` + +On Linux/macOS use `.venv-native/bin/python`. No API key, model endpoint, real +payment account, or external service is required. The provider is a local +simulation with a separate durable SQLite database. CrewAI's event loop needs +local loopback sockets; an environment blocking those can hang before execution. + +The direct dependencies are pinned to the existing lab's `crewai==1.15.22` and +published `once-agent-sdk==0.1.1`. The script does **not** insert the repository +SDK path and verifies its import matches the installed distribution. Evidence +records versions, core origin, core hash, harness hash, counts, and pass/fail. +Transitive dependencies are not locked by `requirements-native.txt`; record them +with `python -m pip freeze --all` for an exact environment record. The dedicated +CI workflow retains this alongside JSON evidence for Windows/Linux and +published/checkout-wheel consumers. + +To validate a built checkout in a separate clean environment, install the same +requirements there, then run: + +```powershell +python -m pip install --no-deps --force-reinstall ../../sdk/python +python native_retry.py --evidence checkout-wheel-evidence.json +``` + +Use the new environment's Python. This builds and installs a wheel, not an +editable source import. The two SDKs share a version number; version alone does +not establish equivalence. Compare the recorded core hashes and package origin. + +## Native failure sequence + +1. The application invokes `Agent.execute_task(task)` exactly once, with + `max_retry_limit=1`, caching disabled, and a deterministic `BaseLLM` subclass. +2. CrewAI parses the scripted ReAct tool request and invokes `PaymentTool`. +3. The independent provider commits a payment row, then raises + `AmbiguousProviderOutcome` **before returning a receipt**. The adapter exposes + a non-retryable `ToolFailure` with an explicit unknown outcome. +4. The next scripted model call raises an executor exception. CrewAI's real + task-error handler increments `_times_executed` and calls `execute_task` again. +5. The script requests the same admitted intent. The unprotected control writes + again. Once reconciles the original provider row and returns that receipt. + +Assertions require two actual tool-body invocations, four model calls, exactly +one native task retry, an unknown observation before retry, a real receipt after +retry, and the separate provider row count. The generated final-answer text is +not used as evidence of payment success. The model is scripted; model behaviour +and live model/provider reliability are not measured. There are two deliberate +faults: the lost provider acknowledgement and the subsequent executor error. +This does not claim that returning `ToolFailure` alone restarts a task. + +## Regression matrix + +| Case | Expected and locally observed result | +| --- | --- | +| Native task retry, control | 2 tool calls, 1 task retry, **2 provider effects** | +| Same native retry, Once | 2 tool calls, 1 task retry, **1 effect**, RECONCILED | +| 8 spawned processes released at a barrier | 1 effect; 7 IN_FLIGHT refusals while winner is held after commit | +| Same ID with changed payer, recipient, amount, currency | 12 CONFLICT cases across CLAIMED, UNKNOWN, CONFIRMED; no added effects or record mutation | +| Distinct admitted IDs, identical fields | 2 effects; retry of A returns A's original receipt | +| Lost provider response; lookup unavailable | UNKNOWN; three retries blocked at 1 effect; restored truth reconciles; later replay preserves receipt | +| Confirmed record, attempted recovery/reacquisition after time advance | Four current/stale-version transition refusals; record and receipt retained; 1 effect | +| Expired claim, no provider effect, non-idempotent provider | ABSENT does not authorize redispatch; 0 effects and no false CONFIRMED result | +| `os._exit(77)` after provider commit, before Once confirmation | Fresh worker: control 2 effects; Once 1 effect and RECONCILED | +| `os._exit(79)` after Once confirmation, before tool response | Fresh worker: 1 effect; REPLAYED original receipt | + +Both hard-exit cases require the exact exit code and absence of a worker output +file. Redispatch is **harness-owned**, not CrewAI resurrecting a dead process. +For the unconfirmed crash, the retry worker's injected test clock advances +120001 ms past a 120000 ms lease. No record is deleted or reset. This tests an +expired lease deterministically, not real-time waiting or clock-skew tolerance. +The concurrent case uses ordinary wall time and a held provider response. + +## Release and trust boundaries + +Python `OnceCore`/`SQLiteOperationStore` exposes no operator release, force, +delete, or TTL-eviction API. Therefore this is **not** a test of a hypothetical +`release(force=True)` feature. It tests the actual `mark_unknown` and +`reacquire_after_absent` transitions against a CONFIRMED record, and the normal +core recovery route for unknown/expired claims. These transitions cannot clear +the tested confirmed effect. Low-level storage methods are trusted protocol +primitives, not an authorization API; invoking reacquisition directly on an +unconfirmed record does not establish provider absence or safe redispatch. + +The provider intentionally has no idempotency constraint on operation ID. A +second dispatch creates a second row. Its lookup is authoritative only within +this local fixture and checks the bound fingerprint. A real integration needs +equally authoritative read-only provider truth; missing receipts alone do not +prove absence. Once's safe redispatch rule also requires provider idempotency. +Use native provider idempotency or a database constraint if that fully solves +the operation. + +The caller admits the logical operation ID before execution and retains it on +retry. Payer/account, recipient, amount in integer minor units, and currency are +all fingerprint-bound independently of identity. Identical intentional payments +use distinct IDs. The fixture is one trusted host with a persistent shared +SQLite authority and cooperating processes. It does not cover multi-host +storage, dishonest provider truth, state deletion/rollback, arbitrary code +modifying records, hidden provider retries, live payments, arbitrary argument +canonicalization, or universal exactly-once execution. + +## Local validation and upstream status (2026-11-05) + +Starting Once commit: `8381734a9be368138546121ae708ec9c55be8e89`. +Windows, Python 3.12.10, CrewAI 1.15.22. Published Python SDK 0.1.1 is the latest +version reported by PyPI at the time of the run. Both the published consumer and +a separately installed checkout-wheel consumer passed all 10 tests. Their loaded +core files had SHA-256 +`aa37acded51c75a08edfd9d81350f725b1cfe292336d833bec1506cdc8bb65d6`. +This establishes equivalence for that core file, not every package file. +Linux/remote CI results, when available, are separate evidence. + +The existing hostile-retry and identity labs pass. The checkout's Python SDK +suite passes 44 tests. Repository-required FIRST 10, packed FIRST 10, tool-wrapper, +packed wrapper, local SQLite, generated-proof and site-surface checks were also +run; these are separate regressions, not extra CrewAI proof cases. Initial +restricted runs could not use loopback sockets/fetch the public proof asset; +normal local execution passed. No SDK fix was necessary for these tested paths. + +[CrewAI #5802](https://github.com/crewAIInc/crewAI/issues/5802) is open. +[PR #5822](https://github.com/crewAIInc/crewAI/pull/5822) is open/unmerged at +`e094bad613806c036a734409dee11c666670ed6d`. +[PR #6048](https://github.com/crewAIInc/crewAI/pull/6048) is open/unmerged at +`9a969dd3895fa5c788e1b7ba860a9aa6041d5040`. The latter now describes atomic claims, +scoped task retries, explicit unknown outcomes, and a pluggable backend; earlier +race reports must not be presented as findings against its current head. Neither +PR head was executed by this lab; this is not a comparison or verdict on them. + +[Once FIRST 10 #294](https://github.com/stringsofthemind-oss/once/issues/294) +requests independent cold developer evaluations and remains open with no comments +at inspection. This owner-assisted, AI-assisted fixture work is not an independent +adopter, a real deployment, or retention evidence. No FIRST 10 ledger is changed. diff --git a/examples/crewai-retry-lab/README.md b/examples/crewai-retry-lab/README.md index 4769c9f1..87e85e87 100644 --- a/examples/crewai-retry-lab/README.md +++ b/examples/crewai-retry-lab/README.md @@ -1,5 +1,9 @@ # CrewAI hostile-retry portability lab +For the additional **native task-retry** reproduction of CrewAI #5802 using the +installed Python SDK, see [NATIVE_RETRY.md](NATIVE_RETRY.md). The original lab +below remains unchanged and deliberately uses harness-owned redispatch. + This lab tests the frozen Once V4 framework-neutral recovery core against CrewAI's native tool conversion/invocation boundary. diff --git a/examples/crewai-retry-lab/native_retry.py b/examples/crewai-retry-lab/native_retry.py new file mode 100644 index 00000000..6b5a35e7 --- /dev/null +++ b/examples/crewai-retry-lab/native_retry.py @@ -0,0 +1,449 @@ +"""Offline #5802 regression: real CrewAI task retry, simulated payment authority. + +Run with an installed once-agent-sdk; this module never inserts repository SDK +source into sys.path. The scripted LLM is the only replacement for CrewAI code. +""" +from __future__ import annotations + +import argparse +from contextlib import closing +import hashlib +import importlib.metadata as metadata +import json +import multiprocessing as mp +import os +from pathlib import Path +import sqlite3 +import sys +import tempfile +import time +import unittest + +for stream in (sys.stdout, sys.stderr): + if hasattr(stream, "reconfigure"): + stream.reconfigure(encoding="utf-8") + +# Set before importing CrewAI; no model or telemetry service is needed. +os.environ["OTEL_SDK_DISABLED"] = "true" +os.environ["CREWAI_TELEMETRY_DISABLED"] = "true" +os.environ["CREWAI_TRACING_ENABLED"] = "false" + +from crewai import Agent, Task +from crewai.llms.base_llm import BaseLLM +from crewai.tools import BaseTool, ToolFailure +from pydantic import PrivateAttr + +import once_agent.core as core_module +from once_agent.core import ( + AmbiguousProviderOutcome, ConcurrentModification, ExecutionSource, + OnceCore, OperationConflict, OperationInFlight, OperationRequest, + OperationState, ProviderCapabilities, ProviderObservation, ProviderTruth, + UnresolvedOutcome, UnsafeProviderCapability, fingerprint_action, +) +from once_agent.storage.sqlite import SQLiteOperationStore + +ACTION = dict(payer="sandbox-account-A", recipient="merchant-7", + amount_cents=5000, currency="GBP") +INTENT = "payment:admitted-intent-A" + + +def request(operation_id=INTENT, action=None): + action = dict(ACTION if action is None else action) + return OperationRequest(operation_id, fingerprint_action("payment", action), action) + + +class PaymentAuthority: + """Separate durable effect ledger; intentionally NO execution deduplication. + + Local simulation, not a network provider. Every execute adds a row. Lookup + is authoritative only for this fixture and verifies the bound fingerprint. + """ + capabilities = ProviderCapabilities(lookup_by_operation_id=True, + authoritative_absence=True) + + def __init__(self, path, *, lose_ack=False, crash=False, entered=None, finish=None): + self.path = Path(path) + self.lose_ack, self.crash = lose_ack, crash + self.entered, self.finish = entered, finish + self.truth_available = True + with closing(self.connect()) as db: + db.execute("CREATE TABLE IF NOT EXISTS effects (id INTEGER PRIMARY KEY, " + "operation_id TEXT NOT NULL, fingerprint TEXT NOT NULL, action TEXT NOT NULL)") + + def connect(self): + db = sqlite3.connect(self.path, timeout=30, isolation_level=None) + db.execute("PRAGMA journal_mode=WAL") + db.execute("PRAGMA synchronous=FULL") + return db + + def execute(self, *, operation_id, action_fingerprint, action, fence_token): + with closing(self.connect()) as db: + row = db.execute("INSERT INTO effects(operation_id,fingerprint,action) VALUES(?,?,?)", + (operation_id, action_fingerprint, json.dumps(action, sort_keys=True))) + receipt = dict(effect_id=row.lastrowid, operation_id=operation_id, + action_fingerprint=action_fingerprint, action=action) + # At every injection point the separate provider transaction has committed. + if self.entered is not None: + self.entered.set() + if not self.finish.wait(45): + raise RuntimeError("test coordinator never released provider response") + if self.crash: + os._exit(77) + if self.lose_ack: + self.lose_ack = False + raise AmbiguousProviderOutcome("payment committed; response deliberately lost") + return receipt + + def lookup(self, *, operation_id, action_fingerprint): + if not self.truth_available: + return ProviderObservation(ProviderTruth.UNKNOWN) + with closing(self.connect()) as db: + rows = db.execute("SELECT id,fingerprint,action FROM effects WHERE operation_id=?", + (operation_id,)).fetchall() + if not rows: + return ProviderObservation(ProviderTruth.ABSENT) + if len(rows) != 1 or rows[0][1] != action_fingerprint: + return ProviderObservation(ProviderTruth.UNKNOWN) + return ProviderObservation(ProviderTruth.CONFIRMED, dict( + effect_id=rows[0][0], operation_id=operation_id, + action_fingerprint=rows[0][1], action=json.loads(rows[0][2]))) + + def count(self): + with closing(self.connect()) as db: + return db.execute("SELECT count(*) FROM effects").fetchone()[0] + + +class PaymentTool(BaseTool): + name: str = "send_payment" + description: str = "Execute the already admitted synthetic payment intent." + _provider: object = PrivateAttr() + _core: object = PrivateAttr() + _calls: list = PrivateAttr(default_factory=list) + _crash_after_confirm: bool = PrivateAttr(default=False) + + def __init__(self, provider, core=None, crash_after_confirm=False): + super().__init__() + self._provider, self._core = provider, core + self._crash_after_confirm = crash_after_confirm + + def _run(self, operation_id: str, payer: str, recipient: str, + amount_cents: int, currency: str): + req = request(operation_id, dict(payer=payer, recipient=recipient, + amount_cents=amount_cents, currency=currency)) + self._calls.append(operation_id) + try: + if self._core is not None: + result = self._core.execute(req, provider=self._provider) + if self._crash_after_confirm: + os._exit(79) + return json.dumps(dict(receipt=result.receipt, source=result.source.value)) + return json.dumps(dict(receipt=self._provider.execute( + operation_id=req.operation_id, action_fingerprint=req.action_fingerprint, + action=req.action, fence_token=0), source="UNPROTECTED")) + except (AmbiguousProviderOutcome, UnresolvedOutcome): + return ToolFailure(message="Outcome unknown; no receipt available.", + code="once_outcome_unknown", retryable=False) + except OperationConflict: + return ToolFailure(message="Intent reused with changed payment fields.", + code="once_operation_conflict", retryable=False) + + +class ScriptedLLM(BaseLLM): + """Deterministic ReAct transcript; does not implement a retry loop.""" + def __init__(self): + super().__init__(model="offline-scripted") + self.calls = 0 + self.observations = [] + + def supports_function_calling(self): + return False + + def call(self, messages, **kwargs): + self.calls += 1 + if self.calls in (1, 3): + return ("Thought: Execute the admitted intent.\nAction: send_payment\n" + "Action Input: " + json.dumps(dict(operation_id=INTENT, **ACTION))) + self.observations.append(json.dumps(messages, default=str)) + if self.calls == 2: + # Tool has already reported uncertain outcome. The executor failure + # here propagates to Agent.execute_task's max_retry_limit handling. + raise RuntimeError("injected executor failure after lost payment acknowledgement") + if self.calls == 4: + return "Final Answer: Script finished; inspect the actual receipt and provider ledger." + raise AssertionError(f"unexpected model call {self.calls}") + + +def run_native(root, protected): + provider = PaymentAuthority(root / "provider.db", lose_ack=True) + store = SQLiteOperationStore(root / "once.db") if protected else None + tool = PaymentTool(provider, OnceCore(store) if protected else None) + llm = ScriptedLLM() + agent = Agent(role="Payment test", goal="Execute one admitted payment", + backstory="Offline deterministic regression", llm=llm, + tools=[tool], max_retry_limit=1, max_iter=5, cache=False, + allow_delegation=False, verbose=False) + task = Task(description="Execute admitted payment intent A.", + expected_output="Payment receipt or explicit unknown outcome", agent=agent) + # Exactly one application invocation. CrewAI owns the recursive retry. + agent.execute_task(task) + assert llm.calls == 4, llm.calls + assert agent._times_executed == 1, agent._times_executed + assert tool._calls == [INTENT, INTENT], tool._calls + assert provider.count() == (1 if protected else 2), provider.count() + first_observation = str(llm.observations[0]) + second_observation = str(llm.observations[1]) + assert "unknown" in first_observation.lower(), first_observation + assert "receipt" in second_observation, second_observation + if protected: + assert store.get(INTENT).state == OperationState.CONFIRMED + assert "RECONCILED" in second_observation, second_observation + return dict(effects=provider.count(), tool_calls=len(tool._calls), + native_task_retries=agent._times_executed, llm_calls=llm.calls) + + +def worker(root, protected, crash, result_path, ready=None, start=None, + entered=None, finish=None, clock_offset_ms=0, crash_after_confirm=False): + """Spawn-safe process entry; actual native structured-tool invocation.""" + root = Path(root) + provider = PaymentAuthority(root / "provider.db", crash=crash, + entered=entered, finish=finish) + store = SQLiteOperationStore(root / "once.db", now_ms=lambda: + time.time_ns() // 1_000_000 + clock_offset_ms) if protected else None + core = OnceCore(store, lease_ms=120_000) if protected else None + tool = PaymentTool(provider, core, crash_after_confirm) + if ready is not None: + ready.set() + if not start.wait(45): + raise RuntimeError("start barrier timed out") + try: + result = tool.to_structured_tool().invoke(dict(operation_id=INTENT, **ACTION)) + outcome = dict(status="returned", result=result, pid=os.getpid()) + except OperationInFlight: + outcome = dict(status="in_flight", pid=os.getpid()) + Path(result_path).write_text(json.dumps(outcome), encoding="utf-8") + + +class RetryTests(unittest.TestCase): + results = {} + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory(prefix="once-crewai-5802-") + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.provider = PaymentAuthority(self.root / "provider.db") + self.clock = [1000] + self.store = SQLiteOperationStore(self.root / "once.db", now_ms=lambda: self.clock[0]) + self.core = OnceCore(self.store, lease_ms=100) + + def test_native_task_retry_control(self): + self.results["native_control"] = run_native(self.root, False) + + def test_native_task_retry_once(self): + self.results["native_once"] = run_native(self.root, True) + + def test_lost_response_unknown_then_authoritative_recovery(self): + self.provider.lose_ack = True + with self.assertRaises(UnresolvedOutcome): + self.core.execute(request(), provider=self.provider) + self.assertEqual(self.store.get(INTENT).state, OperationState.UNKNOWN) + self.provider.truth_available = False + for _ in range(3): + with self.assertRaises(UnresolvedOutcome): + self.core.execute(request(), provider=self.provider) + self.assertEqual(self.provider.count(), 1) + self.provider.truth_available = True + recovered = self.core.execute(request(), provider=self.provider) + replay = self.core.execute(request(), provider=self.provider) + self.assertEqual(recovered.source, ExecutionSource.RECONCILED) + self.assertEqual(replay.source, ExecutionSource.REPLAYED) + self.assertEqual(recovered.receipt, replay.receipt) + self.assertEqual(self.provider.count(), 1) + self.results["lost_response"] = dict(blocked_retries=3, effects=1, + recovery="RECONCILED", replay="REPLAYED") + + def test_mutated_fields_conflict_in_all_record_states(self): + mutations = dict(payer="other-account", recipient="other-merchant", + amount_cents=5001, currency="USD") + for state in (OperationState.CLAIMED, OperationState.UNKNOWN, OperationState.CONFIRMED): + op = "mutation:" + state.value + req = request(op) + if state == OperationState.CLAIMED: + self.store.acquire_or_observe(operation_id=op, action_fingerprint=req.action_fingerprint, + owner_token="held", lease_ms=100) + elif state == OperationState.UNKNOWN: + self.provider.lose_ack = True + with self.assertRaises(UnresolvedOutcome): + self.core.execute(req, provider=self.provider) + else: + self.core.execute(req, provider=self.provider) + before = self.provider.count() + record = self.store.get(op) + for field, value in mutations.items(): + with self.subTest(state=state, field=field): + with self.assertRaises(OperationConflict): + self.core.execute(request(op, {**ACTION, field: value}), provider=self.provider) + self.assertEqual(self.provider.count(), before) + self.assertEqual(self.store.get(op), record) + self.results["mutations"] = dict(conflicts=12, additional_effects=0) + + def test_identical_payload_distinct_intents(self): + a = self.core.execute(request(), provider=self.provider) + b = self.core.execute(request("payment:admitted-intent-B"), provider=self.provider) + self.assertNotEqual(a.receipt["effect_id"], b.receipt["effect_id"]) + self.assertEqual(self.core.execute(request(), provider=self.provider).receipt, a.receipt) + self.assertEqual(self.provider.count(), 2) + self.results["distinct_intents"] = dict(effects=2) + + def test_confirmed_record_cannot_be_released_by_recovery(self): + first = self.core.execute(request(), provider=self.provider) + record = self.store.get(INTENT) + # No release/delete/reset API exists in this Python core. Test the real + # recovery transitions, including current and stale optimistic versions. + self.clock[0] += 1_000_000 + for version in (record.version, record.version - 1): + with self.assertRaises(ConcurrentModification): + self.store.reacquire_after_absent(operation_id=INTENT, + action_fingerprint=record.action_fingerprint, expected_version=version, + owner_token="new-holder", lease_ms=100) + with self.assertRaises(ConcurrentModification): + self.store.mark_unknown(operation_id=INTENT, + action_fingerprint=record.action_fingerprint, expected_version=version) + self.assertEqual(self.store.get(INTENT), record) + replay = self.core.execute(request(), provider=self.provider) + self.assertEqual(replay.receipt, first.receipt) + self.assertEqual(self.provider.count(), 1) + self.results["confirmed_release"] = dict(refused_transitions=4, effects=1) + + def test_expired_claim_without_effect_does_not_authorize_redispatch(self): + req = request() + self.store.acquire_or_observe(operation_id=INTENT, action_fingerprint=req.action_fingerprint, + owner_token="dead-before-dispatch", lease_ms=100) + self.clock[0] += 101 + with self.assertRaises(UnsafeProviderCapability): + self.core.execute(req, provider=self.provider) + self.assertEqual(self.provider.count(), 0) + self.assertNotEqual(self.store.get(INTENT).state, OperationState.CONFIRMED) + self.results["absent_nonidempotent"] = dict(effects=0, refused=True) + + def spawn(self, **kwargs): + process = mp.get_context("spawn").Process(target=worker, kwargs=kwargs) + process.start() + self.addCleanup(self.stop, process) + return process + + @staticmethod + def stop(process): + if process.is_alive(): + process.terminate() + process.join(10) + process.close() + + def test_concurrent_claims_count_external_dispatches(self): + ctx = mp.get_context("spawn") + entered, finish, start = ctx.Event(), ctx.Event(), ctx.Event() + ready = [ctx.Event() for _ in range(8)] + outputs = [self.root / f"worker-{i}.json" for i in range(8)] + workers = [self.spawn(root=str(self.root), protected=True, crash=False, + result_path=str(outputs[i]), ready=ready[i], start=start, + entered=entered, finish=finish) for i in range(8)] + try: + self.assertTrue(all(event.wait(60) for event in ready), "workers not ready") + start.set() + self.assertTrue(entered.wait(30), "no provider dispatch") + # Winner is held after committing. Every loser must finish while + # the winner is still active, proving an actual overlapping claim. + deadline = time.monotonic() + 45 + while sum(path.exists() for path in outputs) < 7 and time.monotonic() < deadline: + time.sleep(0.02) + self.assertEqual(sum(path.exists() for path in outputs), 7) + self.assertEqual(self.provider.count(), 1) + finally: + finish.set() + start.set() + for process in workers: + process.join(30) + self.assertEqual(process.exitcode, 0) + rows = [json.loads(path.read_text()) for path in outputs] + self.assertEqual(sum(row["status"] == "in_flight" for row in rows), 7) + self.assertEqual(len({row["pid"] for row in rows}), 8) + self.assertEqual(self.provider.count(), 1) + self.results["concurrency"] = dict(processes=8, in_flight=7, effects=1) + + def test_process_death_before_acknowledgement(self): + for protected in (False, True): + with self.subTest(protected=protected): + root = self.root / str(protected) + root.mkdir() + provider = PaymentAuthority(root / "provider.db") + output = root / "crashed.json" + child = self.spawn(root=str(root), protected=protected, crash=True, + result_path=str(output)) + child.join(60) + self.assertEqual(child.exitcode, 77) + self.assertFalse(output.exists(), "crashed worker returned an acknowledgement") + self.assertEqual(provider.count(), 1) + if protected: + # Advance only the test clock to expire the dead lease; + # do not delete records or authorize recovery via SQL. + retry_output = root / "reconciled.json" + child2 = self.spawn(root=str(root), protected=True, crash=False, + result_path=str(retry_output), clock_offset_ms=120_001) + child2.join(60) + self.assertEqual(child2.exitcode, 0) + result = json.loads(json.loads(retry_output.read_text())["result"]) + self.assertEqual(result["source"], "RECONCILED") + else: + child2 = self.spawn(root=str(root), protected=False, crash=False, + result_path=str(root / "redispatched.json")) + child2.join(60) + self.assertEqual(child2.exitcode, 0) + self.assertEqual(provider.count(), 1 if protected else 2) + self.results["process_death"] = dict(control_effects=2, once_effects=1, + exit_code=77, redispatch="harness-owned") + + def test_confirmed_result_lost_before_framework_receives_it(self): + output = self.root / "lost-result.json" + child = self.spawn(root=str(self.root), protected=True, crash=False, + result_path=str(output), crash_after_confirm=True) + child.join(60) + self.assertEqual(child.exitcode, 79) + self.assertFalse(output.exists()) + record = self.store.get(INTENT) + self.assertEqual(record.state, OperationState.CONFIRMED) + replay_output = self.root / "replay.json" + retry = self.spawn(root=str(self.root), protected=True, crash=False, + result_path=str(replay_output)) + retry.join(60) + self.assertEqual(retry.exitcode, 0) + result = json.loads(json.loads(replay_output.read_text())["result"]) + self.assertEqual(result["source"], "REPLAYED") + self.assertEqual(result["receipt"], record.receipt) + self.assertEqual(self.provider.count(), 1) + self.results["confirmed_result_lost"] = dict(exit_code=79, effects=1, source="REPLAYED") + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--evidence", type=Path, default=Path("native-evidence.json")) + args = parser.parse_args() + origin = Path(core_module.__file__).resolve() + expected = Path(metadata.distribution("once-agent-sdk").locate_file("once_agent/core.py")).resolve() + if origin != expected: + raise RuntimeError(f"Expected installed SDK at {expected}, imported {origin}") + suite = unittest.defaultTestLoader.loadTestsFromTestCase(RetryTests) + result = unittest.TextTestRunner(verbosity=2).run(suite) + evidence = dict(passed=result.wasSuccessful(), tests=result.testsRun, + failures=len(result.failures), errors=len(result.errors), + python=sys.version, crewai=metadata.version("crewai"), + once=metadata.version("once-agent-sdk"), once_core_origin=str(origin), + once_core_sha256=hashlib.sha256(origin.read_bytes()).hexdigest(), + harness_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), + cases=RetryTests.results) + args.evidence.parent.mkdir(parents=True, exist_ok=True) + args.evidence.write_text(json.dumps(evidence, indent=2) + "\n", encoding="utf-8") + print(json.dumps(evidence, indent=2)) + return 0 if result.wasSuccessful() else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/crewai-retry-lab/requirements-native.txt b/examples/crewai-retry-lab/requirements-native.txt new file mode 100644 index 00000000..a3a3b0a8 --- /dev/null +++ b/examples/crewai-retry-lab/requirements-native.txt @@ -0,0 +1,3 @@ +# Existing lab's CrewAI pin; published Python Once (not the TypeScript SDK). +crewai==1.15.22 +once-agent-sdk==0.1.1