diff --git a/features/a2a_facade_coverage.feature b/features/a2a_facade_coverage.feature index eaee174d2..265261ebe 100644 --- a/features/a2a_facade_coverage.feature +++ b/features/a2a_facade_coverage.feature @@ -223,7 +223,7 @@ Feature: A2A local facade coverage — uncovered handler and edge-case paths When I dispatch facade-cov operation "_cleveragents/sync/push" with params {} Then the facade-cov response status should be "ok" And the facade-cov response data key "stub" should be true - And the facade-cov response data key "status" should equal "not_implemented" + And the facade-cov response data key "status" should equal "no_sync_service" Scenario: Sync status returns not_implemented stub Given a facade-cov facade with no services diff --git a/features/entity_sync.feature b/features/entity_sync.feature new file mode 100644 index 000000000..283b833ea --- /dev/null +++ b/features/entity_sync.feature @@ -0,0 +1,418 @@ +@mock_only +Feature: Entity sync via _cleveragents/sync/* + As a CleverAgents user with multiple devices + I want to sync entities between my client and the server + So that I can access the same actors, skills, actions, plans, and sessions everywhere + + # ------------------------------------------------------------------ + # Sync models + # ------------------------------------------------------------------ + + Scenario: Vector clock increment produces new clock + Given a vector clock with entries {"client1": 1, "client2": 3} + When I increment the clock for node "client1" + Then the clock entry for "client1" should be 2 + And the clock entry for "client2" should be 3 + + Scenario: Vector clock merge takes element-wise max + Given a vector clock with entries {"a": 2, "b": 1} + And another vector clock with entries {"a": 1, "b": 3, "c": 1} + When I merge the two clocks + Then the merged clock should have entries {"a": 2, "b": 3, "c": 1} + + Scenario: Vector clock happens-before detection + Given a vector clock with entries {"a": 1, "b": 1} + And another vector clock with entries {"a": 2, "b": 2} + Then the first clock should happen before the second + + Scenario: Vector clock concurrency detection + Given a vector clock with entries {"a": 2, "b": 1} + And another vector clock with entries {"a": 1, "b": 2} + Then the clocks should be concurrent + + Scenario: Vector clock increment rejects empty node_id + Given an empty vector clock + When I try to increment the clock with an empty node_id + Then a sync ValueError should be raised + + Scenario: Vector clock merge rejects non-VectorClock + Given an empty vector clock + When I try to merge with a non-VectorClock value + Then a sync TypeError should be raised + + Scenario: SyncEntitySnapshot rejects empty entity_id + When I try to create a snapshot with an empty entity_id + Then a sync validation error should be raised + + Scenario: SyncPullRequest rejects local namespace + When I try to create a pull request with namespace "local" + Then a sync validation error should be raised + + Scenario: SyncPushRequest rejects local namespace + When I try to create a push request with namespace "local" + Then a sync validation error should be raised + + Scenario: SyncStatusRequest rejects local namespace + When I try to create a status request with namespace "local" + Then a sync validation error should be raised + + Scenario: SyncConflict requires non-empty conflict_id + When I try to create a conflict with an empty conflict_id + Then a sync validation error should be raised + + Scenario: SyncQueueEntry requires non-empty queue_id + When I try to create a queue entry with an empty queue_id + Then a sync validation error should be raised + + Scenario: SyncState requires non-empty namespace + When I try to create a sync state with an empty namespace + Then a sync validation error should be raised + + # ------------------------------------------------------------------ + # SyncService — pull + # ------------------------------------------------------------------ + + Scenario: Pull entities from server to local cache + Given a sync service with node_id "test-client" + And server entities in namespace "team" with 3 actors + When I pull all entities from namespace "team" + Then the pull response should contain 3 entities + And the pull response should have 0 conflicts + + Scenario: Pull detects concurrent modifications as conflicts + Given a sync service with node_id "test-client" + And a local actor "my-actor" in namespace "team" with clock {"client": 2, "server": 1} + And a server actor "my-actor" in namespace "team" with clock {"client": 1, "server": 2} + When I pull all entities from namespace "team" + Then the pull response should have 1 conflicts + + Scenario: Pull with incremental sync uses since parameter + Given a sync service with node_id "test-client" + And server entities in namespace "team" with timestamps + When I pull entities from namespace "team" since "2025-01-01T00:00:00+00:00" + Then only entities updated after the since timestamp are pulled + + Scenario: Pull with force overwrites local entities + Given a sync service with node_id "test-client" + And a local actor "my-actor" in namespace "team" with clock {"client": 2} + And a server actor "my-actor" in namespace "team" with clock {"server": 1} + When I force pull all entities from namespace "team" + Then the pull response should contain 1 entities + And the pull response should have 0 conflicts + + Scenario: Pull rejects local namespace + Given a sync service with node_id "test-client" + When I try to pull from namespace "local" + Then a sync ValidationError should be raised with message containing "local" + + Scenario: Pull type-checks the request + Given a sync service with node_id "test-client" + When I try to pull with a non-SyncPullRequest argument + Then a sync TypeError should be raised + + # ------------------------------------------------------------------ + # SyncService — push + # ------------------------------------------------------------------ + + Scenario: Push entities from local to server + Given a sync service with node_id "test-client" + And 2 local actors in namespace "shared" + When I push all local entities to namespace "shared" + Then the push response should have 2 accepted + And the push response should have 0 rejected + + Scenario: Push detects and resolves conflicts with last-writer-wins + Given a sync service with node_id "test-client" + And a local actor "conflict-actor" in namespace "team" clocked {"client": 2, "server": 1} at "2025-06-01T12:00:00+00:00" + And a server actor "conflict-actor" in namespace "team" clocked {"client": 1, "server": 2} at "2025-05-01T12:00:00+00:00" + When I push the local actor to namespace "team" with last-writer-wins resolution + Then the push response should have 1 accepted + And the push response should have 1 conflicts + + Scenario: Push with force bypasses conflict detection + Given a sync service with node_id "test-client" + And a server actor "existing" in namespace "team" with clock {"server": 5} + When I force push a local actor "existing" to namespace "team" + Then the push response should have 1 accepted + And the push response should have 0 conflicts + + Scenario: Push rejects local namespace + Given a sync service with node_id "test-client" + When I try to push to namespace "local" + Then a sync ValidationError should be raised with message containing "local" + + Scenario: Push type-checks the request + Given a sync service with node_id "test-client" + When I try to push with a non-SyncPushRequest argument + Then a sync TypeError should be raised + + # ------------------------------------------------------------------ + # SyncService — status + # ------------------------------------------------------------------ + + Scenario: Status detects local-ahead drift + Given a sync service with node_id "test-client" + And a local actor "new-actor" in namespace "team" not on server + When I request sync status for namespace "team" + Then the status should show drift detected + And the status should have 1 local-ahead entities + + Scenario: Status detects server-ahead drift + Given a sync service with node_id "test-client" + And a server actor "server-only" in namespace "team" not on local + When I request sync status for namespace "team" + Then the status should show drift detected + And the status should have 1 server-ahead entities + + Scenario: Status detects no drift when in sync + Given a sync service with node_id "test-client" + And a synced actor "synced-actor" in namespace "team" on both sides + When I request sync status for namespace "team" + Then the status should show no drift + + Scenario: Status rejects local namespace + Given a sync service with node_id "test-client" + When I try requesting sync status for namespace "local" + Then a sync ValidationError should be raised with message containing "local" + + Scenario: Status type-checks the request + Given a sync service with node_id "test-client" + When I try to check status with a non-SyncStatusRequest argument + Then a sync TypeError should be raised + + Scenario: Status filters by entity type + Given a sync service with node_id "test-client" + And a local actor "my-actor" in namespace "team" not on server + And a local skill "my-skill" in namespace "team" not on server + When I request filtered sync status for namespace "team" entity type "actor" + Then the status should have 1 local-ahead entities + + # ------------------------------------------------------------------ + # Conflict resolution + # ------------------------------------------------------------------ + + Scenario: Resolve conflict manually with local winner + Given a sync service with node_id "test-client" + And an unresolved conflict between local and server versions + When I resolve the conflict manually choosing "local" + Then the conflict should be marked as resolved + And the conflict winner should be "local" + + Scenario: Resolve conflict manually with server winner + Given a sync service with node_id "test-client" + And an unresolved conflict between local and server versions + When I resolve the conflict manually choosing "server" + Then the conflict should be marked as resolved + And the conflict winner should be "server" + + Scenario: Manual resolution without winner raises BusinessRuleViolation + Given a sync service with node_id "test-client" + And an unresolved conflict between local and server versions + When I try to resolve manually without specifying a winner + Then a sync BusinessRuleViolation should be raised + + Scenario: Resolve non-existent conflict raises ValueError + Given a sync service with node_id "test-client" + When I try to resolve a conflict with id "nonexistent" + Then a sync ValueError should be raised + + Scenario: Resolve conflict with empty conflict_id raises ValueError + Given a sync service with node_id "test-client" + When I try to resolve a conflict with an empty id + Then a sync ValueError should be raised + + # ------------------------------------------------------------------ + # Offline queue + # ------------------------------------------------------------------ + + Scenario: Enqueue offline push operation + Given a sync service with node_id "test-client" + And a local actor snapshot for offline queuing + When I enqueue an offline push to namespace "team" + Then the offline queue should have 1 entry + And the queued entry should have status "queued" + + Scenario: Process offline queue retries push operations + Given a sync service with node_id "test-client" + And a queued offline push for actor "offline-actor" in namespace "team" + When I process the offline queue + Then the processed entries should include the offline actor + And the offline queue should be empty + + Scenario: Offline queue respects max retries + Given a sync service with node_id "test-client" + And a queued offline push at max retries + When I process the offline queue + Then the failed entry should have status "failed" + + Scenario: Enqueue rejects local namespace + Given a sync service with node_id "test-client" + And a local actor snapshot for offline queuing + When I try to enqueue to namespace "local" + Then a sync ValueError should be raised + + Scenario: Enqueue rejects empty namespace + Given a sync service with node_id "test-client" + And a local actor snapshot for offline queuing + When I try to enqueue to an empty namespace + Then a sync ValueError should be raised + + # ------------------------------------------------------------------ + # Facade integration + # ------------------------------------------------------------------ + + Scenario: Facade sync/pull without service returns stub + Given a facade without sync service + When I dispatch "_cleveragents/sync/pull" with sync params {"namespace": "team"} + Then the facade response should be ok + And the facade response data should have "stub" set to true + + Scenario: Facade sync/push without service returns stub + Given a facade without sync service + When I dispatch "_cleveragents/sync/push" with sync params {"namespace": "team", "entities": []} + Then the facade response should be ok + And the facade response data should have "stub" set to true + + Scenario: Facade sync/status without service returns stub + Given a facade without sync service + When I dispatch "_cleveragents/sync/status" with sync params {"namespace": "team"} + Then the facade response should be ok + And the facade response data should have "stub" set to true + + Scenario: Facade sync/pull with service returns entities + Given a facade with sync service containing server entities + When I dispatch "_cleveragents/sync/pull" with sync params {"namespace": "team"} + Then the facade response should be ok + And the facade response data should contain key "entities" + And the facade response data should contain key "total_pulled" + + Scenario: Facade sync/push with service accepts entities + Given a facade with sync service + When I dispatch "_cleveragents/sync/push" with sync push params + Then the facade response should be ok + And the facade response data should contain key "accepted" + + Scenario: Facade sync/status with service returns drift info + Given a facade with sync service containing local-only entities + When I dispatch "_cleveragents/sync/status" with sync params {"namespace": "team"} + Then the facade response should be ok + And the facade response data should contain key "drift_detected" + + # ------------------------------------------------------------------ + # SyncService constructor validation + # ------------------------------------------------------------------ + + Scenario: SyncService rejects non-string node_id + When I try to create a SyncService with a non-string node_id + Then a sync TypeError should be raised + + Scenario: SyncService with None node_id generates ULID + When I create a SyncService with None node_id + Then the service should have a non-empty node_id + + Scenario: SyncService register_local_entity rejects non-snapshot + Given a sync service with node_id "test-client" + When I try to register a non-snapshot as local entity + Then a sync TypeError should be raised + + Scenario: SyncService register_server_entity rejects non-snapshot + Given a sync service with node_id "test-client" + When I try to register a non-snapshot as server entity + Then a sync TypeError should be raised + + Scenario: Enqueue offline rejects invalid direction type + Given a sync service with node_id "test-client" + And a local actor snapshot for offline queuing + When I try to enqueue with an invalid direction type + Then a sync TypeError should be raised + + Scenario: Enqueue offline rejects invalid entity type + Given a sync service with node_id "test-client" + When I try to enqueue with an invalid entity type + Then a sync TypeError should be raised + + Scenario: Resolve conflict rejects invalid resolution type + Given a sync service with node_id "test-client" + And an unresolved conflict between local and server versions + When I try to resolve with an invalid resolution type + Then a sync TypeError should be raised + + # ------------------------------------------------------------------ + # Edge cases + # ------------------------------------------------------------------ + + Scenario: Pull from empty server returns empty response + Given a sync service with node_id "test-client" + When I pull all entities from namespace "empty-ns" + Then the pull response should contain 0 entities + + Scenario: Push empty entity list returns empty response + Given a sync service with node_id "test-client" + When I push an empty entity list to namespace "team" + Then the push response should have 0 accepted + + Scenario: Vector clock equal clocks are not concurrent + Given a vector clock with entries {"a": 1} + And another vector clock with entries {"a": 1} + Then the clocks should not be concurrent + + Scenario: Status with entity type filter on skills + Given a sync service with node_id "test-client" + And a local skill "my-skill" in namespace "team" not on server + When I request filtered sync status for namespace "team" entity type "skill" + Then the status should have 1 local-ahead entities + + Scenario: Push with server_wins resolution resolves to server entity + Given a sync service with node_id "test-client" + And a local actor "conflict-actor" in namespace "team" with clock {"client": 2, "server": 1} + And a server actor "conflict-actor" in namespace "team" with clock {"client": 1, "server": 2} + When I push the local actor to namespace "team" with server-wins resolution + Then the push response should have 1 accepted + And the push conflict winner should be "server" + + Scenario: Push with manual resolution rejects conflicting entities + Given a sync service with node_id "test-client" + And a local actor "conflict-actor" in namespace "team" with clock {"client": 2, "server": 1} + And a server actor "conflict-actor" in namespace "team" with clock {"client": 1, "server": 2} + When I push the local actor to namespace "team" with manual resolution + Then the push response should have 0 accepted + And the push response should have 1 rejected + + Scenario: Resolve conflict with client_wins + Given a sync service with node_id "test-client" + And an unresolved conflict between local and server versions + When I resolve the conflict with client_wins strategy + Then the conflict should be marked as resolved + And the conflict winner should be "local" + + Scenario: Resolve conflict with server_wins + Given a sync service with node_id "test-client" + And an unresolved conflict between local and server versions + When I resolve the conflict with server_wins strategy + Then the conflict should be marked as resolved + And the conflict winner should be "server" + + Scenario: Resolve conflict with last_writer_wins picks newer timestamp + Given a sync service with node_id "test-client" + And an unresolved conflict where local is newer + When I resolve the conflict with last_writer_wins strategy + Then the conflict should be marked as resolved + And the conflict winner should be "local" + + Scenario: Pull updates sync state + Given a sync service with node_id "test-client" + And server entities in namespace "team" with 1 actors + When I pull all entities from namespace "team" + Then the pull response should have a sync state with last_pull_at set + + Scenario: Push updates sync state + Given a sync service with node_id "test-client" + And 1 local actors in namespace "shared" + When I push all local entities to namespace "shared" + Then the push response should have a sync state with last_push_at set + + Scenario: Offline queue enqueue pull operation + Given a sync service with node_id "test-client" + And a local actor snapshot for offline queuing + When I enqueue an offline pull from namespace "team" + Then the offline queue should have 1 entry diff --git a/features/steps/entity_sync_steps.py b/features/steps/entity_sync_steps.py new file mode 100644 index 000000000..724f824dd --- /dev/null +++ b/features/steps/entity_sync_steps.py @@ -0,0 +1,1176 @@ +"""Step definitions for entity_sync.feature. + +Tests the ``_cleveragents/sync/*`` entity synchronization service, +including vector clock operations, pull/push/status workflows, +conflict resolution, offline queue, and facade integration. +""" + +from __future__ import annotations + +import json +from typing import Any + +from behave import given, then, when +from pydantic import ValidationError as PydanticValidationError + +from cleveragents.a2a.facade import A2aLocalFacade +from cleveragents.a2a.models import A2aRequest +from cleveragents.a2a.sync_models import ( + ConflictResolution, + SyncConflict, + SyncDirection, + SyncEntitySnapshot, + SyncEntityType, + SyncOperationStatus, + SyncPullRequest, + SyncPushRequest, + SyncQueueEntry, + SyncState, + SyncStatusRequest, + VectorClock, +) +from cleveragents.application.services.sync_service import SyncService +from cleveragents.core.exceptions import BusinessRuleViolation, ValidationError + +# ------------------------------------------------------------------ +# Vector clock steps +# ------------------------------------------------------------------ + + +@given("a vector clock with entries {entries_json}") +def step_create_vector_clock(context: Any, entries_json: str) -> None: + entries = json.loads(entries_json) + context.clock1 = VectorClock(entries=entries) + + +@given("another vector clock with entries {entries_json}") +def step_create_second_vector_clock(context: Any, entries_json: str) -> None: + entries = json.loads(entries_json) + context.clock2 = VectorClock(entries=entries) + + +@given("an empty vector clock") +def step_create_empty_clock(context: Any) -> None: + context.clock1 = VectorClock() + + +@when('I increment the clock for node "{node_id}"') +def step_increment_clock(context: Any, node_id: str) -> None: + context.result_clock = context.clock1.increment(node_id) + + +@when("I merge the two clocks") +def step_merge_clocks(context: Any) -> None: + context.result_clock = context.clock1.merge(context.clock2) + + +@then('the clock entry for "{node_id}" should be {value:d}') +def step_check_clock_entry(context: Any, node_id: str, value: int) -> None: + assert context.result_clock.entries.get(node_id) == value, ( + f"Expected {value} for {node_id}, " + f"got {context.result_clock.entries.get(node_id)}" + ) + + +@then("the merged clock should have entries {entries_json}") +def step_check_merged_entries(context: Any, entries_json: str) -> None: + expected = json.loads(entries_json) + assert context.result_clock.entries == expected, ( + f"Expected {expected}, got {context.result_clock.entries}" + ) + + +@then("the first clock should happen before the second") +def step_clock_happens_before(context: Any) -> None: + assert context.clock1.happens_before(context.clock2), ( + "Expected clock1 to happen before clock2" + ) + + +@then("the clocks should be concurrent") +def step_clocks_concurrent(context: Any) -> None: + assert context.clock1.is_concurrent(context.clock2), ( + "Expected clocks to be concurrent" + ) + + +@then("the clocks should not be concurrent") +def step_clocks_not_concurrent(context: Any) -> None: + assert not context.clock1.is_concurrent(context.clock2), ( + "Expected clocks to NOT be concurrent" + ) + + +@when("I try to increment the clock with an empty node_id") +def step_increment_empty_node(context: Any) -> None: + context.caught_exception = None + try: + context.clock1.increment("") + except ValueError as exc: + context.caught_exception = exc + + +@when("I try to merge with a non-VectorClock value") +def step_merge_non_clock(context: Any) -> None: + context.caught_exception = None + try: + context.clock1.merge("not a clock") # type: ignore[arg-type] + except TypeError as exc: + context.caught_exception = exc + + +# ------------------------------------------------------------------ +# Model validation steps +# ------------------------------------------------------------------ + + +@when("I try to create a snapshot with an empty entity_id") +def step_create_bad_snapshot(context: Any) -> None: + context.caught_exception = None + try: + SyncEntitySnapshot( + entity_id="", + entity_type=SyncEntityType.ACTOR, + namespace="test", + name="test", + ) + except (PydanticValidationError, ValueError) as exc: + context.caught_exception = exc + + +@when('I try to create a pull request with namespace "local"') +def step_create_pull_local(context: Any) -> None: + context.caught_exception = None + try: + SyncPullRequest(namespace="local") + except (PydanticValidationError, ValueError) as exc: + context.caught_exception = exc + + +@when('I try to create a push request with namespace "local"') +def step_create_push_local(context: Any) -> None: + context.caught_exception = None + try: + SyncPushRequest(namespace="local") + except (PydanticValidationError, ValueError) as exc: + context.caught_exception = exc + + +@when('I try to create a status request with namespace "local"') +def step_create_status_local(context: Any) -> None: + context.caught_exception = None + try: + SyncStatusRequest(namespace="local") + except (PydanticValidationError, ValueError) as exc: + context.caught_exception = exc + + +@when("I try to create a conflict with an empty conflict_id") +def step_create_bad_conflict(context: Any) -> None: + context.caught_exception = None + try: + SyncConflict( + conflict_id="", + entity_id="e1", + entity_type=SyncEntityType.ACTOR, + namespace="test", + local_snapshot=SyncEntitySnapshot( + entity_id="e1", + entity_type=SyncEntityType.ACTOR, + namespace="test", + name="test", + ), + server_snapshot=SyncEntitySnapshot( + entity_id="e1", + entity_type=SyncEntityType.ACTOR, + namespace="test", + name="test", + ), + local_clock=VectorClock(), + server_clock=VectorClock(), + ) + except (PydanticValidationError, ValueError) as exc: + context.caught_exception = exc + + +@when("I try to create a queue entry with an empty queue_id") +def step_create_bad_queue_entry(context: Any) -> None: + context.caught_exception = None + try: + SyncQueueEntry( + queue_id="", + direction=SyncDirection.PUSH, + namespace="test", + entity_type=SyncEntityType.ACTOR, + entity_id="e1", + snapshot=SyncEntitySnapshot( + entity_id="e1", + entity_type=SyncEntityType.ACTOR, + namespace="test", + name="test", + ), + ) + except (PydanticValidationError, ValueError) as exc: + context.caught_exception = exc + + +@when("I try to create a sync state with an empty namespace") +def step_create_bad_sync_state(context: Any) -> None: + context.caught_exception = None + try: + SyncState(namespace="") + except (PydanticValidationError, ValueError) as exc: + context.caught_exception = exc + + +# ------------------------------------------------------------------ +# SyncService setup steps +# ------------------------------------------------------------------ + + +@given('a sync service with node_id "{node_id}"') +def step_create_sync_service(context: Any, node_id: str) -> None: + context.sync_service = SyncService(node_id=node_id) + + +def _make_actor_snapshot( + entity_id: str, + namespace: str, + clock_entries: dict[str, int] | None = None, + updated_at: str = "", +) -> SyncEntitySnapshot: + """Helper to create an actor snapshot.""" + return SyncEntitySnapshot( + entity_id=entity_id, + entity_type=SyncEntityType.ACTOR, + namespace=namespace, + name=f"actor-{entity_id}", + data={"role": "executor"}, + vector_clock=VectorClock(entries=clock_entries or {}), + updated_at=updated_at, + ) + + +def _make_skill_snapshot( + entity_id: str, + namespace: str, + clock_entries: dict[str, int] | None = None, +) -> SyncEntitySnapshot: + """Helper to create a skill snapshot.""" + return SyncEntitySnapshot( + entity_id=entity_id, + entity_type=SyncEntityType.SKILL, + namespace=namespace, + name=f"skill-{entity_id}", + data={"category": "search"}, + vector_clock=VectorClock(entries=clock_entries or {}), + ) + + +@given('server entities in namespace "{namespace}" with {count:d} actors') +def step_setup_server_entities(context: Any, namespace: str, count: int) -> None: + svc: SyncService = context.sync_service + for i in range(count): + entity = _make_actor_snapshot( + entity_id=f"actor-{i}", + namespace=namespace, + clock_entries={"server": 1}, + ) + svc.register_server_entity(entity) + + +@given('a local actor "{name}" in namespace "{namespace}" with clock {clock_json}') +def step_setup_local_actor_with_clock( + context: Any, name: str, namespace: str, clock_json: str +) -> None: + entries = json.loads(clock_json) + entity = _make_actor_snapshot( + entity_id=name, + namespace=namespace, + clock_entries=entries, + ) + context.sync_service.register_local_entity(entity) + context.local_entity = entity + + +@given( + 'a local actor "{name}" in namespace "{namespace}" clocked {clock_json} at "{timestamp}"' +) +def step_setup_local_actor_with_clock_ts( + context: Any, + name: str, + namespace: str, + clock_json: str, + timestamp: str, +) -> None: + entries = json.loads(clock_json) + entity = _make_actor_snapshot( + entity_id=name, + namespace=namespace, + clock_entries=entries, + updated_at=timestamp, + ) + context.sync_service.register_local_entity(entity) + context.local_entity = entity + + +@given('a server actor "{name}" in namespace "{namespace}" with clock {clock_json}') +def step_setup_server_actor_with_clock( + context: Any, name: str, namespace: str, clock_json: str +) -> None: + entries = json.loads(clock_json) + entity = _make_actor_snapshot( + entity_id=name, + namespace=namespace, + clock_entries=entries, + ) + context.sync_service.register_server_entity(entity) + context.server_entity = entity + + +@given( + 'a server actor "{name}" in namespace "{namespace}" clocked {clock_json} at "{timestamp}"' +) +def step_setup_server_actor_with_clock_ts( + context: Any, + name: str, + namespace: str, + clock_json: str, + timestamp: str, +) -> None: + entries = json.loads(clock_json) + entity = _make_actor_snapshot( + entity_id=name, + namespace=namespace, + clock_entries=entries, + updated_at=timestamp, + ) + context.sync_service.register_server_entity(entity) + context.server_entity = entity + + +@given('server entities in namespace "{namespace}" with timestamps') +def step_setup_server_timestamped(context: Any, namespace: str) -> None: + svc: SyncService = context.sync_service + old = _make_actor_snapshot( + entity_id="old-actor", + namespace=namespace, + clock_entries={"server": 1}, + updated_at="2024-01-01T00:00:00+00:00", + ) + new = _make_actor_snapshot( + entity_id="new-actor", + namespace=namespace, + clock_entries={"server": 2}, + updated_at="2025-06-01T00:00:00+00:00", + ) + svc.register_server_entity(old) + svc.register_server_entity(new) + + +@given('{count:d} local actors in namespace "{namespace}"') +def step_setup_local_actors(context: Any, count: int, namespace: str) -> None: + context.local_push_entities = [] + for i in range(count): + entity = _make_actor_snapshot( + entity_id=f"local-actor-{i}", + namespace=namespace, + clock_entries={"client": 1}, + ) + context.sync_service.register_local_entity(entity) + context.local_push_entities.append(entity) + + +@given('a local actor "{name}" in namespace "{namespace}" not on server') +def step_setup_local_only_actor(context: Any, name: str, namespace: str) -> None: + entity = _make_actor_snapshot( + entity_id=name, + namespace=namespace, + clock_entries={"client": 1}, + ) + context.sync_service.register_local_entity(entity) + + +@given('a local skill "{name}" in namespace "{namespace}" not on server') +def step_setup_local_only_skill(context: Any, name: str, namespace: str) -> None: + entity = _make_skill_snapshot( + entity_id=name, + namespace=namespace, + clock_entries={"client": 1}, + ) + context.sync_service.register_local_entity(entity) + + +@given('a server actor "{name}" in namespace "{namespace}" not on local') +def step_setup_server_only_actor(context: Any, name: str, namespace: str) -> None: + entity = _make_actor_snapshot( + entity_id=name, + namespace=namespace, + clock_entries={"server": 1}, + ) + context.sync_service.register_server_entity(entity) + + +@given('a synced actor "{name}" in namespace "{namespace}" on both sides') +def step_setup_synced_actor(context: Any, name: str, namespace: str) -> None: + entity = _make_actor_snapshot( + entity_id=name, + namespace=namespace, + clock_entries={"client": 1, "server": 1}, + ) + context.sync_service.register_local_entity(entity) + context.sync_service.register_server_entity(entity) + + +@given("an unresolved conflict between local and server versions") +def step_setup_unresolved_conflict(context: Any) -> None: + svc: SyncService = context.sync_service + local_entity = _make_actor_snapshot( + entity_id="conflict-ent", + namespace="team", + clock_entries={"client": 2, "server": 1}, + updated_at="2025-06-01T00:00:00+00:00", + ) + server_entity = _make_actor_snapshot( + entity_id="conflict-ent", + namespace="team", + clock_entries={"client": 1, "server": 2}, + updated_at="2025-05-01T00:00:00+00:00", + ) + svc.register_local_entity(local_entity) + svc.register_server_entity(server_entity) + # Trigger conflict via pull + pull_req = SyncPullRequest(namespace="team") + svc.pull(pull_req) + context.conflict_id = svc.conflicts[0].conflict_id + + +@given("an unresolved conflict where local is newer") +def step_setup_conflict_local_newer(context: Any) -> None: + svc: SyncService = context.sync_service + local_entity = _make_actor_snapshot( + entity_id="ts-conflict-ent", + namespace="team", + clock_entries={"client": 2, "server": 1}, + updated_at="2025-07-01T00:00:00+00:00", + ) + server_entity = _make_actor_snapshot( + entity_id="ts-conflict-ent", + namespace="team", + clock_entries={"client": 1, "server": 2}, + updated_at="2025-01-01T00:00:00+00:00", + ) + svc.register_local_entity(local_entity) + svc.register_server_entity(server_entity) + pull_req = SyncPullRequest(namespace="team") + svc.pull(pull_req) + context.conflict_id = svc.conflicts[0].conflict_id + + +@given("a local actor snapshot for offline queuing") +def step_setup_offline_snapshot(context: Any) -> None: + context.offline_entity = _make_actor_snapshot( + entity_id="offline-actor", + namespace="team", + clock_entries={"client": 1}, + ) + + +@given('a queued offline push for actor "{name}" in namespace "{namespace}"') +def step_setup_queued_push(context: Any, name: str, namespace: str) -> None: + entity = _make_actor_snapshot( + entity_id=name, + namespace=namespace, + clock_entries={"client": 1}, + ) + context.sync_service.enqueue_offline( + direction=SyncDirection.PUSH, + namespace=namespace, + entity=entity, + ) + + +@given("a queued offline push at max retries") +def step_setup_max_retries(context: Any) -> None: + entity = _make_actor_snapshot( + entity_id="max-retry-actor", + namespace="team", + clock_entries={"client": 1}, + ) + entry = context.sync_service.enqueue_offline( + direction=SyncDirection.PUSH, + namespace="team", + entity=entity, + ) + entry.retry_count = entry.max_retries + + +# ------------------------------------------------------------------ +# Pull action steps +# ------------------------------------------------------------------ + + +@when('I pull all entities from namespace "{namespace}"') +def step_pull_all(context: Any, namespace: str) -> None: + request = SyncPullRequest(namespace=namespace) + context.pull_response = context.sync_service.pull(request) + + +@when('I pull entities from namespace "{namespace}" since "{since}"') +def step_pull_since(context: Any, namespace: str, since: str) -> None: + request = SyncPullRequest(namespace=namespace, since=since) + context.pull_response = context.sync_service.pull(request) + + +@when('I force pull all entities from namespace "{namespace}"') +def step_force_pull(context: Any, namespace: str) -> None: + request = SyncPullRequest(namespace=namespace, force=True) + context.pull_response = context.sync_service.pull(request) + + +@when('I try to pull from namespace "{namespace}"') +def step_try_pull_local(context: Any, namespace: str) -> None: + context.caught_exception = None + try: + request = SyncPullRequest(namespace=namespace) + context.sync_service.pull(request) + except (ValidationError, PydanticValidationError) as exc: + context.caught_exception = exc + + +@when("I try to pull with a non-SyncPullRequest argument") +def step_pull_bad_type(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.pull("not a request") # type: ignore[arg-type] + except TypeError as exc: + context.caught_exception = exc + + +# ------------------------------------------------------------------ +# Push action steps +# ------------------------------------------------------------------ + + +@when('I push all local entities to namespace "{namespace}"') +def step_push_all(context: Any, namespace: str) -> None: + entities = getattr(context, "local_push_entities", []) + request = SyncPushRequest(namespace=namespace, entities=entities) + context.push_response = context.sync_service.push(request) + + +@when( + 'I push the local actor to namespace "{namespace}" with last-writer-wins resolution' +) +def step_push_lww(context: Any, namespace: str) -> None: + request = SyncPushRequest( + namespace=namespace, + entities=[context.local_entity], + resolution=ConflictResolution.LAST_WRITER_WINS, + ) + context.push_response = context.sync_service.push(request) + + +@when('I push the local actor to namespace "{namespace}" with server-wins resolution') +def step_push_server_wins(context: Any, namespace: str) -> None: + request = SyncPushRequest( + namespace=namespace, + entities=[context.local_entity], + resolution=ConflictResolution.SERVER_WINS, + ) + context.push_response = context.sync_service.push(request) + + +@when('I push the local actor to namespace "{namespace}" with manual resolution') +def step_push_manual(context: Any, namespace: str) -> None: + request = SyncPushRequest( + namespace=namespace, + entities=[context.local_entity], + resolution=ConflictResolution.MANUAL, + ) + context.push_response = context.sync_service.push(request) + + +@when('I force push a local actor "{name}" to namespace "{namespace}"') +def step_force_push(context: Any, name: str, namespace: str) -> None: + entity = _make_actor_snapshot( + entity_id=name, + namespace=namespace, + clock_entries={"client": 1}, + ) + request = SyncPushRequest(namespace=namespace, entities=[entity], force=True) + context.push_response = context.sync_service.push(request) + + +@when('I try to push to namespace "{namespace}"') +def step_try_push_local(context: Any, namespace: str) -> None: + context.caught_exception = None + try: + request = SyncPushRequest(namespace=namespace, entities=[]) + context.sync_service.push(request) + except (ValidationError, PydanticValidationError) as exc: + context.caught_exception = exc + + +@when("I try to push with a non-SyncPushRequest argument") +def step_push_bad_type(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.push("not a request") # type: ignore[arg-type] + except TypeError as exc: + context.caught_exception = exc + + +@when('I push an empty entity list to namespace "{namespace}"') +def step_push_empty(context: Any, namespace: str) -> None: + request = SyncPushRequest(namespace=namespace, entities=[]) + context.push_response = context.sync_service.push(request) + + +# ------------------------------------------------------------------ +# Status action steps +# ------------------------------------------------------------------ + + +@when('I request sync status for namespace "{namespace}"') +def step_check_status(context: Any, namespace: str) -> None: + request = SyncStatusRequest(namespace=namespace) + context.status_response = context.sync_service.status(request) + + +@when( + 'I request filtered sync status for namespace "{namespace}" entity type "{entity_type}"' +) +def step_check_status_filtered(context: Any, namespace: str, entity_type: str) -> None: + et = SyncEntityType(entity_type) + request = SyncStatusRequest(namespace=namespace, entity_types=[et]) + context.status_response = context.sync_service.status(request) + + +@when('I try requesting sync status for namespace "{namespace}"') +def step_try_status_local(context: Any, namespace: str) -> None: + context.caught_exception = None + try: + request = SyncStatusRequest(namespace=namespace) + context.sync_service.status(request) + except (ValidationError, PydanticValidationError) as exc: + context.caught_exception = exc + + +@when("I try to check status with a non-SyncStatusRequest argument") +def step_status_bad_type(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.status("not a request") # type: ignore[arg-type] + except TypeError as exc: + context.caught_exception = exc + + +# ------------------------------------------------------------------ +# Conflict resolution steps +# ------------------------------------------------------------------ + + +@when('I resolve the conflict manually choosing "{winner}"') +def step_resolve_manual(context: Any, winner: str) -> None: + svc: SyncService = context.sync_service + context.resolved_conflict = svc.resolve_conflict( + conflict_id=context.conflict_id, + resolution=ConflictResolution.MANUAL, + winner=winner, + ) + + +@when("I try to resolve manually without specifying a winner") +def step_try_resolve_no_winner(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.resolve_conflict( + conflict_id=context.conflict_id, + resolution=ConflictResolution.MANUAL, + ) + except BusinessRuleViolation as exc: + context.caught_exception = exc + + +@when('I try to resolve a conflict with id "{conflict_id}"') +def step_try_resolve_missing(context: Any, conflict_id: str) -> None: + context.caught_exception = None + try: + context.sync_service.resolve_conflict( + conflict_id=conflict_id, + resolution=ConflictResolution.LAST_WRITER_WINS, + ) + except ValueError as exc: + context.caught_exception = exc + + +@when("I try to resolve a conflict with an empty id") +def step_try_resolve_empty_id(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.resolve_conflict( + conflict_id="", + resolution=ConflictResolution.LAST_WRITER_WINS, + ) + except ValueError as exc: + context.caught_exception = exc + + +@when("I resolve the conflict with client_wins strategy") +def step_resolve_client_wins(context: Any) -> None: + context.resolved_conflict = context.sync_service.resolve_conflict( + conflict_id=context.conflict_id, + resolution=ConflictResolution.CLIENT_WINS, + ) + + +@when("I resolve the conflict with server_wins strategy") +def step_resolve_server_wins(context: Any) -> None: + context.resolved_conflict = context.sync_service.resolve_conflict( + conflict_id=context.conflict_id, + resolution=ConflictResolution.SERVER_WINS, + ) + + +@when("I resolve the conflict with last_writer_wins strategy") +def step_resolve_lww(context: Any) -> None: + context.resolved_conflict = context.sync_service.resolve_conflict( + conflict_id=context.conflict_id, + resolution=ConflictResolution.LAST_WRITER_WINS, + ) + + +@when("I try to resolve with an invalid resolution type") +def step_resolve_bad_type(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.resolve_conflict( + conflict_id=context.conflict_id, + resolution="invalid", # type: ignore[arg-type] + ) + except TypeError as exc: + context.caught_exception = exc + + +# ------------------------------------------------------------------ +# Offline queue steps +# ------------------------------------------------------------------ + + +@when('I enqueue an offline push to namespace "{namespace}"') +def step_enqueue_push(context: Any, namespace: str) -> None: + context.queue_entry = context.sync_service.enqueue_offline( + direction=SyncDirection.PUSH, + namespace=namespace, + entity=context.offline_entity, + ) + + +@when('I enqueue an offline pull from namespace "{namespace}"') +def step_enqueue_pull(context: Any, namespace: str) -> None: + context.queue_entry = context.sync_service.enqueue_offline( + direction=SyncDirection.PULL, + namespace=namespace, + entity=context.offline_entity, + ) + + +@when("I process the offline queue") +def step_process_queue(context: Any) -> None: + context.processed_entries = context.sync_service.process_offline_queue() + + +@when("I try to enqueue to an empty namespace") +def step_try_enqueue_empty_ns(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.enqueue_offline( + direction=SyncDirection.PUSH, + namespace="", + entity=context.offline_entity, + ) + except ValueError as exc: + context.caught_exception = exc + + +@when('I try to enqueue to namespace "{namespace}"') +def step_try_enqueue_local(context: Any, namespace: str) -> None: + context.caught_exception = None + try: + context.sync_service.enqueue_offline( + direction=SyncDirection.PUSH, + namespace=namespace, + entity=context.offline_entity, + ) + except ValueError as exc: + context.caught_exception = exc + + +@when("I try to enqueue with an invalid direction type") +def step_try_enqueue_bad_direction(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.enqueue_offline( + direction="invalid", # type: ignore[arg-type] + namespace="team", + entity=context.offline_entity, + ) + except TypeError as exc: + context.caught_exception = exc + + +@when("I try to enqueue with an invalid entity type") +def step_try_enqueue_bad_entity(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.enqueue_offline( + direction=SyncDirection.PUSH, + namespace="team", + entity="not an entity", # type: ignore[arg-type] + ) + except TypeError as exc: + context.caught_exception = exc + + +# ------------------------------------------------------------------ +# Pull response assertions +# ------------------------------------------------------------------ + + +@then("the pull response should contain {count:d} entities") +def step_pull_count(context: Any, count: int) -> None: + assert context.pull_response.total_pulled == count, ( + f"Expected {count} pulled, got {context.pull_response.total_pulled}" + ) + + +@then("the pull response should have {count:d} conflicts") +def step_pull_conflicts(context: Any, count: int) -> None: + assert len(context.pull_response.conflicts) == count, ( + f"Expected {count} conflicts, got {len(context.pull_response.conflicts)}" + ) + + +@then("only entities updated after the since timestamp are pulled") +def step_pull_only_newer(context: Any) -> None: + for entity in context.pull_response.entities: + if entity.updated_at: + assert entity.updated_at > "2025-01-01T00:00:00+00:00", ( + f"Entity {entity.entity_id} updated_at {entity.updated_at} " + f"should be after the since timestamp" + ) + + +@then("the pull response should have a sync state with last_pull_at set") +def step_pull_sync_state(context: Any) -> None: + assert context.pull_response.sync_state is not None + assert context.pull_response.sync_state.last_pull_at is not None + + +# ------------------------------------------------------------------ +# Push response assertions +# ------------------------------------------------------------------ + + +@then("the push response should have {count:d} accepted") +def step_push_accepted(context: Any, count: int) -> None: + assert context.push_response.total_accepted == count, ( + f"Expected {count} accepted, got {context.push_response.total_accepted}" + ) + + +@then("the push response should have {count:d} rejected") +def step_push_rejected(context: Any, count: int) -> None: + assert context.push_response.total_rejected == count, ( + f"Expected {count} rejected, got {context.push_response.total_rejected}" + ) + + +@then("the push response should have {count:d} conflicts") +def step_push_conflicts(context: Any, count: int) -> None: + assert len(context.push_response.conflicts) == count, ( + f"Expected {count} conflicts, got {len(context.push_response.conflicts)}" + ) + + +@then('the push conflict winner should be "{winner}"') +def step_push_conflict_winner(context: Any, winner: str) -> None: + assert len(context.push_response.conflicts) > 0, "No conflicts found" + assert context.push_response.conflicts[0].winner == winner, ( + f"Expected winner '{winner}', got '{context.push_response.conflicts[0].winner}'" + ) + + +@then("the push response should have a sync state with last_push_at set") +def step_push_sync_state(context: Any) -> None: + sync_state = context.push_response.model_dump().get("sync_state") + assert sync_state is not None + + +# ------------------------------------------------------------------ +# Status response assertions +# ------------------------------------------------------------------ + + +@then("the status should show drift detected") +def step_status_drift(context: Any) -> None: + assert context.status_response.drift_detected is True, ( + "Expected drift_detected to be True" + ) + + +@then("the status should show no drift") +def step_status_no_drift(context: Any) -> None: + assert context.status_response.drift_detected is False, ( + "Expected drift_detected to be False" + ) + + +@then("the status should have {count:d} local-ahead entities") +def step_status_local_ahead(context: Any, count: int) -> None: + assert len(context.status_response.local_ahead) == count, ( + f"Expected {count} local-ahead, got {len(context.status_response.local_ahead)}" + ) + + +@then("the status should have {count:d} server-ahead entities") +def step_status_server_ahead(context: Any, count: int) -> None: + assert len(context.status_response.server_ahead) == count, ( + f"Expected {count} server-ahead, " + f"got {len(context.status_response.server_ahead)}" + ) + + +# ------------------------------------------------------------------ +# Conflict resolution assertions +# ------------------------------------------------------------------ + + +@then("the conflict should be marked as resolved") +def step_conflict_resolved(context: Any) -> None: + assert context.resolved_conflict.resolved is True, ( + "Expected conflict to be resolved" + ) + + +@then('the conflict winner should be "{winner}"') +def step_conflict_winner(context: Any, winner: str) -> None: + assert context.resolved_conflict.winner == winner, ( + f"Expected winner '{winner}', got '{context.resolved_conflict.winner}'" + ) + + +# ------------------------------------------------------------------ +# Offline queue assertions +# ------------------------------------------------------------------ + + +@then("the offline queue should have {count:d} entry") +def step_queue_count(context: Any, count: int) -> None: + assert len(context.sync_service.offline_queue) == count, ( + f"Expected {count} queue entries, got {len(context.sync_service.offline_queue)}" + ) + + +@then('the queued entry should have status "{status}"') +def step_queue_status(context: Any, status: str) -> None: + assert context.queue_entry.status.value == status, ( + f"Expected status '{status}', got '{context.queue_entry.status.value}'" + ) + + +@then("the processed entries should include the offline actor") +def step_processed_includes_actor(context: Any) -> None: + assert len(context.processed_entries) > 0, "No entries processed" + + +@then("the offline queue should be empty") +def step_queue_empty(context: Any) -> None: + assert len(context.sync_service.offline_queue) == 0, ( + f"Expected empty queue, got {len(context.sync_service.offline_queue)}" + ) + + +@then('the failed entry should have status "failed"') +def step_failed_entry(context: Any) -> None: + failed = [ + e for e in context.processed_entries if e.status == SyncOperationStatus.FAILED + ] + assert len(failed) > 0, "Expected at least one failed entry" + + +# ------------------------------------------------------------------ +# Facade integration steps +# ------------------------------------------------------------------ + + +@given("a facade without sync service") +def step_facade_no_sync(context: Any) -> None: + context.facade = A2aLocalFacade() + + +@given("a facade with sync service containing server entities") +def step_facade_with_server_entities(context: Any) -> None: + svc = SyncService(node_id="facade-test") + entity = _make_actor_snapshot( + entity_id="facade-actor", + namespace="team", + clock_entries={"server": 1}, + ) + svc.register_server_entity(entity) + context.facade = A2aLocalFacade(services={"sync_service": svc}) + + +@given("a facade with sync service") +def step_facade_with_sync(context: Any) -> None: + svc = SyncService(node_id="facade-test") + context.facade = A2aLocalFacade(services={"sync_service": svc}) + + +@given("a facade with sync service containing local-only entities") +def step_facade_with_local_entities(context: Any) -> None: + svc = SyncService(node_id="facade-test") + entity = _make_actor_snapshot( + entity_id="local-only", + namespace="team", + clock_entries={"client": 1}, + ) + svc.register_local_entity(entity) + context.facade = A2aLocalFacade(services={"sync_service": svc}) + + +@when('I dispatch "{operation}" with sync params {params_json}') +def step_dispatch_sync(context: Any, operation: str, params_json: str) -> None: + params = json.loads(params_json) + request = A2aRequest(operation=operation, params=params) + context.facade_response = context.facade.dispatch(request) + + +@when('I dispatch "{operation}" with sync push params') +def step_dispatch_push(context: Any, operation: str) -> None: + params = { + "namespace": "team", + "entities": [ + { + "entity_id": "push-actor", + "entity_type": "actor", + "namespace": "team", + "name": "push-test", + }, + ], + } + request = A2aRequest(operation=operation, params=params) + context.facade_response = context.facade.dispatch(request) + + +@then("the facade response should be ok") +def step_facade_ok(context: Any) -> None: + assert context.facade_response.status == "ok", ( + f"Expected 'ok', got '{context.facade_response.status}'" + ) + + +@then('the facade response data should have "{key}" set to true') +def step_facade_stub_true(context: Any, key: str) -> None: + assert context.facade_response.data.get(key) is True, ( + f"Expected data['{key}'] to be True, " + f"got {context.facade_response.data.get(key)}" + ) + + +@then('the facade response data should contain key "{key}"') +def step_facade_has_key(context: Any, key: str) -> None: + assert key in context.facade_response.data, ( + f"Expected key '{key}' in data, " + f"got: {list(context.facade_response.data.keys())}" + ) + + +# ------------------------------------------------------------------ +# Validation error with message +# ------------------------------------------------------------------ + + +# ------------------------------------------------------------------ +# Constructor validation steps +# ------------------------------------------------------------------ + + +@when("I try to create a SyncService with a non-string node_id") +def step_create_svc_bad_nodeid(context: Any) -> None: + context.caught_exception = None + try: + SyncService(node_id=123) # type: ignore[arg-type] + except TypeError as exc: + context.caught_exception = exc + + +@when("I create a SyncService with None node_id") +def step_create_svc_none_nodeid(context: Any) -> None: + svc = SyncService(node_id=None) + context.created_service = svc + + +@then("the service should have a non-empty node_id") +def step_svc_has_nodeid(context: Any) -> None: + assert context.created_service.node_id, "Expected non-empty node_id" + + +@when("I try to register a non-snapshot as local entity") +def step_register_bad_local(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.register_local_entity("not an entity") # type: ignore[arg-type] + except TypeError as exc: + context.caught_exception = exc + + +@when("I try to register a non-snapshot as server entity") +def step_register_bad_server(context: Any) -> None: + context.caught_exception = None + try: + context.sync_service.register_server_entity("not an entity") # type: ignore[arg-type] + except TypeError as exc: + context.caught_exception = exc + + +# ------------------------------------------------------------------ +# Entity-sync-specific assertion steps (avoids conflicts with global steps) +# ------------------------------------------------------------------ + + +@then("a sync ValueError should be raised") +def step_sync_assert_value_error(context: Any) -> None: + assert isinstance(context.caught_exception, ValueError), ( + f"Expected ValueError, got {type(context.caught_exception)}" + ) + + +@then("a sync TypeError should be raised") +def step_sync_assert_type_error(context: Any) -> None: + assert isinstance(context.caught_exception, TypeError), ( + f"Expected TypeError, got {type(context.caught_exception)}" + ) + + +@then("a sync validation error should be raised") +def step_sync_assert_validation_error(context: Any) -> None: + assert isinstance( + context.caught_exception, (PydanticValidationError, ValueError) + ), f"Expected validation error, got {type(context.caught_exception)}" + + +@then("a sync BusinessRuleViolation should be raised") +def step_sync_assert_business_rule(context: Any) -> None: + assert isinstance(context.caught_exception, BusinessRuleViolation), ( + f"Expected BusinessRuleViolation, got {type(context.caught_exception)}" + ) + + +@then('a sync ValidationError should be raised with message containing "{text}"') +def step_sync_validation_error_message(context: Any, text: str) -> None: + assert context.caught_exception is not None, "No exception was raised" + assert text in str(context.caught_exception), ( + f"Expected '{text}' in error message, got: {context.caught_exception}" + ) diff --git a/features/steps/server_lifecycle_steps.py b/features/steps/server_lifecycle_steps.py index 6418f129d..646468af7 100644 --- a/features/steps/server_lifecycle_steps.py +++ b/features/steps/server_lifecycle_steps.py @@ -11,7 +11,7 @@ import contextlib from typing import Any from unittest.mock import MagicMock -from behave import given, then, when # type: ignore[import-untyped] +from behave import given, then, use_step_matcher, when # type: ignore[import-untyped] from cleveragents.a2a.facade import A2aLocalFacade from cleveragents.config.settings import Settings @@ -77,7 +77,10 @@ def step_check_value_error_port(context: Any) -> None: assert isinstance(context.caught_exception, ValueError) -@when('I try to create an ASGI application with host "{host}"') +use_step_matcher("re") + + +@when('I try to create an ASGI application with host "(?P[^"]*)"') def step_create_asgi_bad_host(context: Any, host: str) -> None: context.caught_exception = None try: @@ -95,6 +98,9 @@ def step_create_asgi_empty_host(context: Any) -> None: context.caught_exception = exc +use_step_matcher("parse") + + @then("a ValueError should be raised for invalid host") def step_check_value_error_host(context: Any) -> None: assert context.caught_exception is not None, "Expected ValueError" @@ -190,7 +196,10 @@ def step_check_not_stopped(context: Any) -> None: assert not context.lifecycle.is_stopped -@when('I try to create a ServerLifecycle with host "{host}"') +use_step_matcher("re") + + +@when('I try to create a ServerLifecycle with host "(?P[^"]*)"') def step_create_lifecycle_bad_host(context: Any, host: str) -> None: context.caught_exception = None try: @@ -208,6 +217,9 @@ def step_create_lifecycle_empty_host(context: Any) -> None: context.caught_exception = exc +use_step_matcher("parse") + + @when("I try to create a ServerLifecycle with port {port:d}") def step_create_lifecycle_bad_port(context: Any, port: int) -> None: context.caught_exception = None diff --git a/robot/entity_sync.robot b/robot/entity_sync.robot new file mode 100644 index 000000000..8016629bc --- /dev/null +++ b/robot/entity_sync.robot @@ -0,0 +1,74 @@ +*** Settings *** +Documentation Integration tests for entity sync (_cleveragents/sync/*) +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_entity_sync.py + +*** Test Cases *** +Sync Pull Via Facade + [Documentation] Verify sync/pull works through A2A facade with SyncService + ${result}= Run Process ${PYTHON} ${HELPER} sync-pull cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} sync-pull-ok + +Sync Push Via Facade + [Documentation] Verify sync/push works through A2A facade with SyncService + ${result}= Run Process ${PYTHON} ${HELPER} sync-push cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} sync-push-ok + +Sync Status Via Facade + [Documentation] Verify sync/status detects drift through A2A facade + ${result}= Run Process ${PYTHON} ${HELPER} sync-status cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} sync-status-ok + +Sync Conflict Resolution + [Documentation] Verify conflict detection and resolution with vector clocks + ${result}= Run Process ${PYTHON} ${HELPER} sync-conflict-resolution cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} sync-conflict-resolution-ok + +Sync Offline Queue + [Documentation] Verify offline queue enqueue and processing + ${result}= Run Process ${PYTHON} ${HELPER} sync-offline-queue cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} sync-offline-queue-ok + +Sync Vector Clock Operations + [Documentation] Verify vector clock increment, merge, happens-before, concurrent + ${result}= Run Process ${PYTHON} ${HELPER} sync-vector-clock cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} sync-vector-clock-ok + +Sync Facade Without Service Returns Stubs + [Documentation] Verify sync operations return stubs when no SyncService is registered + ${result}= Run Process ${PYTHON} ${HELPER} sync-facade-no-service cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} sync-facade-no-service-ok + +Sync Incremental Pull + [Documentation] Verify incremental sync only pulls entities updated after since timestamp + ${result}= Run Process ${PYTHON} ${HELPER} sync-incremental cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} sync-incremental-ok +*** Keywords *** diff --git a/robot/helper_entity_sync.py b/robot/helper_entity_sync.py new file mode 100644 index 000000000..394bba50d --- /dev/null +++ b/robot/helper_entity_sync.py @@ -0,0 +1,289 @@ +"""Helper script for entity_sync.robot integration tests. + +Each subcommand is a self-contained check that prints a sentinel on success. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +# Ensure local source tree is importable +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +from cleveragents.a2a.facade import A2aLocalFacade # noqa: E402 +from cleveragents.a2a.models import A2aRequest # noqa: E402 +from cleveragents.a2a.sync_models import ( # noqa: E402 + ConflictResolution, + SyncDirection, + SyncEntitySnapshot, + SyncEntityType, + SyncOperationStatus, + SyncPullRequest, + VectorClock, +) +from cleveragents.application.services.sync_service import SyncService # noqa: E402 + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + + +def sync_pull() -> None: + """Test sync pull via facade.""" + svc = SyncService(node_id="robot-client") + entity = SyncEntitySnapshot( + entity_id="robot-actor", + entity_type=SyncEntityType.ACTOR, + namespace="team", + name="robot-actor", + vector_clock=VectorClock(entries={"server": 1}), + ) + svc.register_server_entity(entity) + + facade = A2aLocalFacade(services={"sync_service": svc}) + request = A2aRequest( + operation="_cleveragents/sync/pull", + params={"namespace": "team"}, + ) + response = facade.dispatch(request) + if response.status == "ok" and response.data.get("total_pulled", 0) >= 1: + print("sync-pull-ok") + else: + print(f"FAIL: {response.data}", file=sys.stderr) + sys.exit(1) + + +def sync_push() -> None: + """Test sync push via facade.""" + svc = SyncService(node_id="robot-client") + facade = A2aLocalFacade(services={"sync_service": svc}) + + params = { + "namespace": "team", + "entities": [ + { + "entity_id": "push-actor", + "entity_type": "actor", + "namespace": "team", + "name": "push-test", + }, + ], + } + request = A2aRequest( + operation="_cleveragents/sync/push", + params=params, + ) + response = facade.dispatch(request) + if response.status == "ok" and response.data.get("total_accepted", 0) >= 1: + print("sync-push-ok") + else: + print(f"FAIL: {response.data}", file=sys.stderr) + sys.exit(1) + + +def sync_status() -> None: + """Test sync status via facade.""" + svc = SyncService(node_id="robot-client") + entity = SyncEntitySnapshot( + entity_id="status-actor", + entity_type=SyncEntityType.ACTOR, + namespace="team", + name="status-actor", + vector_clock=VectorClock(entries={"client": 1}), + ) + svc.register_local_entity(entity) + + facade = A2aLocalFacade(services={"sync_service": svc}) + request = A2aRequest( + operation="_cleveragents/sync/status", + params={"namespace": "team"}, + ) + response = facade.dispatch(request) + if response.status == "ok" and response.data.get("drift_detected") is True: + print("sync-status-ok") + else: + print(f"FAIL: {response.data}", file=sys.stderr) + sys.exit(1) + + +def sync_conflict_resolution() -> None: + """Test conflict detection and resolution.""" + svc = SyncService(node_id="robot-client") + + local_ent = SyncEntitySnapshot( + entity_id="conflict-actor", + entity_type=SyncEntityType.ACTOR, + namespace="team", + name="conflict-actor", + vector_clock=VectorClock(entries={"client": 2, "server": 1}), + updated_at="2025-06-01T12:00:00+00:00", + ) + server_ent = SyncEntitySnapshot( + entity_id="conflict-actor", + entity_type=SyncEntityType.ACTOR, + namespace="team", + name="conflict-actor", + vector_clock=VectorClock(entries={"client": 1, "server": 2}), + updated_at="2025-05-01T12:00:00+00:00", + ) + svc.register_local_entity(local_ent) + svc.register_server_entity(server_ent) + + pull_req = SyncPullRequest(namespace="team") + pull_resp = svc.pull(pull_req) + if len(pull_resp.conflicts) != 1: + print( + f"FAIL: expected 1 conflict, got {len(pull_resp.conflicts)}", + file=sys.stderr, + ) + sys.exit(1) + + conflict_id = svc.conflicts[0].conflict_id + resolved = svc.resolve_conflict( + conflict_id=conflict_id, + resolution=ConflictResolution.LAST_WRITER_WINS, + ) + if resolved.resolved and resolved.winner == "local": + print("sync-conflict-resolution-ok") + else: + print( + f"FAIL: resolved={resolved.resolved}, winner={resolved.winner}", + file=sys.stderr, + ) + sys.exit(1) + + +def sync_offline_queue() -> None: + """Test offline queue operations.""" + svc = SyncService(node_id="robot-client") + entity = SyncEntitySnapshot( + entity_id="offline-actor", + entity_type=SyncEntityType.ACTOR, + namespace="team", + name="offline-actor", + vector_clock=VectorClock(entries={"client": 1}), + ) + entry = svc.enqueue_offline( + direction=SyncDirection.PUSH, + namespace="team", + entity=entity, + ) + if entry.status != SyncOperationStatus.QUEUED: + print(f"FAIL: expected QUEUED, got {entry.status}", file=sys.stderr) + sys.exit(1) + + processed = svc.process_offline_queue() + if len(processed) >= 1 and len(svc.offline_queue) == 0: + print("sync-offline-queue-ok") + else: + print( + f"FAIL: processed={len(processed)}, remaining={len(svc.offline_queue)}", + file=sys.stderr, + ) + sys.exit(1) + + +def sync_vector_clock() -> None: + """Test vector clock operations.""" + c1 = VectorClock(entries={"a": 1, "b": 2}) + c2 = VectorClock(entries={"a": 2, "b": 3}) + + if not c1.happens_before(c2): + print("FAIL: c1 should happen before c2", file=sys.stderr) + sys.exit(1) + + c3 = VectorClock(entries={"a": 2, "b": 1}) + if not c1.is_concurrent(c3): + print("FAIL: c1 and c3 should be concurrent", file=sys.stderr) + sys.exit(1) + + merged = c1.merge(c2) + if merged.entries != {"a": 2, "b": 3}: + print(f"FAIL: unexpected merge result {merged.entries}", file=sys.stderr) + sys.exit(1) + + incremented = c1.increment("a") + if incremented.entries["a"] != 2: + print(f"FAIL: expected a=2, got {incremented.entries['a']}", file=sys.stderr) + sys.exit(1) + + print("sync-vector-clock-ok") + + +def sync_facade_no_service() -> None: + """Test facade sync stubs when no service is registered.""" + facade = A2aLocalFacade() + for op in ( + "_cleveragents/sync/pull", + "_cleveragents/sync/push", + "_cleveragents/sync/status", + ): + request = A2aRequest(operation=op, params={"namespace": "test"}) + response = facade.dispatch(request) + if response.status != "ok" or not response.data.get("stub"): + print( + f"FAIL: {op} should return stub, got {response.data}", file=sys.stderr + ) + sys.exit(1) + print("sync-facade-no-service-ok") + + +def sync_incremental() -> None: + """Test incremental sync with since parameter.""" + svc = SyncService(node_id="robot-client") + old_entity = SyncEntitySnapshot( + entity_id="old-actor", + entity_type=SyncEntityType.ACTOR, + namespace="team", + name="old-actor", + vector_clock=VectorClock(entries={"server": 1}), + updated_at="2024-01-01T00:00:00+00:00", + ) + new_entity = SyncEntitySnapshot( + entity_id="new-actor", + entity_type=SyncEntityType.ACTOR, + namespace="team", + name="new-actor", + vector_clock=VectorClock(entries={"server": 2}), + updated_at="2025-06-01T00:00:00+00:00", + ) + svc.register_server_entity(old_entity) + svc.register_server_entity(new_entity) + + pull_req = SyncPullRequest( + namespace="team", + since="2025-01-01T00:00:00+00:00", + ) + resp = svc.pull(pull_req) + if resp.total_pulled == 1 and resp.entities[0].entity_id == "new-actor": + print("sync-incremental-ok") + else: + print(f"FAIL: expected 1 entity, got {resp.total_pulled}", file=sys.stderr) + sys.exit(1) + + +# --------------------------------------------------------------------------- +# CLI dispatch +# --------------------------------------------------------------------------- + +_COMMANDS: dict[str, object] = { + "sync-pull": sync_pull, + "sync-push": sync_push, + "sync-status": sync_status, + "sync-conflict-resolution": sync_conflict_resolution, + "sync-offline-queue": sync_offline_queue, + "sync-vector-clock": sync_vector_clock, + "sync-facade-no-service": sync_facade_no_service, + "sync-incremental": sync_incremental, +} + +if __name__ == "__main__": + if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: + print(f"Usage: {sys.argv[0]} <{'|'.join(sorted(_COMMANDS))}>>", file=sys.stderr) + sys.exit(2) + cmd = _COMMANDS[sys.argv[1]] + if callable(cmd): + cmd() diff --git a/src/cleveragents/a2a/__init__.py b/src/cleveragents/a2a/__init__.py index 5a8228fa3..4b743333e 100644 --- a/src/cleveragents/a2a/__init__.py +++ b/src/cleveragents/a2a/__init__.py @@ -47,6 +47,23 @@ from cleveragents.a2a.models import ( ) from cleveragents.a2a.server_config import ServerConnectionConfig from cleveragents.a2a.stdio_transport import A2aStdioTransport +from cleveragents.a2a.sync_models import ( + ConflictResolution, + SyncConflict, + SyncDirection, + SyncEntitySnapshot, + SyncEntityType, + SyncOperationStatus, + SyncPullRequest, + SyncPullResponse, + SyncPushRequest, + SyncPushResponse, + SyncQueueEntry, + SyncState, + SyncStatusRequest, + SyncStatusResponse, + VectorClock, +) from cleveragents.a2a.transport import A2aHttpTransport from cleveragents.a2a.transport_selector import TransportSelector from cleveragents.a2a.versioning import A2aVersionNegotiator @@ -67,11 +84,26 @@ __all__ = [ "A2aVersionMismatchError", "A2aVersionNegotiator", "AuthClient", + "ConflictResolution", "RemoteExecutionClient", "ServerClient", "ServerConnectionConfig", "StubAuthClient", "StubRemoteExecutionClient", "StubServerClient", + "SyncConflict", + "SyncDirection", + "SyncEntitySnapshot", + "SyncEntityType", + "SyncOperationStatus", + "SyncPullRequest", + "SyncPullResponse", + "SyncPushRequest", + "SyncPushResponse", + "SyncQueueEntry", + "SyncState", + "SyncStatusRequest", + "SyncStatusResponse", "TransportSelector", + "VectorClock", ] diff --git a/src/cleveragents/a2a/facade.py b/src/cleveragents/a2a/facade.py index 4a186cfb1..517cc02ef 100644 --- a/src/cleveragents/a2a/facade.py +++ b/src/cleveragents/a2a/facade.py @@ -21,7 +21,7 @@ response so the facade never crashes due to missing wiring. from __future__ import annotations import time -from typing import Any, cast +from typing import TYPE_CHECKING, Any, cast import structlog from ulid import ULID @@ -55,6 +55,9 @@ from cleveragents.domain.models.core.session import ( from cleveragents.providers.registry import ProviderRegistry from cleveragents.tool.registry import ToolRegistry +if TYPE_CHECKING: + from cleveragents.application.services.sync_service import SyncService + logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__) # --------------------------------------------------------------------------- @@ -178,6 +181,10 @@ class A2aLocalFacade: def _event_queue(self) -> A2aEventQueue | None: return cast(A2aEventQueue | None, self._services.get("event_queue")) + @property + def _sync_service(self) -> SyncService | None: + return self._services.get("sync_service") # type: ignore[return-value] + # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ @@ -311,9 +318,9 @@ class A2aLocalFacade: "_cleveragents/health/check": self._handle_health_check, "_cleveragents/diagnostics/run": self._handle_diagnostics_run, # Sync (stub) - "_cleveragents/sync/pull": self._handle_sync_stub, - "_cleveragents/sync/push": self._handle_sync_stub, - "_cleveragents/sync/status": self._handle_sync_stub, + "_cleveragents/sync/pull": self._handle_sync_pull, + "_cleveragents/sync/push": self._handle_sync_push, + "_cleveragents/sync/status": self._handle_sync_status, # Namespace (stub) "_cleveragents/namespace/list": self._handle_namespace_stub, "_cleveragents/namespace/show": self._handle_namespace_stub, @@ -763,11 +770,38 @@ class A2aLocalFacade: return {"status": "ok", "diagnostics": {}, "stub": True} # ------------------------------------------------------------------ - # Extension handlers — _cleveragents/sync/* (stubs) + # Extension handlers — _cleveragents/sync/* # ------------------------------------------------------------------ - def _handle_sync_stub(self, params: dict[str, Any]) -> dict[str, Any]: - return {"status": "not_implemented", "stub": True} + def _handle_sync_pull(self, params: dict[str, Any]) -> dict[str, Any]: + svc = self._sync_service + if svc is None: + return {"status": "no_sync_service", "entities": [], "stub": True} + from cleveragents.a2a.sync_models import SyncPullRequest + + request = SyncPullRequest(**params) + response = svc.pull(request) + return response.model_dump() + + def _handle_sync_push(self, params: dict[str, Any]) -> dict[str, Any]: + svc = self._sync_service + if svc is None: + return {"status": "no_sync_service", "accepted": [], "stub": True} + from cleveragents.a2a.sync_models import SyncPushRequest + + request = SyncPushRequest(**params) + response = svc.push(request) + return response.model_dump() + + def _handle_sync_status(self, params: dict[str, Any]) -> dict[str, Any]: + svc = self._sync_service + if svc is None: + return {"status": "no_sync_service", "drift_detected": False, "stub": True} + from cleveragents.a2a.sync_models import SyncStatusRequest + + request = SyncStatusRequest(**params) + response = svc.status(request) + return response.model_dump() # ------------------------------------------------------------------ # Extension handlers — _cleveragents/namespace/* (stubs) diff --git a/src/cleveragents/a2a/sync_models.py b/src/cleveragents/a2a/sync_models.py new file mode 100644 index 000000000..f930a0e1b --- /dev/null +++ b/src/cleveragents/a2a/sync_models.py @@ -0,0 +1,403 @@ +"""Pydantic models for entity synchronization. + +Defines the request/response models, sync state, conflict records, and +vector clocks used by the ``_cleveragents/sync/*`` extension methods. + +Entity types eligible for sync: actors, skills, actions, plans, sessions. +The ``local/`` namespace is **never** synced — it exists only on the client. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, ConfigDict, field_validator + +# --------------------------------------------------------------------------- +# Enumerations +# --------------------------------------------------------------------------- + + +class SyncDirection(StrEnum): + """Direction of a sync operation.""" + + PULL = "pull" + PUSH = "push" + + +class SyncEntityType(StrEnum): + """Entity types eligible for synchronization.""" + + ACTOR = "actor" + SKILL = "skill" + ACTION = "action" + PLAN = "plan" + SESSION = "session" + + +class ConflictResolution(StrEnum): + """Strategies for resolving sync conflicts.""" + + LAST_WRITER_WINS = "last_writer_wins" + MANUAL = "manual" + SERVER_WINS = "server_wins" + CLIENT_WINS = "client_wins" + + +class SyncOperationStatus(StrEnum): + """Status of a sync operation.""" + + PENDING = "pending" + IN_PROGRESS = "in_progress" + COMPLETED = "completed" + FAILED = "failed" + CONFLICT = "conflict" + QUEUED = "queued" + + +# --------------------------------------------------------------------------- +# Vector clock +# --------------------------------------------------------------------------- + + +class VectorClock(BaseModel): + """Vector clock for conflict detection across distributed nodes. + + Each entry maps a node identifier to a monotonically increasing + counter. Two clocks are compared element-wise: if every component + of A is ``<=`` the corresponding component of B (and at least one + is strictly less) then A *happened before* B. Otherwise the clocks + are concurrent and a conflict exists. + """ + + model_config = ConfigDict(strict=False) + + entries: dict[str, int] = {} + + def increment(self, node_id: str) -> VectorClock: + """Return a new clock with *node_id* incremented by one. + + Args: + node_id: Identifier of the node to increment. + + Returns: + A new ``VectorClock`` with the updated counter. + + Raises: + ValueError: If *node_id* is empty. + """ + if not node_id: + raise ValueError("node_id must be a non-empty string") + new_entries = dict(self.entries) + new_entries[node_id] = new_entries.get(node_id, 0) + 1 + return VectorClock(entries=new_entries) + + def merge(self, other: VectorClock) -> VectorClock: + """Return a new clock that is the element-wise max of both clocks. + + Args: + other: The vector clock to merge with. + + Returns: + A new ``VectorClock`` with the merged counters. + + Raises: + TypeError: If *other* is not a ``VectorClock``. + """ + if not isinstance(other, VectorClock): + raise TypeError("other must be a VectorClock instance") + all_nodes = set(self.entries) | set(other.entries) + merged: dict[str, int] = {} + for node in all_nodes: + merged[node] = max( + self.entries.get(node, 0), + other.entries.get(node, 0), + ) + return VectorClock(entries=merged) + + def happens_before(self, other: VectorClock) -> bool: + """Return ``True`` if this clock causally precedes *other*. + + Args: + other: The vector clock to compare against. + + Returns: + ``True`` if every component of *self* is ``<=`` the + corresponding component of *other* and at least one is + strictly less. + + Raises: + TypeError: If *other* is not a ``VectorClock``. + """ + if not isinstance(other, VectorClock): + raise TypeError("other must be a VectorClock instance") + all_nodes = set(self.entries) | set(other.entries) + at_least_one_less = False + for node in all_nodes: + self_val = self.entries.get(node, 0) + other_val = other.entries.get(node, 0) + if self_val > other_val: + return False + if self_val < other_val: + at_least_one_less = True + return at_least_one_less + + def is_concurrent(self, other: VectorClock) -> bool: + """Return ``True`` if the two clocks are concurrent (conflict). + + Args: + other: The vector clock to compare against. + + Returns: + ``True`` if neither clock happens-before the other. + + Raises: + TypeError: If *other* is not a ``VectorClock``. + """ + if not isinstance(other, VectorClock): + raise TypeError("other must be a VectorClock instance") + return ( + not self.happens_before(other) + and not other.happens_before(self) + and self.entries != other.entries + ) + + +# --------------------------------------------------------------------------- +# Sync entity snapshot +# --------------------------------------------------------------------------- + + +class SyncEntitySnapshot(BaseModel): + """A snapshot of a single entity for sync purposes.""" + + model_config = ConfigDict(strict=False) + + entity_id: str + entity_type: SyncEntityType + namespace: str + name: str + version: int = 1 + data: dict[str, Any] = {} + vector_clock: VectorClock = VectorClock() + updated_at: str = "" + + @field_validator("entity_id", "namespace", "name") + @classmethod + def _must_be_non_empty(cls, value: str) -> str: + if not value: + raise ValueError("field must not be empty") + return value + + +# --------------------------------------------------------------------------- +# Conflict record +# --------------------------------------------------------------------------- + + +class SyncConflict(BaseModel): + """A record of a detected sync conflict between local and server.""" + + model_config = ConfigDict(strict=False) + + conflict_id: str + entity_id: str + entity_type: SyncEntityType + namespace: str + local_snapshot: SyncEntitySnapshot + server_snapshot: SyncEntitySnapshot + local_clock: VectorClock + server_clock: VectorClock + resolution: ConflictResolution = ConflictResolution.LAST_WRITER_WINS + resolved: bool = False + resolved_at: str | None = None + winner: str | None = None + + @field_validator("conflict_id", "entity_id", "namespace") + @classmethod + def _must_be_non_empty(cls, value: str) -> str: + if not value: + raise ValueError("field must not be empty") + return value + + +# --------------------------------------------------------------------------- +# Sync state +# --------------------------------------------------------------------------- + + +class SyncState(BaseModel): + """Tracks the sync state between client and server for a namespace.""" + + model_config = ConfigDict(strict=False) + + namespace: str + last_sync_at: str | None = None + last_pull_at: str | None = None + last_push_at: str | None = None + local_clock: VectorClock = VectorClock() + server_clock: VectorClock = VectorClock() + pending_pushes: int = 0 + pending_pulls: int = 0 + conflicts: list[SyncConflict] = [] + + @field_validator("namespace") + @classmethod + def _namespace_non_empty(cls, value: str) -> str: + if not value: + raise ValueError("namespace must not be empty") + return value + + +# --------------------------------------------------------------------------- +# Request / Response models +# --------------------------------------------------------------------------- + + +class SyncPullRequest(BaseModel): + """Request model for ``_cleveragents/sync/pull``.""" + + model_config = ConfigDict(strict=False) + + namespace: str = "" + entity_types: list[SyncEntityType] = [] + since: str | None = None + force: bool = False + + @field_validator("namespace") + @classmethod + def _validate_namespace(cls, value: str) -> str: + if value == "local": + raise ValueError("Cannot sync the 'local' namespace") + return value + + +class SyncPullResponse(BaseModel): + """Response model for ``_cleveragents/sync/pull``.""" + + model_config = ConfigDict(strict=False) + + entities: list[SyncEntitySnapshot] = [] + conflicts: list[SyncConflict] = [] + sync_state: SyncState | None = None + pulled_at: str = "" + total_pulled: int = 0 + + +class SyncPushRequest(BaseModel): + """Request model for ``_cleveragents/sync/push``.""" + + model_config = ConfigDict(strict=False) + + namespace: str = "" + entities: list[SyncEntitySnapshot] = [] + force: bool = False + resolution: ConflictResolution = ConflictResolution.LAST_WRITER_WINS + + @field_validator("namespace") + @classmethod + def _validate_namespace(cls, value: str) -> str: + if value == "local": + raise ValueError("Cannot sync the 'local' namespace") + return value + + +class SyncPushResponse(BaseModel): + """Response model for ``_cleveragents/sync/push``.""" + + model_config = ConfigDict(strict=False) + + accepted: list[SyncEntitySnapshot] = [] + rejected: list[SyncEntitySnapshot] = [] + conflicts: list[SyncConflict] = [] + sync_state: SyncState | None = None + pushed_at: str = "" + total_accepted: int = 0 + total_rejected: int = 0 + + +class SyncStatusRequest(BaseModel): + """Request model for ``_cleveragents/sync/status``.""" + + model_config = ConfigDict(strict=False) + + namespace: str = "" + entity_types: list[SyncEntityType] = [] + + @field_validator("namespace") + @classmethod + def _validate_namespace(cls, value: str) -> str: + if value == "local": + raise ValueError("Cannot sync the 'local' namespace") + return value + + +class SyncStatusResponse(BaseModel): + """Response model for ``_cleveragents/sync/status``.""" + + model_config = ConfigDict(strict=False) + + namespace: str = "" + sync_state: SyncState | None = None + drift_detected: bool = False + local_ahead: list[SyncEntitySnapshot] = [] + server_ahead: list[SyncEntitySnapshot] = [] + conflicts: list[SyncConflict] = [] + checked_at: str = "" + + +# --------------------------------------------------------------------------- +# Offline queue entry +# --------------------------------------------------------------------------- + + +class SyncQueueEntry(BaseModel): + """An offline-queued sync operation awaiting retry on reconnect.""" + + model_config = ConfigDict(strict=False) + + queue_id: str + direction: SyncDirection + namespace: str + entity_type: SyncEntityType + entity_id: str + snapshot: SyncEntitySnapshot + status: SyncOperationStatus = SyncOperationStatus.QUEUED + retry_count: int = 0 + max_retries: int = 5 + created_at: str = "" + last_attempt_at: str | None = None + error_message: str | None = None + + @field_validator("queue_id", "namespace", "entity_id") + @classmethod + def _must_be_non_empty(cls, value: str) -> str: + if not value: + raise ValueError("field must not be empty") + return value + + +def _iso_now() -> str: + """Return the current UTC time as an ISO-8601 string.""" + return datetime.now(tz=UTC).isoformat() + + +__all__ = [ + "ConflictResolution", + "SyncConflict", + "SyncDirection", + "SyncEntitySnapshot", + "SyncEntityType", + "SyncOperationStatus", + "SyncPullRequest", + "SyncPullResponse", + "SyncPushRequest", + "SyncPushResponse", + "SyncQueueEntry", + "SyncState", + "SyncStatusRequest", + "SyncStatusResponse", + "VectorClock", +] diff --git a/src/cleveragents/application/services/sync_service.py b/src/cleveragents/application/services/sync_service.py new file mode 100644 index 000000000..0b5cfaa9d --- /dev/null +++ b/src/cleveragents/application/services/sync_service.py @@ -0,0 +1,733 @@ +"""Entity synchronization service for multi-device sync. + +Implements ``_cleveragents/sync/*`` operations: pull, push, and status. +Entities (actors, skills, actions, plans, sessions) are synchronised +between client and server using vector clocks for conflict detection +and last-writer-wins as the default conflict resolution strategy. + +The ``local/`` namespace is **never** synced — it exists only on the +client. Incremental sync is supported via ``since`` timestamps. + +Offline support is provided through a queue of pending operations that +are retried on reconnect. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import structlog +from ulid import ULID + +from cleveragents.a2a.sync_models import ( + ConflictResolution, + SyncConflict, + SyncDirection, + SyncEntitySnapshot, + SyncEntityType, + SyncOperationStatus, + SyncPullRequest, + SyncPullResponse, + SyncPushRequest, + SyncPushResponse, + SyncQueueEntry, + SyncState, + SyncStatusRequest, + SyncStatusResponse, + VectorClock, +) +from cleveragents.core.exceptions import BusinessRuleViolation, ValidationError + +_logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__) + + +def _iso_now() -> str: + """Return the current UTC time as an ISO-8601 string.""" + return datetime.now(tz=UTC).isoformat() + + +class SyncService: + """Application-layer service for entity synchronization. + + Coordinates pull, push, and status operations between local client + storage and a remote server namespace. + + Args: + node_id: Unique identifier for this client node (used in + vector clocks). + """ + + def __init__(self, node_id: str | None = None) -> None: + if node_id is not None and not isinstance(node_id, str): + raise TypeError("node_id must be a string or None") + self._node_id: str = node_id or str(ULID()) + self._sync_states: dict[str, SyncState] = {} + self._local_entities: dict[str, SyncEntitySnapshot] = {} + self._server_entities: dict[str, SyncEntitySnapshot] = {} + self._conflicts: list[SyncConflict] = [] + self._offline_queue: list[SyncQueueEntry] = [] + + # ------------------------------------------------------------------ + # Public properties + # ------------------------------------------------------------------ + + @property + def node_id(self) -> str: + """Return the unique node identifier for this client.""" + return self._node_id + + @property + def offline_queue(self) -> list[SyncQueueEntry]: + """Return the current offline operation queue.""" + return list(self._offline_queue) + + @property + def conflicts(self) -> list[SyncConflict]: + """Return all unresolved sync conflicts.""" + return [c for c in self._conflicts if not c.resolved] + + # ------------------------------------------------------------------ + # Core operations + # ------------------------------------------------------------------ + + def pull(self, request: SyncPullRequest) -> SyncPullResponse: + """Pull entities from the server to the local cache. + + Downloads server namespace entities and detects conflicts + using vector clocks. + + Args: + request: The pull request parameters. + + Returns: + A ``SyncPullResponse`` with pulled entities and any + conflicts detected. + + Raises: + TypeError: If *request* is not a ``SyncPullRequest``. + ValidationError: If the namespace is ``local``. + """ + if not isinstance(request, SyncPullRequest): + raise TypeError("request must be a SyncPullRequest instance") + if request.namespace == "local": + raise ValidationError("Cannot sync the 'local' namespace") + + namespace = request.namespace or "default" + now = _iso_now() + + _logger.info( + "sync.pull.start", + namespace=namespace, + entity_types=[et.value for et in request.entity_types], + since=request.since, + ) + + pulled: list[SyncEntitySnapshot] = [] + conflicts: list[SyncConflict] = [] + + server_entities = self._get_server_entities( + namespace=namespace, + entity_types=request.entity_types, + since=request.since, + ) + + for server_entity in server_entities: + local_key = self._entity_key(server_entity) + local_entity = self._local_entities.get(local_key) + + if local_entity is None or request.force: + self._local_entities[local_key] = server_entity + pulled.append(server_entity) + elif server_entity.vector_clock.is_concurrent(local_entity.vector_clock): + conflict = self._create_conflict( + local_entity=local_entity, + server_entity=server_entity, + namespace=namespace, + ) + conflicts.append(conflict) + self._conflicts.append(conflict) + elif local_entity.vector_clock.happens_before(server_entity.vector_clock): + self._local_entities[local_key] = server_entity + pulled.append(server_entity) + + sync_state = self._update_sync_state( + namespace=namespace, + last_pull_at=now, + conflicts=conflicts, + ) + + _logger.info( + "sync.pull.complete", + namespace=namespace, + total_pulled=len(pulled), + total_conflicts=len(conflicts), + ) + + return SyncPullResponse( + entities=pulled, + conflicts=conflicts, + sync_state=sync_state, + pulled_at=now, + total_pulled=len(pulled), + ) + + def push(self, request: SyncPushRequest) -> SyncPushResponse: + """Push local entities to the server namespace. + + Pushes entity definitions from the local cache to the server + and handles conflicts according to the specified resolution + strategy. + + Args: + request: The push request parameters. + + Returns: + A ``SyncPushResponse`` with accepted/rejected entities + and any conflicts detected. + + Raises: + TypeError: If *request* is not a ``SyncPushRequest``. + ValidationError: If the namespace is ``local``. + """ + if not isinstance(request, SyncPushRequest): + raise TypeError("request must be a SyncPushRequest instance") + if request.namespace == "local": + raise ValidationError("Cannot sync the 'local' namespace") + + namespace = request.namespace or "default" + now = _iso_now() + + _logger.info( + "sync.push.start", + namespace=namespace, + entity_count=len(request.entities), + ) + + accepted: list[SyncEntitySnapshot] = [] + rejected: list[SyncEntitySnapshot] = [] + conflicts: list[SyncConflict] = [] + + for entity in request.entities: + server_key = self._entity_key(entity) + server_entity = self._server_entities.get(server_key) + + if server_entity is None or request.force: + updated = self._stamp_entity(entity, namespace) + self._server_entities[server_key] = updated + self._local_entities[server_key] = updated + accepted.append(updated) + elif entity.vector_clock.is_concurrent(server_entity.vector_clock): + conflict = self._create_conflict( + local_entity=entity, + server_entity=server_entity, + namespace=namespace, + ) + resolved_entity = self._resolve_conflict( + conflict=conflict, + resolution=request.resolution, + ) + if resolved_entity is not None: + self._server_entities[server_key] = resolved_entity + self._local_entities[server_key] = resolved_entity + accepted.append(resolved_entity) + conflict.resolved = True + conflict.resolved_at = now + else: + rejected.append(entity) + conflicts.append(conflict) + self._conflicts.append(conflict) + elif server_entity.vector_clock.happens_before(entity.vector_clock): + updated = self._stamp_entity(entity, namespace) + self._server_entities[server_key] = updated + self._local_entities[server_key] = updated + accepted.append(updated) + else: + rejected.append(entity) + + sync_state = self._update_sync_state( + namespace=namespace, + last_push_at=now, + conflicts=conflicts, + ) + + _logger.info( + "sync.push.complete", + namespace=namespace, + total_accepted=len(accepted), + total_rejected=len(rejected), + total_conflicts=len(conflicts), + ) + + return SyncPushResponse( + accepted=accepted, + rejected=rejected, + conflicts=conflicts, + sync_state=sync_state, + pushed_at=now, + total_accepted=len(accepted), + total_rejected=len(rejected), + ) + + def status(self, request: SyncStatusRequest) -> SyncStatusResponse: + """Compare local and server entity versions to detect drift. + + Args: + request: The status request parameters. + + Returns: + A ``SyncStatusResponse`` with drift information and + any detected conflicts. + + Raises: + TypeError: If *request* is not a ``SyncStatusRequest``. + ValidationError: If the namespace is ``local``. + """ + if not isinstance(request, SyncStatusRequest): + raise TypeError("request must be a SyncStatusRequest instance") + if request.namespace == "local": + raise ValidationError("Cannot sync the 'local' namespace") + + namespace = request.namespace or "default" + now = _iso_now() + + _logger.info( + "sync.status.start", + namespace=namespace, + entity_types=[et.value for et in request.entity_types], + ) + + local_ahead: list[SyncEntitySnapshot] = [] + server_ahead: list[SyncEntitySnapshot] = [] + conflicts: list[SyncConflict] = [] + + all_keys = set(self._local_entities.keys()) | set(self._server_entities.keys()) + + for key in all_keys: + local_entity = self._local_entities.get(key) + server_entity = self._server_entities.get(key) + + if local_entity is not None and server_entity is not None: + if namespace and local_entity.namespace != namespace: + continue + if ( + request.entity_types + and local_entity.entity_type not in request.entity_types + ): + continue + + if local_entity.vector_clock.is_concurrent( + server_entity.vector_clock, + ): + conflict = self._create_conflict( + local_entity=local_entity, + server_entity=server_entity, + namespace=namespace, + ) + conflicts.append(conflict) + elif server_entity.vector_clock.happens_before( + local_entity.vector_clock, + ): + local_ahead.append(local_entity) + elif local_entity.vector_clock.happens_before( + server_entity.vector_clock, + ): + server_ahead.append(server_entity) + elif local_entity is not None: + if namespace and local_entity.namespace != namespace: + continue + if ( + request.entity_types + and local_entity.entity_type not in request.entity_types + ): + continue + local_ahead.append(local_entity) + elif server_entity is not None: + if namespace and server_entity.namespace != namespace: + continue + if ( + request.entity_types + and server_entity.entity_type not in request.entity_types + ): + continue + server_ahead.append(server_entity) + + sync_state = self._sync_states.get(namespace) + drift_detected = bool(local_ahead or server_ahead or conflicts) + + _logger.info( + "sync.status.complete", + namespace=namespace, + drift_detected=drift_detected, + local_ahead=len(local_ahead), + server_ahead=len(server_ahead), + conflicts=len(conflicts), + ) + + return SyncStatusResponse( + namespace=namespace, + sync_state=sync_state, + drift_detected=drift_detected, + local_ahead=local_ahead, + server_ahead=server_ahead, + conflicts=conflicts, + checked_at=now, + ) + + # ------------------------------------------------------------------ + # Offline queue management + # ------------------------------------------------------------------ + + def enqueue_offline( + self, + direction: SyncDirection, + namespace: str, + entity: SyncEntitySnapshot, + ) -> SyncQueueEntry: + """Queue a sync operation for retry when offline. + + Args: + direction: Whether this is a pull or push operation. + namespace: Target namespace. + entity: The entity snapshot to sync. + + Returns: + The created ``SyncQueueEntry``. + + Raises: + ValueError: If *namespace* is empty or ``local``. + TypeError: If arguments have incorrect types. + """ + if not isinstance(direction, SyncDirection): + raise TypeError("direction must be a SyncDirection") + if not isinstance(entity, SyncEntitySnapshot): + raise TypeError("entity must be a SyncEntitySnapshot") + if not namespace: + raise ValueError("namespace must be a non-empty string") + if namespace == "local": + raise ValueError("Cannot queue sync for the 'local' namespace") + + entry = SyncQueueEntry( + queue_id=str(ULID()), + direction=direction, + namespace=namespace, + entity_type=entity.entity_type, + entity_id=entity.entity_id, + snapshot=entity, + status=SyncOperationStatus.QUEUED, + created_at=_iso_now(), + ) + self._offline_queue.append(entry) + + _logger.info( + "sync.offline.enqueued", + queue_id=entry.queue_id, + direction=direction.value, + namespace=namespace, + entity_id=entity.entity_id, + ) + + return entry + + def process_offline_queue(self) -> list[SyncQueueEntry]: + """Process all queued offline operations. + + Attempts to execute each queued operation. Successfully + processed entries are marked as completed; failed entries + have their retry count incremented. + + Returns: + List of processed queue entries with updated statuses. + """ + processed: list[SyncQueueEntry] = [] + remaining: list[SyncQueueEntry] = [] + + for entry in self._offline_queue: + if entry.retry_count >= entry.max_retries: + entry.status = SyncOperationStatus.FAILED + entry.error_message = "Max retries exceeded" + processed.append(entry) + continue + + entry.status = SyncOperationStatus.IN_PROGRESS + entry.last_attempt_at = _iso_now() + + try: + if entry.direction == SyncDirection.PUSH: + push_req = SyncPushRequest( + namespace=entry.namespace, + entities=[entry.snapshot], + ) + result = self.push(push_req) + if result.total_accepted > 0: + entry.status = SyncOperationStatus.COMPLETED + processed.append(entry) + else: + entry.status = SyncOperationStatus.QUEUED + entry.retry_count += 1 + remaining.append(entry) + elif entry.direction == SyncDirection.PULL: + pull_req = SyncPullRequest( + namespace=entry.namespace, + entity_types=[entry.entity_type], + ) + result = self.pull(pull_req) + entry.status = SyncOperationStatus.COMPLETED + processed.append(entry) + except Exception as exc: + entry.status = SyncOperationStatus.QUEUED + entry.retry_count += 1 + entry.error_message = str(exc) + remaining.append(entry) + + _logger.warning( + "sync.offline.retry_failed", + queue_id=entry.queue_id, + retry_count=entry.retry_count, + error=str(exc), + ) + + self._offline_queue = remaining + + _logger.info( + "sync.offline.processed", + total_processed=len(processed), + total_remaining=len(remaining), + ) + + return processed + + def resolve_conflict( + self, + conflict_id: str, + resolution: ConflictResolution, + winner: str | None = None, + ) -> SyncConflict: + """Manually resolve a sync conflict. + + Args: + conflict_id: Identifier of the conflict to resolve. + resolution: The resolution strategy to apply. + winner: For manual resolution, which side wins + (``'local'`` or ``'server'``). + + Returns: + The updated ``SyncConflict``. + + Raises: + ValueError: If *conflict_id* is empty or the conflict + is not found. + BusinessRuleViolation: If manual resolution is chosen + without specifying a winner. + """ + if not conflict_id: + raise ValueError("conflict_id must be a non-empty string") + if not isinstance(resolution, ConflictResolution): + raise TypeError("resolution must be a ConflictResolution") + + target: SyncConflict | None = None + for conflict in self._conflicts: + if conflict.conflict_id == conflict_id: + target = conflict + break + + if target is None: + raise ValueError(f"Conflict '{conflict_id}' not found") + + if resolution == ConflictResolution.MANUAL and not winner: + raise BusinessRuleViolation( + "Manual resolution requires specifying a winner ('local' or 'server')" + ) + + now = _iso_now() + target.resolution = resolution + target.resolved = True + target.resolved_at = now + + if resolution == ConflictResolution.LAST_WRITER_WINS: + local_ts = target.local_snapshot.updated_at or "" + server_ts = target.server_snapshot.updated_at or "" + target.winner = "local" if local_ts >= server_ts else "server" + elif resolution == ConflictResolution.CLIENT_WINS: + target.winner = "local" + elif resolution == ConflictResolution.SERVER_WINS: + target.winner = "server" + elif resolution == ConflictResolution.MANUAL: + target.winner = winner + + winning_snapshot = ( + target.local_snapshot + if target.winner == "local" + else target.server_snapshot + ) + key = self._entity_key(winning_snapshot) + self._local_entities[key] = winning_snapshot + self._server_entities[key] = winning_snapshot + + _logger.info( + "sync.conflict.resolved", + conflict_id=conflict_id, + resolution=resolution.value, + winner=target.winner, + ) + + return target + + # ------------------------------------------------------------------ + # Entity management helpers (for testing / local mode) + # ------------------------------------------------------------------ + + def register_local_entity(self, entity: SyncEntitySnapshot) -> None: + """Register an entity in the local cache. + + Args: + entity: The entity snapshot to register. + + Raises: + TypeError: If *entity* is not a ``SyncEntitySnapshot``. + """ + if not isinstance(entity, SyncEntitySnapshot): + raise TypeError("entity must be a SyncEntitySnapshot") + key = self._entity_key(entity) + self._local_entities[key] = entity + + def register_server_entity(self, entity: SyncEntitySnapshot) -> None: + """Register an entity in the server cache. + + Args: + entity: The entity snapshot to register. + + Raises: + TypeError: If *entity* is not a ``SyncEntitySnapshot``. + """ + if not isinstance(entity, SyncEntitySnapshot): + raise TypeError("entity must be a SyncEntitySnapshot") + key = self._entity_key(entity) + self._server_entities[key] = entity + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _entity_key(self, entity: SyncEntitySnapshot) -> str: + """Return a unique key for an entity snapshot.""" + return f"{entity.namespace}/{entity.entity_type.value}/{entity.entity_id}" + + def _get_server_entities( + self, + namespace: str, + entity_types: list[SyncEntityType], + since: str | None = None, + ) -> list[SyncEntitySnapshot]: + """Retrieve server entities matching the filter criteria.""" + result: list[SyncEntitySnapshot] = [] + for entity in self._server_entities.values(): + if entity.namespace != namespace: + continue + if entity_types and entity.entity_type not in entity_types: + continue + if since and entity.updated_at and entity.updated_at <= since: + continue + result.append(entity) + return result + + def _create_conflict( + self, + local_entity: SyncEntitySnapshot, + server_entity: SyncEntitySnapshot, + namespace: str, + ) -> SyncConflict: + """Create a sync conflict record.""" + return SyncConflict( + conflict_id=str(ULID()), + entity_id=local_entity.entity_id, + entity_type=local_entity.entity_type, + namespace=namespace, + local_snapshot=local_entity, + server_snapshot=server_entity, + local_clock=local_entity.vector_clock, + server_clock=server_entity.vector_clock, + ) + + def _resolve_conflict( + self, + conflict: SyncConflict, + resolution: ConflictResolution, + ) -> SyncEntitySnapshot | None: + """Apply automatic conflict resolution and return the winner. + + Returns ``None`` if manual resolution is required. + """ + if resolution == ConflictResolution.MANUAL: + return None + + if resolution == ConflictResolution.CLIENT_WINS: + conflict.winner = "local" + return conflict.local_snapshot + + if resolution == ConflictResolution.SERVER_WINS: + conflict.winner = "server" + return conflict.server_snapshot + + # Default: last_writer_wins + local_ts = conflict.local_snapshot.updated_at or "" + server_ts = conflict.server_snapshot.updated_at or "" + if local_ts >= server_ts: + conflict.winner = "local" + return conflict.local_snapshot + conflict.winner = "server" + return conflict.server_snapshot + + def _stamp_entity( + self, + entity: SyncEntitySnapshot, + namespace: str, + ) -> SyncEntitySnapshot: + """Return a copy of *entity* with updated clock and timestamp.""" + new_clock = entity.vector_clock.increment(self._node_id) + return SyncEntitySnapshot( + entity_id=entity.entity_id, + entity_type=entity.entity_type, + namespace=namespace, + name=entity.name, + version=entity.version + 1, + data=dict(entity.data), + vector_clock=new_clock, + updated_at=_iso_now(), + ) + + def _update_sync_state( + self, + namespace: str, + last_pull_at: str | None = None, + last_push_at: str | None = None, + conflicts: list[SyncConflict] | None = None, + ) -> SyncState: + """Update and return the sync state for a namespace.""" + state = self._sync_states.get(namespace) + now = _iso_now() + + if state is None: + state = SyncState( + namespace=namespace, + last_sync_at=now, + local_clock=VectorClock(), + server_clock=VectorClock(), + ) + + state.last_sync_at = now + if last_pull_at is not None: + state.last_pull_at = last_pull_at + if last_push_at is not None: + state.last_push_at = last_push_at + if conflicts is not None: + state.conflicts = conflicts + + state.local_clock = state.local_clock.increment(self._node_id) + self._sync_states[namespace] = state + + return state + + +__all__ = [ + "SyncService", +]