From f942f275038c55707ec930a27c6668012020d500 Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Tue, 25 Aug 2026 16:24:43 +0800 Subject: [PATCH 1/2] feat(runtime): expand orchestrated runtime storage --- ...260825_0049_runtime_orchestrated_expand.py | 254 ++++++++++ .../infrastructure/postgres/models.py | 135 +++++ ...st_runtime_orchestrated_expand_postgres.py | 468 ++++++++++++++++++ ...t_runtime_orchestrated_expand_migration.py | 116 +++++ 4 files changed, 973 insertions(+) create mode 100644 alembic/versions/20260825_0049_runtime_orchestrated_expand.py create mode 100644 tests/integration/test_runtime_orchestrated_expand_postgres.py create mode 100644 tests/test_runtime_orchestrated_expand_migration.py diff --git a/alembic/versions/20260825_0049_runtime_orchestrated_expand.py b/alembic/versions/20260825_0049_runtime_orchestrated_expand.py new file mode 100644 index 0000000..b864537 --- /dev/null +++ b/alembic/versions/20260825_0049_runtime_orchestrated_expand.py @@ -0,0 +1,254 @@ +"""Expand Runtime storage for orchestrated reader compatibility. + +This revision is deliberately expand-only. It adds the immutable evidence +tables and the lifecycle worker columns, but does not schedule, claim, or +otherwise write any new lifecycle values. +""" + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +revision = "20260825_0049" +down_revision = "20260821_0048" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "runtime_assignment_snapshots", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("tenant_id", sa.String(length=128), nullable=False), + sa.Column( + "runtime_execution_id", + sa.Uuid(), + sa.ForeignKey("runtime_executions.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("contract_name", sa.String(length=128), nullable=False), + sa.Column("contract_major", sa.Integer(), nullable=False), + sa.Column("assignment_id", sa.Uuid(), nullable=False), + sa.Column("assignment_digest", sa.String(length=64), nullable=False), + sa.Column("canonical_payload", postgresql.JSONB(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.CheckConstraint("contract_major >= 1", name="ck_runtime_assignment_snapshot_major"), + sa.CheckConstraint( + "assignment_digest ~ '^[0-9a-f]{64}$'", + name="ck_runtime_assignment_snapshot_digest", + ), + sa.CheckConstraint( + "octet_length(canonical_payload::text) <= 262144", + name="ck_runtime_assignment_snapshot_payload_size", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "runtime_execution_id", name="uq_runtime_assignment_snapshot_execution" + ), + ) + op.create_index( + "ix_runtime_assignment_snapshots_tenant_created", + "runtime_assignment_snapshots", + ["tenant_id", "created_at"], + ) + + op.create_table( + "runtime_handle_snapshots", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("tenant_id", sa.String(length=128), nullable=False), + sa.Column( + "runtime_execution_id", + sa.Uuid(), + sa.ForeignKey("runtime_executions.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("handle_digest", sa.String(length=64), nullable=False), + sa.Column("canonical_payload", postgresql.JSONB(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.CheckConstraint( + "handle_digest ~ '^[0-9a-f]{64}$'", name="ck_runtime_handle_snapshot_digest" + ), + sa.CheckConstraint( + "octet_length(canonical_payload::text) <= 65536", + name="ck_runtime_handle_snapshot_payload_size", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("runtime_execution_id", name="uq_runtime_handle_snapshot_execution"), + ) + op.create_index( + "ix_runtime_handle_snapshots_tenant_created", + "runtime_handle_snapshots", + ["tenant_id", "created_at"], + ) + + op.create_table( + "runtime_integrity_incidents", + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("tenant_id", sa.String(length=128), nullable=False), + sa.Column( + "runtime_execution_id", + sa.Uuid(), + sa.ForeignKey("runtime_executions.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("accepted_observation_id", sa.String(length=512), nullable=False), + sa.Column("accepted_observation_digest", sa.String(length=64), nullable=False), + sa.Column("accepted_phase", sa.String(length=32), nullable=False), + sa.Column("conflicting_observation_id", sa.String(length=512), nullable=False), + sa.Column("conflicting_observation_digest", sa.String(length=64), nullable=False), + sa.Column("conflicting_phase", sa.String(length=32), nullable=False), + sa.Column("status", sa.String(length=16), nullable=False), + sa.Column("reason", sa.String(length=4096), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.CheckConstraint( + "accepted_observation_digest ~ '^[0-9a-f]{64}$' AND " + "conflicting_observation_digest ~ '^[0-9a-f]{64}$'", + name="ck_runtime_integrity_incident_digests", + ), + sa.CheckConstraint( + "accepted_phase IN ('SUCCEEDED', 'FAILED', 'CANCELED', 'TIMED_OUT', 'LOST') AND " + "conflicting_phase IN ('SUCCEEDED', 'FAILED', 'CANCELED', 'TIMED_OUT', 'LOST')", + name="ck_runtime_integrity_incident_terminal_phases", + ), + sa.CheckConstraint( + "status IN ('OPEN', 'ACKNOWLEDGED', 'ESCALATED')", + name="ck_runtime_integrity_incident_status", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "tenant_id", + "runtime_execution_id", + "accepted_observation_digest", + "conflicting_observation_digest", + name="uq_runtime_integrity_incident_conflict", + ), + ) + op.create_index( + "ix_runtime_integrity_incidents_tenant_status", + "runtime_integrity_incidents", + ["tenant_id", "status", "created_at"], + ) + op.create_index( + "ix_runtime_integrity_incidents_execution_created", + "runtime_integrity_incidents", + ["runtime_execution_id", "created_at"], + ) + + op.add_column( + "runtime_lifecycle_operations", + sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"), + ) + op.add_column( + "runtime_lifecycle_operations", + sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True), + ) + op.add_column( + "runtime_lifecycle_operations", + sa.Column("claim_token", sa.Uuid(), nullable=True), + ) + op.add_column( + "runtime_lifecycle_operations", + sa.Column("claim_acquired_at", sa.DateTime(timezone=True), nullable=True), + ) + op.add_column( + "runtime_lifecycle_operations", + sa.Column("claim_expires_at", sa.DateTime(timezone=True), nullable=True), + ) + op.add_column( + "runtime_lifecycle_operations", + sa.Column("last_error_code", sa.String(length=128), nullable=True), + ) + op.create_check_constraint( + "ck_runtime_lifecycle_attempt_count", + "runtime_lifecycle_operations", + "attempt_count >= 0", + ) + op.create_check_constraint( + "ck_runtime_lifecycle_claim_triple", + "runtime_lifecycle_operations", + "(claim_token IS NULL AND claim_acquired_at IS NULL AND claim_expires_at IS NULL) OR " + "(claim_token IS NOT NULL AND claim_acquired_at IS NOT NULL AND " + "claim_expires_at IS NOT NULL)", + ) + op.create_check_constraint( + "ck_runtime_lifecycle_claim_expiry", + "runtime_lifecycle_operations", + "claim_expires_at IS NULL OR claim_expires_at > claim_acquired_at", + ) + op.create_index( + "ix_runtime_lifecycle_due", + "runtime_lifecycle_operations", + ["status", "next_attempt_at", "deadline"], + ) + + +def _refuse_if_written() -> None: + """Refuse loss of any A4.2a writer marker before changing the schema.""" + + bind = op.get_bind() + row = bind.execute( + sa.text( + "SELECT 1 FROM runtime_lifecycle_operations WHERE " + "attempt_count <> 0 OR next_attempt_at IS NOT NULL OR " + "claim_token IS NOT NULL OR claim_acquired_at IS NOT NULL OR " + "claim_expires_at IS NOT NULL OR last_error_code IS NOT NULL LIMIT 1" + ) + ).first() + if row is not None: + raise RuntimeError( + "Cannot downgrade 0049: Runtime lifecycle writer markers exist; " + "drain A4.2 lifecycle operations before retrying" + ) + for table in ( + "runtime_assignment_snapshots", + "runtime_handle_snapshots", + "runtime_integrity_incidents", + ): + row = bind.execute(sa.text(f"SELECT 1 FROM {table} LIMIT 1")).first() + if row is not None: + raise RuntimeError(f"Cannot downgrade 0049: {table} contains rows; refusing data loss") + + +def downgrade() -> None: + _refuse_if_written() + + op.drop_index("ix_runtime_lifecycle_due", table_name="runtime_lifecycle_operations") + op.drop_constraint( + "ck_runtime_lifecycle_claim_expiry", "runtime_lifecycle_operations", type_="check" + ) + op.drop_constraint( + "ck_runtime_lifecycle_claim_triple", "runtime_lifecycle_operations", type_="check" + ) + op.drop_constraint( + "ck_runtime_lifecycle_attempt_count", "runtime_lifecycle_operations", type_="check" + ) + for column in ( + "last_error_code", + "claim_expires_at", + "claim_acquired_at", + "claim_token", + "next_attempt_at", + "attempt_count", + ): + op.drop_column("runtime_lifecycle_operations", column) + + op.drop_index( + "ix_runtime_integrity_incidents_execution_created", + table_name="runtime_integrity_incidents", + ) + op.drop_index( + "ix_runtime_integrity_incidents_tenant_status", + table_name="runtime_integrity_incidents", + ) + op.drop_table("runtime_integrity_incidents") + op.drop_index( + "ix_runtime_handle_snapshots_tenant_created", table_name="runtime_handle_snapshots" + ) + op.drop_table("runtime_handle_snapshots") + op.drop_index( + "ix_runtime_assignment_snapshots_tenant_created", + table_name="runtime_assignment_snapshots", + ) + op.drop_table("runtime_assignment_snapshots") diff --git a/src/agentmesh/infrastructure/postgres/models.py b/src/agentmesh/infrastructure/postgres/models.py index 2783a1a..d333612 100644 --- a/src/agentmesh/infrastructure/postgres/models.py +++ b/src/agentmesh/infrastructure/postgres/models.py @@ -1749,6 +1749,18 @@ class RuntimeLifecycleOperationRecord(Base): status: Mapped[str] = mapped_column(String(16), nullable=False) deadline: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) receipt_summary: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) + # A4.2a.0 expand columns. They are intentionally storage-only until the + # lifecycle writer is activated in a later slice. + attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0")) + next_attempt_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + claim_token: Mapped[UUID | None] = mapped_column(Uuid, nullable=True) + claim_acquired_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + claim_expires_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + last_error_code: Mapped[str | None] = mapped_column(String(128), nullable=True) version: Mapped[int] = mapped_column(Integer, nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) @@ -1767,10 +1779,133 @@ class RuntimeLifecycleOperationRecord(Base): "intent_digest ~ '^[0-9a-f]{64}$'", name="ck_runtime_lifecycle_digest", ), + CheckConstraint("attempt_count >= 0", name="ck_runtime_lifecycle_attempt_count"), + CheckConstraint( + "(claim_token IS NULL AND claim_acquired_at IS NULL AND claim_expires_at IS NULL) OR " + "(claim_token IS NOT NULL AND claim_acquired_at IS NOT NULL AND " + "claim_expires_at IS NOT NULL)", + name="ck_runtime_lifecycle_claim_triple", + ), + CheckConstraint( + "claim_expires_at IS NULL OR claim_expires_at > claim_acquired_at", + name="ck_runtime_lifecycle_claim_expiry", + ), UniqueConstraint( "runtime_execution_id", "operation_id", name="uq_runtime_lifecycle_operation" ), Index("ix_runtime_lifecycle_tenant_status", "tenant_id", "status", "deadline"), + Index("ix_runtime_lifecycle_due", "status", "next_attempt_at", "deadline"), + ) + + +class RuntimeAssignmentSnapshotRecord(Base): + """Immutable, bounded canonical Assignment evidence for one execution.""" + + __tablename__ = "runtime_assignment_snapshots" + + id: Mapped[UUID] = mapped_column(Uuid, primary_key=True) + tenant_id: Mapped[str] = mapped_column(String(128), nullable=False) + runtime_execution_id: Mapped[UUID] = mapped_column( + Uuid, ForeignKey("runtime_executions.id", ondelete="CASCADE"), nullable=False + ) + contract_name: Mapped[str] = mapped_column(String(128), nullable=False) + contract_major: Mapped[int] = mapped_column(Integer, nullable=False) + assignment_id: Mapped[UUID] = mapped_column(Uuid, nullable=False) + assignment_digest: Mapped[str] = mapped_column(String(64), nullable=False) + canonical_payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + __table_args__ = ( + CheckConstraint("contract_major >= 1", name="ck_runtime_assignment_snapshot_major"), + CheckConstraint( + "assignment_digest ~ '^[0-9a-f]{64}$'", + name="ck_runtime_assignment_snapshot_digest", + ), + CheckConstraint( + "octet_length(canonical_payload::text) <= 262144", + name="ck_runtime_assignment_snapshot_payload_size", + ), + UniqueConstraint("runtime_execution_id", name="uq_runtime_assignment_snapshot_execution"), + Index("ix_runtime_assignment_snapshots_tenant_created", "tenant_id", "created_at"), + ) + + +class RuntimeHandleSnapshotRecord(Base): + """Immutable, bounded canonical execution handle for lifecycle replay.""" + + __tablename__ = "runtime_handle_snapshots" + + id: Mapped[UUID] = mapped_column(Uuid, primary_key=True) + tenant_id: Mapped[str] = mapped_column(String(128), nullable=False) + runtime_execution_id: Mapped[UUID] = mapped_column( + Uuid, ForeignKey("runtime_executions.id", ondelete="CASCADE"), nullable=False + ) + handle_digest: Mapped[str] = mapped_column(String(64), nullable=False) + canonical_payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + __table_args__ = ( + CheckConstraint( + "handle_digest ~ '^[0-9a-f]{64}$'", name="ck_runtime_handle_snapshot_digest" + ), + CheckConstraint( + "octet_length(canonical_payload::text) <= 65536", + name="ck_runtime_handle_snapshot_payload_size", + ), + UniqueConstraint("runtime_execution_id", name="uq_runtime_handle_snapshot_execution"), + Index("ix_runtime_handle_snapshots_tenant_created", "tenant_id", "created_at"), + ) + + +class RuntimeIntegrityIncidentRecord(Base): + """Safe projection of a late terminal conflict; raw provider bodies are absent.""" + + __tablename__ = "runtime_integrity_incidents" + + id: Mapped[UUID] = mapped_column(Uuid, primary_key=True) + tenant_id: Mapped[str] = mapped_column(String(128), nullable=False) + runtime_execution_id: Mapped[UUID] = mapped_column( + Uuid, ForeignKey("runtime_executions.id", ondelete="CASCADE"), nullable=False + ) + accepted_observation_id: Mapped[str] = mapped_column(String(512), nullable=False) + accepted_observation_digest: Mapped[str] = mapped_column(String(64), nullable=False) + accepted_phase: Mapped[str] = mapped_column(String(32), nullable=False) + conflicting_observation_id: Mapped[str] = mapped_column(String(512), nullable=False) + conflicting_observation_digest: Mapped[str] = mapped_column(String(64), nullable=False) + conflicting_phase: Mapped[str] = mapped_column(String(32), nullable=False) + status: Mapped[str] = mapped_column(String(16), nullable=False) + reason: Mapped[str] = mapped_column(String(4096), nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + __table_args__ = ( + CheckConstraint( + "accepted_observation_digest ~ '^[0-9a-f]{64}$' AND " + "conflicting_observation_digest ~ '^[0-9a-f]{64}$'", + name="ck_runtime_integrity_incident_digests", + ), + CheckConstraint( + "accepted_phase IN ('SUCCEEDED', 'FAILED', 'CANCELED', 'TIMED_OUT', 'LOST') AND " + "conflicting_phase IN ('SUCCEEDED', 'FAILED', 'CANCELED', 'TIMED_OUT', 'LOST')", + name="ck_runtime_integrity_incident_terminal_phases", + ), + CheckConstraint( + "status IN ('OPEN', 'ACKNOWLEDGED', 'ESCALATED')", + name="ck_runtime_integrity_incident_status", + ), + UniqueConstraint( + "tenant_id", + "runtime_execution_id", + "accepted_observation_digest", + "conflicting_observation_digest", + name="uq_runtime_integrity_incident_conflict", + ), + Index("ix_runtime_integrity_incidents_tenant_status", "tenant_id", "status", "created_at"), + Index( + "ix_runtime_integrity_incidents_execution_created", + "runtime_execution_id", + "created_at", + ), ) diff --git a/tests/integration/test_runtime_orchestrated_expand_postgres.py b/tests/integration/test_runtime_orchestrated_expand_postgres.py new file mode 100644 index 0000000..56cb985 --- /dev/null +++ b/tests/integration/test_runtime_orchestrated_expand_postgres.py @@ -0,0 +1,468 @@ +"""Real PostgreSQL checks for the A4.2a.0 expand-only schema floor.""" + +from __future__ import annotations + +import os +from datetime import datetime, timedelta, timezone +from uuid import uuid4 + +import pytest +from alembic.config import Config +from sqlalchemy import create_engine, inspect, text +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from agentmesh.config import get_settings +from agentmesh.infrastructure.postgres.models import ( + PrincipalRecord, + RuntimeExecutionRecord, + RuntimeRegistrationRecord, + RuntimeVersionRecord, + TaskRecord, + TaskRunRecord, +) +from alembic import command + +pytestmark = [ + pytest.mark.postgres, + pytest.mark.skipif( + os.getenv("AGENTMESH_RUN_POSTGRES_TESTS") != "1", + reason="set AGENTMESH_RUN_POSTGRES_TESTS=1 to run PostgreSQL migration tests", + ), +] + + +def _config() -> Config: + return Config("alembic.ini") + + +def _head_to_0048() -> None: + command.upgrade(_config(), "head") + command.downgrade(_config(), "20260821_0048") + + +def _create_runtime_execution(engine): + from agentmesh.domain.runtime_execution import RuntimeExecution + + now = datetime.now(timezone.utc) + tenant = f"runtime-expand-{uuid4().hex}" + principal_id, task_id, run_id = uuid4(), uuid4(), uuid4() + registration_id, version_id = uuid4(), uuid4() + execution_id = uuid4() + with Session(engine) as session, session.begin(): + session.add( + PrincipalRecord( + id=principal_id, + tenant_id=tenant, + principal_type="SERVICE", + status="ACTIVE", + display_name="expand migration fixture", + created_at=now, + updated_at=now, + revision=1, + ) + ) + session.add( + TaskRecord( + id=task_id, + tenant_id=tenant, + project_id="migration", + objective="migration fixture", + input={}, + status="READY", + current_run_id=None, + output=None, + error=None, + execution_mode="DIRECT", + acceptance_criteria=[], + max_revisions=0, + revision_count=0, + review_deadline=None, + candidate_output=None, + latest_review=None, + plan_version=None, + plan_digest=None, + max_concurrency=1, + budget=None, + settled_tokens=0, + reserved_tokens=0, + settled_cost_micros=0, + reserved_cost_micros=0, + budget_exhausted_reason=None, + budget_revision=0, + version=1, + created_at=now, + updated_at=now, + ) + ) + session.add_all( + [ + RuntimeRegistrationRecord( + id=registration_id, + tenant_id=None, + name=f"expand-{uuid4().hex}", + owner_principal_id=principal_id, + visibility="platform", + status="ACTIVE", + default_version_id=version_id, + version=1, + created_at=now, + updated_at=now, + ), + RuntimeVersionRecord( + id=version_id, + runtime_id=registration_id, + api_version=1, + adapter_kind="python-in-process", + artifact_digest="a" * 64, + configuration_digest="b" * 64, + descriptor={"limits": {"max_assignment_bytes": 262144}}, + trust_profile="built_in", + compatibility={}, + status="PUBLISHED", + created_at=now, + published_at=now, + revoked_at=None, + ), + TaskRunRecord( + id=run_id, + task_id=task_id, + thread_id=str(run_id), + agent_id="migration-fixture", + agent_version_id=None, + agent_version_digest=None, + runtime_version_id=version_id, + runtime_execution_id=None, + runtime_execution_intent_id=None, + runtime_authority="legacy", + comparison_mode="off", + role="EXECUTOR", + revision_number=0, + subtask_id=None, + status="QUEUED", + output=None, + error=None, + queued_at=now, + started_at=None, + completed_at=None, + pause_requested_at=None, + paused_at=None, + resumed_at=None, + paused_from_status=None, + ), + ] + ) + session.flush() + execution = RuntimeExecution.prepare( + execution_id=execution_id, + tenant_id=tenant, + run_id=run_id, + runtime_version_id=version_id, + assignment_id=uuid4(), + assignment_digest="c" * 64, + dispatch_key=f"migration:{uuid4()}", + dispatch_digest="d" * 64, + now=now, + ) + session.add( + RuntimeExecutionRecord( + **{ + "id": execution.id, + "tenant_id": execution.tenant_id, + "run_id": execution.run_id, + "runtime_version_id": execution.runtime_version_id, + "assignment_id": execution.assignment_id, + "assignment_digest": execution.assignment_digest, + "dispatch_key": execution.dispatch_key, + "dispatch_digest": execution.dispatch_digest, + "provider_execution_ref": None, + "provider_generation": None, + "phase": execution.phase.value, + "current_owner_attempt_id": None, + "current_fencing_token": None, + "provider_sequence": None, + "checkpoint_ref": None, + "workspace_ref": None, + "version": 1, + "created_at": now, + "updated_at": now, + "terminal_at": None, + } + ) + ) + return execution + + +def _delete_runtime_execution(engine, execution) -> None: + with engine.begin() as connection: + task_id = connection.scalar( + text("SELECT task_id FROM task_runs WHERE id = :id"), {"id": execution.run_id} + ) + connection.execute( + text("DELETE FROM runtime_lifecycle_operations WHERE runtime_execution_id = :id"), + {"id": execution.id}, + ) + for table in ( + "runtime_assignment_snapshots", + "runtime_handle_snapshots", + "runtime_integrity_incidents", + ): + if inspect(connection).has_table(table): + connection.execute( + text(f"DELETE FROM {table} WHERE runtime_execution_id = :id"), + {"id": execution.id}, + ) + connection.execute( + text("DELETE FROM runtime_executions WHERE id = :id"), {"id": execution.id} + ) + if task_id is not None: + connection.execute(text("DELETE FROM tasks WHERE id = :id"), {"id": task_id}) + connection.execute( + text( + "UPDATE runtime_registrations SET default_version_id = NULL " + "WHERE name LIKE 'expand-%'" + ) + ) + connection.execute( + text( + "DELETE FROM runtime_versions WHERE runtime_id IN " + "(SELECT id FROM runtime_registrations WHERE name LIKE 'expand-%')" + ) + ) + connection.execute(text("DELETE FROM runtime_registrations WHERE name LIKE 'expand-%'")) + connection.execute( + text( + "DELETE FROM principals WHERE id NOT IN " + "(SELECT owner_principal_id FROM runtime_registrations) " + "AND tenant_id LIKE 'runtime-expand-%'" + ) + ) + + +def _insert_lifecycle(engine, execution, **extra) -> str: + operation_id = f"migration-test:{uuid4()}" + now = datetime.now(timezone.utc) + values = { + "id": uuid4(), + "tenant_id": execution.tenant_id, + "runtime_execution_id": execution.id, + "operation_id": operation_id, + "operation": "cancel", + "intent_digest": "a" * 64, + "status": "REQUESTED", + "deadline": now + timedelta(hours=1), + "receipt_summary": None, + "version": 1, + "created_at": now, + "updated_at": now, + **extra, + } + columns = ", ".join(values) + binds = ", ".join(f":{key}" for key in values) + with engine.begin() as connection: + connection.execute( + text(f"INSERT INTO runtime_lifecycle_operations ({columns}) VALUES ({binds})"), + values, + ) + return operation_id + + +def test_upgrade_backfills_zero_and_default_only_downgrade() -> None: + engine = create_engine(get_settings().database_url) + execution = None + try: + _head_to_0048() + execution = _create_runtime_execution(engine) + _insert_lifecycle(engine, execution) + command.upgrade(_config(), "head") + with engine.connect() as connection: + assert connection.scalar( + text( + "SELECT column_default FROM information_schema.columns WHERE " + "table_name='runtime_lifecycle_operations' AND column_name='attempt_count'" + ) + ) + assert ( + connection.scalar( + text( + "SELECT count(*) FROM runtime_lifecycle_operations " + "WHERE attempt_count <> 0 OR next_attempt_at IS NOT NULL OR " + "claim_token IS NOT NULL OR claim_acquired_at IS NOT NULL OR " + "claim_expires_at IS NOT NULL OR last_error_code IS NOT NULL" + ) + ) + == 0 + ) + command.downgrade(_config(), "20260821_0048") + with engine.connect() as connection: + columns = { + row.column_name + for row in connection.execute( + text( + "SELECT column_name FROM information_schema.columns " + "WHERE table_name='runtime_lifecycle_operations'" + ) + ) + } + assert "attempt_count" not in columns + assert not inspect(connection).has_table("runtime_assignment_snapshots") + assert not inspect(connection).has_table("runtime_handle_snapshots") + assert not inspect(connection).has_table("runtime_integrity_incidents") + finally: + if execution is not None: + _delete_runtime_execution(engine, execution) + command.upgrade(_config(), "head") + engine.dispose() + + +@pytest.mark.parametrize( + "marker,values", + [ + ("attempt_count", {"attempt_count": 1}), + ("next_attempt_at", {"next_attempt_at": datetime.now(timezone.utc)}), + ( + "claim_token", + { + "claim_token": uuid4(), + "claim_acquired_at": datetime.now(timezone.utc), + "claim_expires_at": datetime.now(timezone.utc) + timedelta(minutes=1), + }, + ), + ( + "claim_acquired_at", + { + "claim_acquired_at": datetime.now(timezone.utc), + "claim_token": uuid4(), + "claim_expires_at": datetime.now(timezone.utc) + timedelta(minutes=1), + }, + ), + ( + "claim_expires_at", + { + "claim_expires_at": datetime.now(timezone.utc), + "claim_token": uuid4(), + "claim_acquired_at": datetime.now(timezone.utc) - timedelta(minutes=1), + }, + ), + ("last_error_code", {"last_error_code": "runtime.transport"}), + ], +) +def test_downgrade_refuses_each_lifecycle_writer_marker( + marker: str, values: dict[str, object] +) -> None: + engine = create_engine(get_settings().database_url) + try: + _head_to_0048() + command.upgrade(_config(), "head") + execution = _create_runtime_execution(engine) + _insert_lifecycle(engine, execution, **values) + with pytest.raises(RuntimeError, match="writer markers"): + command.downgrade(_config(), "20260821_0048") + with engine.begin() as connection: + connection.execute( + text( + "DELETE FROM runtime_lifecycle_operations " + "WHERE operation_id LIKE 'migration-test:%'" + ) + ) + _delete_runtime_execution(engine, execution) + command.downgrade(_config(), "20260821_0048") + finally: + if "execution" in locals(): + _delete_runtime_execution(engine, execution) + command.upgrade(_config(), "head") + engine.dispose() + + +def test_postgres_enforces_claim_triple_and_strict_expiry() -> None: + engine = create_engine(get_settings().database_url) + try: + _head_to_0048() + command.upgrade(_config(), "head") + execution = _create_runtime_execution(engine) + now = datetime.now(timezone.utc) + with pytest.raises(IntegrityError): + _insert_lifecycle(engine, execution, claim_token=uuid4()) + with pytest.raises(IntegrityError): + _insert_lifecycle( + engine, + execution, + claim_token=uuid4(), + claim_acquired_at=now, + claim_expires_at=now, + ) + finally: + with engine.begin() as connection: + connection.execute( + text( + "DELETE FROM runtime_lifecycle_operations " + "WHERE operation_id LIKE 'migration-test:%'" + ) + ) + _delete_runtime_execution(engine, execution) + command.upgrade(_config(), "head") + engine.dispose() + + +@pytest.mark.parametrize( + "table", + [ + "runtime_assignment_snapshots", + "runtime_handle_snapshots", + "runtime_integrity_incidents", + ], +) +def test_postgres_downgrade_refuses_each_new_table_row(table: str) -> None: + engine = create_engine(get_settings().database_url) + try: + _head_to_0048() + command.upgrade(_config(), "head") + execution = _create_runtime_execution(engine) + now = datetime.now(timezone.utc) + values: dict[str, object] = { + "id": uuid4(), + "tenant_id": execution.tenant_id, + "runtime_execution_id": execution.id, + "created_at": now, + } + if table == "runtime_assignment_snapshots": + values.update( + { + "contract_name": "agentmesh.runtime-assignment", + "contract_major": 1, + "assignment_id": uuid4(), + "assignment_digest": "a" * 64, + "canonical_payload": '{"bounded":true}', + } + ) + elif table == "runtime_handle_snapshots": + values.update({"handle_digest": "b" * 64, "canonical_payload": '{"handle":true}'}) + else: + values.update( + { + "accepted_observation_id": "accepted", + "accepted_observation_digest": "c" * 64, + "accepted_phase": "SUCCEEDED", + "conflicting_observation_id": "conflict", + "conflicting_observation_digest": "d" * 64, + "conflicting_phase": "FAILED", + "status": "OPEN", + "reason": "migration test", + "updated_at": now, + } + ) + columns = ", ".join(values) + binds = ", ".join(f":{key}" for key in values) + with engine.begin() as connection: + connection.execute(text(f"INSERT INTO {table} ({columns}) VALUES ({binds})"), values) + with pytest.raises(RuntimeError, match=f"{table} contains rows"): + command.downgrade(_config(), "20260821_0048") + with engine.begin() as connection: + connection.execute(text(f"DELETE FROM {table} WHERE id = :id"), {"id": values["id"]}) + _delete_runtime_execution(engine, execution) + command.downgrade(_config(), "20260821_0048") + finally: + if "execution" in locals(): + _delete_runtime_execution(engine, execution) + command.upgrade(_config(), "head") + engine.dispose() diff --git a/tests/test_runtime_orchestrated_expand_migration.py b/tests/test_runtime_orchestrated_expand_migration.py new file mode 100644 index 0000000..37dbba1 --- /dev/null +++ b/tests/test_runtime_orchestrated_expand_migration.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path +from types import ModuleType +from unittest.mock import Mock + +import pytest + + +def _load_migration() -> ModuleType: + path = ( + Path(__file__).parents[1] + / "alembic" + / "versions" + / "20260825_0049_runtime_orchestrated_expand.py" + ) + spec = importlib.util.spec_from_file_location("runtime_orchestrated_expand_0049", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_revision_follows_0048_and_upgrade_is_expand_only(monkeypatch: pytest.MonkeyPatch) -> None: + migration = _load_migration() + assert migration.down_revision == "20260821_0048" + + calls: list[tuple[str, tuple[object, ...]]] = [] + for name in ( + "create_table", + "create_index", + "add_column", + "create_check_constraint", + ): + monkeypatch.setattr( + migration.op, + name, + lambda *args, _name=name, **kwargs: calls.append((_name, args)), + ) + + migration.upgrade() + + tables = [args[0] for name, args in calls if name == "create_table"] + assert tables == [ + "runtime_assignment_snapshots", + "runtime_handle_snapshots", + "runtime_integrity_incidents", + ] + columns = [args[1].name for name, args in calls if name == "add_column"] + assert columns == [ + "attempt_count", + "next_attempt_at", + "claim_token", + "claim_acquired_at", + "claim_expires_at", + "last_error_code", + ] + assert not any(name in {"execute", "update"} for name, _ in calls) + + +def test_default_only_downgrade_drops_expand_objects(monkeypatch: pytest.MonkeyPatch) -> None: + migration = _load_migration() + bind = Mock() + bind.execute.return_value.first.return_value = None + monkeypatch.setattr(migration.op, "get_bind", lambda: bind) + calls: list[tuple[str, str]] = [] + for name in ("drop_index", "drop_constraint", "drop_column", "drop_table"): + monkeypatch.setattr( + migration.op, + name, + lambda *args, _name=name, **kwargs: calls.append((_name, str(args))), + ) + + migration.downgrade() + + assert [name for name, _ in calls].count("drop_table") == 3 + assert [name for name, _ in calls].count("drop_column") == 6 + + +def test_downgrade_refuses_lifecycle_writer_markers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + migration = _load_migration() + bind = Mock() + bind.execute.return_value.first.return_value = (1,) + monkeypatch.setattr(migration.op, "get_bind", lambda: bind) + with pytest.raises(RuntimeError, match="writer markers"): + migration.downgrade() + + +@pytest.mark.parametrize( + "table", + [ + "runtime_assignment_snapshots", + "runtime_handle_snapshots", + "runtime_integrity_incidents", + ], +) +def test_downgrade_refuses_rows_without_cross_tenant_cleanup( + monkeypatch: pytest.MonkeyPatch, table: str +) -> None: + migration = _load_migration() + bind = Mock() + results = iter( + [ + None, + (1,) if table == "runtime_assignment_snapshots" else None, + (1,) if table == "runtime_handle_snapshots" else None, + (1,) if table == "runtime_integrity_incidents" else None, + ] + ) + bind.execute.return_value.first.side_effect = lambda: next(results) + monkeypatch.setattr(migration.op, "get_bind", lambda: bind) + with pytest.raises(RuntimeError, match=f"{table} contains rows"): + migration.downgrade() From c0984a0dd6580c3dce42e05c3e2e491dd5ca34ba Mon Sep 17 00:00:00 2001 From: 0YHR0 <97868579@qq.com> Date: Tue, 25 Aug 2026 16:58:16 +0800 Subject: [PATCH 2/2] feat(runtime): add immutable snapshot readers --- docs/implementation-status.md | 18 +- docs/roadmap.md | 3 +- src/agentmesh/application/ports.py | 24 ++ .../application/runtime_snapshots.py | 171 ++++++++ src/agentmesh/domain/runtime_execution.py | 62 +++ .../postgres/runtime_repositories.py | 378 +++++++++++++++++- ...st_runtime_snapshot_repository_postgres.py | 224 +++++++++++ tests/test_runtime_snapshot_domain.py | 179 +++++++++ 8 files changed, 1054 insertions(+), 5 deletions(-) create mode 100644 src/agentmesh/application/runtime_snapshots.py create mode 100644 tests/integration/test_runtime_snapshot_repository_postgres.py create mode 100644 tests/test_runtime_snapshot_domain.py diff --git a/docs/implementation-status.md b/docs/implementation-status.md index 371da12..b004ca9 100644 --- a/docs/implementation-status.md +++ b/docs/implementation-status.md @@ -1,7 +1,7 @@ # Implementation status Status: Alpha baseline -Last updated: 2026-08-23 +Last updated: 2026-08-25 This page records what the repository actually implements. The formal L2 documents describe the target architecture; an implemented vertical slice does not imply that every capability in its @@ -175,6 +175,22 @@ A4.1b.2b evidence-driven outcome reconciliation (writer slice): not to a pre-0048 reader. Reviewed/coordinated authority and production durable reattach remain outside this slice. +A4.2a.0 orchestrated Runtime expand compatibility: + +- Migration 0049 adds immutable Assignment/handle snapshot storage, late-terminal integrity + incidents, and nullable/default lifecycle due-worker columns. It does not schedule lifecycle + work, enable a cutover gate, or change Task/Run authority. +- Framework-neutral persistence projections validate complete Runtime SDK Assignment/handle + contracts, JCS bounds, digest and execution-chain identity. Tenant-scoped repositories provide + exact replay and conflict primitives without exposing payloads through the public Runtime DTO. +- Clean/default-only downgrade to 0048 is supported. Any snapshot/incident row or non-default + lifecycle writer marker makes downgrade refuse without deleting data or performing cross-tenant + cleanup. +- Local evidence includes 48 focused contract/architecture/migration tests, 11 real PostgreSQL + migration round-trip tests, and a real PostgreSQL repository round-trip covering tenant scope, + replay, conflict, and full Task/Run/Runtime/Assignment/handle binding. A4.2a.1 writer behavior and + reviewed/coordinated admission remain disabled and unimplemented in this slice. + ## Current runnable baseline AgentMesh currently provides durable direct, independently reviewed, and coordinated Subtask DAG diff --git a/docs/roadmap.md b/docs/roadmap.md index 1f135e5..23587e1 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -174,7 +174,8 @@ Exit signal：用户可从模板创建公司、绑定真实 Agent，在不伪造 - [x] A4.1b.2a reconciliation reader/schema compatibility（0048 expand-only；不含 writer/API） - [x] A4.1b.2b evidence-driven privileged outcome reconciliation（canonical observation、原子收敛、 幂等/并发保护、无 provider redispatch；默认关闭） -- [ ] A4.2a shared terminal/work-item/outcome semantics（关闭 #154；不开放新 admission） +- [x] A4.2a.0 expand compatibility（0049 snapshots/incidents/lifecycle due-reader 基础；无 writer） +- [ ] A4.2a.1 shared terminal/work-item/outcome semantics（关闭 #154；不开放新 admission） - [ ] A4.2b reviewed managed authority（cohort inheritance、review/revision reconciliation） - [ ] A4.2c coordinated managed authority（Subtask reader expand、并行 reconciliation barrier、 sibling lifecycle safety） diff --git a/src/agentmesh/application/ports.py b/src/agentmesh/application/ports.py index a17b0cf..669e609 100644 --- a/src/agentmesh/application/ports.py +++ b/src/agentmesh/application/ports.py @@ -7,6 +7,10 @@ from typing import Any, Protocol from uuid import UUID +from agentmesh.application.runtime_snapshots import ( + RuntimeAssignmentSnapshot, + RuntimeHandleSnapshot, +) from agentmesh.domain.a2a_delegation import RemoteTaskCorrelation from agentmesh.domain.a2a_registry import A2APeer, AgentCardSnapshot from agentmesh.domain.activity import ReplayBookmark @@ -87,6 +91,7 @@ from agentmesh.domain.runtime_execution import ( ReattachEvidence, RuntimeExecution, + RuntimeIntegrityIncident, RuntimeLifecycleIntent, RuntimeLifecycleStatus, RuntimeObservationEvidence, @@ -226,6 +231,25 @@ def update_lifecycle_status( status: RuntimeLifecycleStatus, now: datetime, ) -> None: ... + def get_assignment_snapshot( + self, execution_id: UUID, *, tenant_id: str + ) -> RuntimeAssignmentSnapshot | None: ... + def add_assignment_snapshot( + self, value: RuntimeAssignmentSnapshot + ) -> RuntimeAssignmentSnapshot: ... + def get_handle_snapshot( + self, execution_id: UUID, *, tenant_id: str + ) -> RuntimeHandleSnapshot | None: ... + def add_handle_snapshot(self, value: RuntimeHandleSnapshot) -> RuntimeHandleSnapshot: ... + def get_integrity_incident( + self, incident_id: UUID, *, tenant_id: str + ) -> RuntimeIntegrityIncident | None: ... + def list_integrity_incidents( + self, execution_id: UUID, *, tenant_id: str, limit: int, offset: int + ) -> list[RuntimeIntegrityIncident]: ... + def add_integrity_incident( + self, value: RuntimeIntegrityIncident + ) -> RuntimeIntegrityIncident: ... class RuntimeComparisonRepository(Protocol): diff --git a/src/agentmesh/application/runtime_snapshots.py b/src/agentmesh/application/runtime_snapshots.py new file mode 100644 index 0000000..8c1d1c3 --- /dev/null +++ b/src/agentmesh/application/runtime_snapshots.py @@ -0,0 +1,171 @@ +"""Bounded persistence projections for orchestrated runtime snapshots.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from datetime import datetime +from types import MappingProxyType +from typing import Any +from uuid import UUID + +from agentmesh.domain.errors import InvalidTaskInput +from agentmesh.runtime_sdk.assignment import RuntimeAssignment, RuntimeExecutionHandle +from agentmesh.runtime_sdk.canonical import ( + CanonicalizationError, + canonical_digest, + canonical_json_bytes, + decode_json, +) +from agentmesh.runtime_sdk.common import RuntimeContractError + +_DIGEST = re.compile(r"^[0-9a-f]{64}$") + + +@dataclass(frozen=True) +class RuntimeAssignmentSnapshot: + """Bounded immutable canonical Assignment projection kept off public DTOs.""" + + id: UUID + tenant_id: str + runtime_execution_id: UUID + contract_name: str + contract_major: int + assignment_id: UUID + assignment_digest: str + canonical_payload: MappingProxyType + created_at: datetime + + def __post_init__(self) -> None: + if any( + type(value) is not UUID + for value in (self.id, self.runtime_execution_id, self.assignment_id) + ): + raise InvalidTaskInput("Runtime Assignment snapshot identity is invalid") + if ( + type(self.tenant_id) is not str + or not self.tenant_id.strip() + or len(self.tenant_id) > 128 + or type(self.contract_name) is not str + or not self.contract_name.strip() + or len(self.contract_name) > 128 + or type(self.contract_major) is not int + or self.contract_major < 1 + or type(self.assignment_digest) is not str + or _DIGEST.fullmatch(self.assignment_digest) is None + or type(self.created_at) is not datetime + or self.created_at.tzinfo is None + ): + raise InvalidTaskInput("Runtime Assignment snapshot is invalid") + payload = snapshot_payload(self.canonical_payload, limit=262_144) + assignment = parse_assignment_payload(payload) + if ( + self.contract_name != assignment.schema_name + or self.contract_major != assignment.schema_version + or self.tenant_id != assignment.tenant_id + or str(self.assignment_id) != assignment.assignment_id + or self.assignment_digest != assignment.assignment_digest + ): + raise InvalidTaskInput("Runtime Assignment snapshot identity does not match payload") + object.__setattr__(self, "canonical_payload", payload) + + +@dataclass(frozen=True) +class RuntimeHandleSnapshot: + """Bounded immutable canonical execution handle projection.""" + + id: UUID + tenant_id: str + runtime_execution_id: UUID + handle_digest: str + canonical_payload: MappingProxyType + created_at: datetime + + def __post_init__(self) -> None: + if any(type(value) is not UUID for value in (self.id, self.runtime_execution_id)): + raise InvalidTaskInput("Runtime handle snapshot identity is invalid") + if ( + type(self.tenant_id) is not str + or not self.tenant_id.strip() + or len(self.tenant_id) > 128 + or type(self.handle_digest) is not str + or _DIGEST.fullmatch(self.handle_digest) is None + or type(self.created_at) is not datetime + or self.created_at.tzinfo is None + ): + raise InvalidTaskInput("Runtime handle snapshot is invalid") + payload = snapshot_payload(self.canonical_payload, limit=65_536) + handle = parse_handle_payload(payload) + if ( + str(self.runtime_execution_id) != handle.runtime_execution_id + or self.handle_digest != canonical_digest(handle.to_dict()) + ): + raise InvalidTaskInput("Runtime handle snapshot identity does not match payload") + object.__setattr__(self, "canonical_payload", payload) + + +def snapshot_payload(value: Any, *, limit: int) -> MappingProxyType: + """Freeze and validate a JCS-compatible JSON object at a byte boundary.""" + if type(value) not in (dict, MappingProxyType): + raise InvalidTaskInput("Runtime snapshot payload must be an object") + + def check_shape(item: Any, depth: int = 0) -> None: + if depth > 32: + raise InvalidTaskInput("Runtime snapshot payload is too deep") + if type(item) in (dict, MappingProxyType): + for key, child in item.items(): + if type(key) is not str: + raise InvalidTaskInput("Runtime snapshot object key is invalid") + check_shape(child, depth + 1) + return + if type(item) in (list, tuple): + for child in item: + check_shape(child, depth + 1) + + check_shape(value) + normalized = _thaw_json(value) + try: + encoded = canonical_json_bytes(normalized) + normalized = decode_json(encoded) + except (CanonicalizationError, TypeError, ValueError) as exc: + raise InvalidTaskInput("Runtime snapshot payload is not canonical JSON") from exc + if len(encoded) > limit: + raise InvalidTaskInput("Runtime snapshot payload exceeds its byte limit") + return _freeze_json(normalized) + + +def parse_assignment_payload(value: Any) -> RuntimeAssignment: + try: + return RuntimeAssignment.from_dict(_thaw_json(value)) + except (RuntimeContractError, TypeError, ValueError) as exc: + raise InvalidTaskInput("Runtime Assignment snapshot payload is invalid") from exc + + +def parse_handle_payload(value: Any) -> RuntimeExecutionHandle: + try: + return RuntimeExecutionHandle.from_dict(_thaw_json(value)) + except (RuntimeContractError, TypeError, ValueError) as exc: + raise InvalidTaskInput("Runtime handle snapshot payload is invalid") from exc + + +def _freeze_json(value: Any) -> Any: + if type(value) is dict: + return MappingProxyType({key: _freeze_json(item) for key, item in value.items()}) + if type(value) is list: + return tuple(_freeze_json(item) for item in value) + return value + + +def _thaw_json(value: Any) -> Any: + if isinstance(value, MappingProxyType): + return {key: _thaw_json(item) for key, item in value.items()} + if type(value) is dict: + return {key: _thaw_json(item) for key, item in value.items()} + if type(value) is tuple: + return [_thaw_json(item) for item in value] + if type(value) is list: + return [_thaw_json(item) for item in value] + return value + + +__all__ = ["RuntimeAssignmentSnapshot", "RuntimeHandleSnapshot", "snapshot_payload"] diff --git a/src/agentmesh/domain/runtime_execution.py b/src/agentmesh/domain/runtime_execution.py index b8567dd..06c557a 100644 --- a/src/agentmesh/domain/runtime_execution.py +++ b/src/agentmesh/domain/runtime_execution.py @@ -101,6 +101,12 @@ class RuntimeLifecycleStatus(str, Enum): EXPIRED = "EXPIRED" +class RuntimeIntegrityIncidentStatus(str, Enum): + OPEN = "OPEN" + ACKNOWLEDGED = "ACKNOWLEDGED" + ESCALATED = "ESCALATED" + + @dataclass(frozen=True) class RuntimeObservationEvidence: """Safe immutable projection of one received provider observation. @@ -203,6 +209,62 @@ def __post_init__(self) -> None: _validate_bounded_json(self.receipt_summary) +@dataclass(frozen=True) +class RuntimeIntegrityIncident: + """Safe projection of an immutable late-terminal conflict incident.""" + + id: UUID + tenant_id: str + runtime_execution_id: UUID + accepted_observation_id: str + accepted_observation_digest: str + accepted_phase: RuntimeExecutionPhase + conflicting_observation_id: str + conflicting_observation_digest: str + conflicting_phase: RuntimeExecutionPhase + status: RuntimeIntegrityIncidentStatus + reason: str + created_at: datetime + updated_at: datetime + + def __post_init__(self) -> None: + if any(type(value) is not UUID for value in (self.id, self.runtime_execution_id)): + raise InvalidTaskInput("Runtime integrity incident identity is invalid") + accepted_phases = { + RuntimeExecutionPhase.SUCCEEDED, + RuntimeExecutionPhase.FAILED, + RuntimeExecutionPhase.CANCELED, + RuntimeExecutionPhase.TIMED_OUT, + } + conflicting_phases = accepted_phases | {RuntimeExecutionPhase.LOST} + if ( + type(self.tenant_id) is not str + or not self.tenant_id.strip() + or len(self.tenant_id) > 128 + or type(self.accepted_observation_id) is not str + or not self.accepted_observation_id.strip() + or len(self.accepted_observation_id) > 512 + or type(self.conflicting_observation_id) is not str + or not self.conflicting_observation_id.strip() + or len(self.conflicting_observation_id) > 512 + or _DIGEST.fullmatch(self.accepted_observation_digest) is None + or _DIGEST.fullmatch(self.conflicting_observation_digest) is None + or self.accepted_observation_digest == self.conflicting_observation_digest + or self.accepted_phase not in accepted_phases + or self.conflicting_phase not in conflicting_phases + or type(self.status) is not RuntimeIntegrityIncidentStatus + or type(self.reason) is not str + or not self.reason.strip() + or len(self.reason) > 4096 + or type(self.created_at) is not datetime + or self.created_at.tzinfo is None + or type(self.updated_at) is not datetime + or self.updated_at.tzinfo is None + or self.updated_at < self.created_at + ): + raise InvalidTaskInput("Runtime integrity incident is invalid") + + @dataclass(frozen=True) class ReattachEvidence: execution_id: UUID diff --git a/src/agentmesh/infrastructure/postgres/runtime_repositories.py b/src/agentmesh/infrastructure/postgres/runtime_repositories.py index 88e861f..0d9a1e6 100644 --- a/src/agentmesh/infrastructure/postgres/runtime_repositories.py +++ b/src/agentmesh/infrastructure/postgres/runtime_repositories.py @@ -16,6 +16,12 @@ RuntimeComparisonRecord, RuntimeComparisonReport, ) +from agentmesh.application.runtime_snapshots import ( + RuntimeAssignmentSnapshot, + RuntimeHandleSnapshot, + parse_assignment_payload, + parse_handle_payload, +) from agentmesh.domain.errors import ( InvalidTaskInput, InvalidTaskTransition, @@ -25,6 +31,8 @@ ReattachEvidence, RuntimeExecution, RuntimeExecutionPhase, + RuntimeIntegrityIncident, + RuntimeIntegrityIncidentStatus, RuntimeLifecycleIntent, RuntimeLifecycleOperation, RuntimeLifecycleStatus, @@ -38,10 +46,10 @@ RuntimeVisibility, ) from agentmesh.infrastructure.postgres.models import ( - RuntimeComparisonRecord as RuntimeComparisonRow, -) -from agentmesh.infrastructure.postgres.models import ( + RuntimeAssignmentSnapshotRecord, RuntimeExecutionRecord, + RuntimeHandleSnapshotRecord, + RuntimeIntegrityIncidentRecord, RuntimeLifecycleOperationRecord, RuntimeObservationRecord, RuntimeOwnershipHistoryRecord, @@ -51,6 +59,9 @@ TaskRecord, TaskRunRecord, ) +from agentmesh.infrastructure.postgres.models import ( + RuntimeComparisonRecord as RuntimeComparisonRow, +) from agentmesh.runtime_sdk.descriptor import RuntimeDescriptor @@ -665,6 +676,211 @@ def update_lifecycle_status( record.updated_at = now record.version += 1 + def get_assignment_snapshot( + self, execution_id: UUID, *, tenant_id: str + ) -> RuntimeAssignmentSnapshot | None: + record = self._session.scalar( + select(RuntimeAssignmentSnapshotRecord).where( + RuntimeAssignmentSnapshotRecord.runtime_execution_id == execution_id, + RuntimeAssignmentSnapshotRecord.tenant_id == tenant_id, + ) + ) + return _assignment_snapshot_projection(record) + + def add_assignment_snapshot( + self, value: RuntimeAssignmentSnapshot + ) -> RuntimeAssignmentSnapshot: + execution = _require_execution_tenant( + self._session, value.runtime_execution_id, value.tenant_id + ) + assignment = parse_assignment_payload(value.canonical_payload) + run = self._session.get(TaskRunRecord, execution.run_id) + if ( + run is None + or execution.run_id != UUID(assignment.run_id) + or run.task_id != UUID(assignment.task_id) + or execution.runtime_version_id != UUID(assignment.runtime_version_id) + or execution.assignment_id != UUID(assignment.assignment_id) + or execution.assignment_digest != assignment.assignment_digest + ): + raise RuntimeExecutionConflict("Runtime Assignment snapshot binding conflicts") + existing = self._session.scalar( + select(RuntimeAssignmentSnapshotRecord).where( + RuntimeAssignmentSnapshotRecord.runtime_execution_id == value.runtime_execution_id + ) + ) + if existing is not None: + if existing.tenant_id != value.tenant_id: + raise RuntimeExecutionConflict( + "Runtime Assignment snapshot belongs to another tenant" + ) + current = _assignment_snapshot_projection(existing) + if current is not None and _assignment_snapshot_semantically_equal(current, value): + return current + raise RuntimeExecutionConflict("Runtime Assignment snapshot has conflicting bytes") + record = RuntimeAssignmentSnapshotRecord(**_assignment_snapshot_values(value)) + if self._session.bind is not None and self._session.bind.dialect.name == "postgresql": + inserted = self._session.execute( + postgres_insert(RuntimeAssignmentSnapshotRecord) + .values(**_assignment_snapshot_values(value)) + .on_conflict_do_nothing( + constraint="uq_runtime_assignment_snapshot_execution" + ) + ) + if inserted.rowcount: + return value + existing = self._session.scalar( + select(RuntimeAssignmentSnapshotRecord).where( + RuntimeAssignmentSnapshotRecord.runtime_execution_id + == value.runtime_execution_id + ) + ) + if existing is not None and existing.tenant_id == value.tenant_id: + current = _assignment_snapshot_projection(existing) + if current is not None and _assignment_snapshot_semantically_equal(current, value): + return current + raise RuntimeExecutionConflict("Runtime Assignment snapshot has conflicting bytes") + raise RuntimeExecutionConflict("Runtime Assignment snapshot belongs to another tenant") + self._session.add(record) + self._session.flush() + return value + + def get_handle_snapshot( + self, execution_id: UUID, *, tenant_id: str + ) -> RuntimeHandleSnapshot | None: + record = self._session.scalar( + select(RuntimeHandleSnapshotRecord).where( + RuntimeHandleSnapshotRecord.runtime_execution_id == execution_id, + RuntimeHandleSnapshotRecord.tenant_id == tenant_id, + ) + ) + return _handle_snapshot_projection(record) + + def add_handle_snapshot(self, value: RuntimeHandleSnapshot) -> RuntimeHandleSnapshot: + execution = _require_execution_tenant( + self._session, value.runtime_execution_id, value.tenant_id + ) + handle = parse_handle_payload(value.canonical_payload) + if ( + execution.runtime_version_id != UUID(handle.runtime_version_id) + or execution.assignment_id != UUID(handle.assignment_id) + or execution.assignment_digest != handle.assignment_digest + ): + raise RuntimeExecutionConflict("Runtime handle snapshot binding conflicts") + existing = self._session.scalar( + select(RuntimeHandleSnapshotRecord).where( + RuntimeHandleSnapshotRecord.runtime_execution_id == value.runtime_execution_id + ) + ) + if existing is not None: + if existing.tenant_id != value.tenant_id: + raise RuntimeExecutionConflict("Runtime handle snapshot belongs to another tenant") + current = _handle_snapshot_projection(existing) + if current is not None and _handle_snapshot_semantically_equal(current, value): + return current + raise RuntimeExecutionConflict("Runtime handle snapshot has conflicting bytes") + record = RuntimeHandleSnapshotRecord(**_handle_snapshot_values(value)) + if self._session.bind is not None and self._session.bind.dialect.name == "postgresql": + inserted = self._session.execute( + postgres_insert(RuntimeHandleSnapshotRecord) + .values(**_handle_snapshot_values(value)) + .on_conflict_do_nothing(constraint="uq_runtime_handle_snapshot_execution") + ) + if inserted.rowcount: + return value + existing = self._session.scalar( + select(RuntimeHandleSnapshotRecord).where( + RuntimeHandleSnapshotRecord.runtime_execution_id + == value.runtime_execution_id + ) + ) + if existing is not None and existing.tenant_id == value.tenant_id: + current = _handle_snapshot_projection(existing) + if current is not None and _handle_snapshot_semantically_equal(current, value): + return current + raise RuntimeExecutionConflict("Runtime handle snapshot has conflicting bytes") + raise RuntimeExecutionConflict("Runtime handle snapshot belongs to another tenant") + self._session.add(record) + self._session.flush() + return value + + def get_integrity_incident( + self, incident_id: UUID, *, tenant_id: str + ) -> RuntimeIntegrityIncident | None: + record = self._session.scalar( + select(RuntimeIntegrityIncidentRecord).where( + RuntimeIntegrityIncidentRecord.id == incident_id, + RuntimeIntegrityIncidentRecord.tenant_id == tenant_id, + ) + ) + return _integrity_incident_projection(record) + + def list_integrity_incidents( + self, execution_id: UUID, *, tenant_id: str, limit: int, offset: int + ) -> list[RuntimeIntegrityIncident]: + records = self._session.scalars( + select(RuntimeIntegrityIncidentRecord) + .where( + RuntimeIntegrityIncidentRecord.runtime_execution_id == execution_id, + RuntimeIntegrityIncidentRecord.tenant_id == tenant_id, + ) + .order_by( + RuntimeIntegrityIncidentRecord.created_at.asc(), + RuntimeIntegrityIncidentRecord.id.asc(), + ) + .limit(max(1, min(limit, 100))) + .offset(max(0, offset)) + ) + return [_integrity_incident_projection(record) for record in records] + + def add_integrity_incident(self, value: RuntimeIntegrityIncident) -> RuntimeIntegrityIncident: + _require_execution_tenant(self._session, value.runtime_execution_id, value.tenant_id) + existing = self._session.scalar( + select(RuntimeIntegrityIncidentRecord).where( + RuntimeIntegrityIncidentRecord.tenant_id == value.tenant_id, + RuntimeIntegrityIncidentRecord.runtime_execution_id == value.runtime_execution_id, + RuntimeIntegrityIncidentRecord.accepted_observation_digest + == value.accepted_observation_digest, + RuntimeIntegrityIncidentRecord.conflicting_observation_digest + == value.conflicting_observation_digest, + ) + ) + if existing is not None: + current = _integrity_incident_projection(existing) + if current is not None and _incident_evidence_semantically_equal(current, value): + return current + raise RuntimeExecutionConflict("Runtime integrity incident has conflicting evidence") + record_values = _integrity_incident_values(value) + if self._session.bind is not None and self._session.bind.dialect.name == "postgresql": + inserted = self._session.execute( + postgres_insert(RuntimeIntegrityIncidentRecord) + .values(**record_values) + .on_conflict_do_nothing(constraint="uq_runtime_integrity_incident_conflict") + ) + if inserted.rowcount: + return value + existing = self._session.scalar( + select(RuntimeIntegrityIncidentRecord).where( + RuntimeIntegrityIncidentRecord.tenant_id == value.tenant_id, + RuntimeIntegrityIncidentRecord.runtime_execution_id + == value.runtime_execution_id, + RuntimeIntegrityIncidentRecord.accepted_observation_digest + == value.accepted_observation_digest, + RuntimeIntegrityIncidentRecord.conflicting_observation_digest + == value.conflicting_observation_digest, + ) + ) + if existing is not None: + current = _integrity_incident_projection(existing) + if current is not None and _incident_evidence_semantically_equal(current, value): + return current + raise RuntimeExecutionConflict( + "Runtime integrity incident has conflicting evidence" + ) + self._session.add(RuntimeIntegrityIncidentRecord(**record_values)) + self._session.flush() + return value + @staticmethod def _scope(model: Any, *, tenant_id: str, principal_id: UUID | None) -> Any: return (model.visibility == RuntimeVisibility.PLATFORM.value) | ( @@ -824,6 +1040,162 @@ def _lifecycle_projection( ) +def _assignment_snapshot_values(value: RuntimeAssignmentSnapshot) -> dict[str, Any]: + return { + "id": value.id, + "tenant_id": value.tenant_id, + "runtime_execution_id": value.runtime_execution_id, + "contract_name": value.contract_name, + "contract_major": value.contract_major, + "assignment_id": value.assignment_id, + "assignment_digest": value.assignment_digest, + "canonical_payload": _unfreeze(value.canonical_payload), + "created_at": value.created_at, + } + + +def _assignment_snapshot_semantically_equal( + current: RuntimeAssignmentSnapshot, candidate: RuntimeAssignmentSnapshot +) -> bool: + """Compare immutable assignment meaning, excluding persistence identity/time.""" + return ( + current.tenant_id == candidate.tenant_id + and current.runtime_execution_id == candidate.runtime_execution_id + and current.contract_name == candidate.contract_name + and current.contract_major == candidate.contract_major + and current.assignment_id == candidate.assignment_id + and current.assignment_digest == candidate.assignment_digest + and current.canonical_payload == candidate.canonical_payload + ) + + +def _require_execution_tenant( + session: Session, execution_id: UUID, tenant_id: str +) -> RuntimeExecutionRecord: + """Reject snapshot writes whose execution is outside the caller tenant.""" + execution = session.scalar( + select(RuntimeExecutionRecord).where( + RuntimeExecutionRecord.id == execution_id, + RuntimeExecutionRecord.tenant_id == tenant_id, + ) + ) + if execution is None: + raise RuntimeExecutionConflict("Runtime snapshot execution tenant scope denied") + return execution + + +def _handle_snapshot_values(value: RuntimeHandleSnapshot) -> dict[str, Any]: + return { + "id": value.id, + "tenant_id": value.tenant_id, + "runtime_execution_id": value.runtime_execution_id, + "handle_digest": value.handle_digest, + "canonical_payload": _unfreeze(value.canonical_payload), + "created_at": value.created_at, + } + + +def _handle_snapshot_semantically_equal( + current: RuntimeHandleSnapshot, candidate: RuntimeHandleSnapshot +) -> bool: + """Compare immutable handle meaning, excluding persistence identity/time.""" + return ( + current.tenant_id == candidate.tenant_id + and current.runtime_execution_id == candidate.runtime_execution_id + and current.handle_digest == candidate.handle_digest + and current.canonical_payload == candidate.canonical_payload + ) + + +def _integrity_incident_values(value: RuntimeIntegrityIncident) -> dict[str, Any]: + return { + "id": value.id, + "tenant_id": value.tenant_id, + "runtime_execution_id": value.runtime_execution_id, + "accepted_observation_id": value.accepted_observation_id, + "accepted_observation_digest": value.accepted_observation_digest, + "accepted_phase": value.accepted_phase.value, + "conflicting_observation_id": value.conflicting_observation_id, + "conflicting_observation_digest": value.conflicting_observation_digest, + "conflicting_phase": value.conflicting_phase.value, + "status": value.status.value, + "reason": value.reason, + "created_at": value.created_at, + "updated_at": value.updated_at, + } + + +def _incident_evidence_semantically_equal( + current: RuntimeIntegrityIncident, candidate: RuntimeIntegrityIncident +) -> bool: + """Compare immutable conflict evidence, excluding incident metadata.""" + return ( + current.tenant_id == candidate.tenant_id + and current.runtime_execution_id == candidate.runtime_execution_id + and current.accepted_observation_digest == candidate.accepted_observation_digest + and current.conflicting_observation_digest == candidate.conflicting_observation_digest + and current.accepted_observation_id == candidate.accepted_observation_id + and current.conflicting_observation_id == candidate.conflicting_observation_id + and current.accepted_phase == candidate.accepted_phase + and current.conflicting_phase == candidate.conflicting_phase + ) + + +def _assignment_snapshot_projection( + record: RuntimeAssignmentSnapshotRecord | None, +) -> RuntimeAssignmentSnapshot | None: + if record is None: + return None + return RuntimeAssignmentSnapshot( + id=record.id, + tenant_id=record.tenant_id, + runtime_execution_id=record.runtime_execution_id, + contract_name=record.contract_name, + contract_major=record.contract_major, + assignment_id=record.assignment_id, + assignment_digest=record.assignment_digest, + canonical_payload=_freeze_projection(record.canonical_payload) or MappingProxyType({}), + created_at=record.created_at, + ) + + +def _handle_snapshot_projection( + record: RuntimeHandleSnapshotRecord | None, +) -> RuntimeHandleSnapshot | None: + if record is None: + return None + return RuntimeHandleSnapshot( + id=record.id, + tenant_id=record.tenant_id, + runtime_execution_id=record.runtime_execution_id, + handle_digest=record.handle_digest, + canonical_payload=_freeze_projection(record.canonical_payload) or MappingProxyType({}), + created_at=record.created_at, + ) + + +def _integrity_incident_projection( + record: RuntimeIntegrityIncidentRecord | None, +) -> RuntimeIntegrityIncident | None: + if record is None: + return None + return RuntimeIntegrityIncident( + id=record.id, + tenant_id=record.tenant_id, + runtime_execution_id=record.runtime_execution_id, + accepted_observation_id=record.accepted_observation_id, + accepted_observation_digest=record.accepted_observation_digest, + accepted_phase=RuntimeExecutionPhase(record.accepted_phase), + conflicting_observation_id=record.conflicting_observation_id, + conflicting_observation_digest=record.conflicting_observation_digest, + conflicting_phase=RuntimeExecutionPhase(record.conflicting_phase), + status=RuntimeIntegrityIncidentStatus(record.status), + reason=record.reason, + created_at=record.created_at, + updated_at=record.updated_at, + ) + + class SqlAlchemyRuntimeComparisonRepository: """Tenant-scoped durable parity audit; no raw provider body is retained.""" diff --git a/tests/integration/test_runtime_snapshot_repository_postgres.py b/tests/integration/test_runtime_snapshot_repository_postgres.py new file mode 100644 index 0000000..e40e181 --- /dev/null +++ b/tests/integration/test_runtime_snapshot_repository_postgres.py @@ -0,0 +1,224 @@ +"""PostgreSQL compatibility readers for immutable Runtime snapshots.""" + +from __future__ import annotations + +import os +from datetime import datetime, timedelta, timezone +from uuid import UUID, uuid4 + +import pytest +from sqlalchemy import create_engine, func, select +from sqlalchemy.orm import Session + +from agentmesh.application.runtime_snapshots import ( + RuntimeAssignmentSnapshot, + RuntimeHandleSnapshot, +) +from agentmesh.config import get_settings +from agentmesh.domain.errors import RuntimeExecutionConflict +from agentmesh.domain.runtime_execution import ( + RuntimeExecutionPhase, + RuntimeIntegrityIncident, + RuntimeIntegrityIncidentStatus, +) +from agentmesh.infrastructure.postgres.models import ( + RuntimeAssignmentSnapshotRecord, + RuntimeExecutionRecord, + RuntimeHandleSnapshotRecord, + TaskRunRecord, +) +from agentmesh.runtime_sdk.assignment import RuntimeAssignment, RuntimeExecutionHandle +from agentmesh.runtime_sdk.canonical import canonical_digest +from tests.integration.test_runtime_control_plane_postgres import _fixture + +pytestmark = [ + pytest.mark.postgres, + pytest.mark.skipif( + os.getenv("AGENTMESH_RUN_POSTGRES_TESTS") != "1", + reason="set AGENTMESH_RUN_POSTGRES_TESTS=1 to run PostgreSQL tests", + ), +] + + +def test_snapshot_roundtrip_tenant_scope_replay_and_conflict() -> None: + engine = create_engine(get_settings().database_url) + try: + with Session(engine) as session: + repository, execution = _fixture(session) + now = datetime.now(timezone.utc) + run_record = session.get(TaskRunRecord, execution.run_id) + assert run_record is not None + assignment_dto = RuntimeAssignment( + assignment_id=str(execution.assignment_id), + tenant_id=execution.tenant_id, + task_id=str(run_record.task_id), + run_id=str(execution.run_id), + agent_definition_id=str(uuid4()), + agent_version_id=str(uuid4()), + agent_version_digest="a" * 64, + runtime_version_id=str(execution.runtime_version_id), + runtime_descriptor_digest="b" * 64, + execution_mode="managed_async", + run_role="EXECUTOR", + revision=0, + objective="bounded", + structured_input={"n": 1}, + ) + execution_record = session.get(RuntimeExecutionRecord, execution.id) + assert execution_record is not None + execution_record.assignment_digest = assignment_dto.assignment_digest or "" + session.flush() + assignment = RuntimeAssignmentSnapshot( + id=uuid4(), + tenant_id=execution.tenant_id, + runtime_execution_id=execution.id, + contract_name=assignment_dto.schema_name, + contract_major=assignment_dto.schema_version, + assignment_id=execution.assignment_id, + assignment_digest=assignment_dto.assignment_digest or "", + canonical_payload=assignment_dto.to_dict(), + created_at=now, + ) + assert repository.add_assignment_snapshot(assignment) == assignment + assert repository.get_assignment_snapshot( + execution.id, tenant_id=execution.tenant_id + ) == assignment + assert repository.get_assignment_snapshot( + execution.id, tenant_id="other-tenant" + ) is None + assignment_replay = RuntimeAssignmentSnapshot( + **{ + **assignment.__dict__, + "id": uuid4(), + "created_at": now + timedelta(seconds=1), + } + ) + assert repository.add_assignment_snapshot(assignment_replay) == assignment + assert session.scalar( + select(func.count(RuntimeAssignmentSnapshotRecord.id)).where( + RuntimeAssignmentSnapshotRecord.runtime_execution_id == execution.id + ) + ) == 1 + wrong_run_assignment = RuntimeAssignment( + **{ + **assignment_dto.__dict__, + "run_id": str(uuid4()), + "assignment_digest": None, + } + ) + with pytest.raises(RuntimeExecutionConflict, match="binding conflicts"): + repository.add_assignment_snapshot( + RuntimeAssignmentSnapshot( + id=uuid4(), + tenant_id=execution.tenant_id, + runtime_execution_id=execution.id, + contract_name=wrong_run_assignment.schema_name, + contract_major=wrong_run_assignment.schema_version, + assignment_id=UUID(wrong_run_assignment.assignment_id), + assignment_digest=wrong_run_assignment.assignment_digest or "", + canonical_payload=wrong_run_assignment.to_dict(), + created_at=now + timedelta(seconds=2), + ) + ) + changed_assignment_dto = RuntimeAssignment( + **{**assignment_dto.__dict__, "objective": "different", "assignment_digest": None} + ) + with pytest.raises(RuntimeExecutionConflict, match="binding conflicts"): + repository.add_assignment_snapshot( + RuntimeAssignmentSnapshot( + id=uuid4(), + tenant_id=execution.tenant_id, + runtime_execution_id=execution.id, + contract_name=changed_assignment_dto.schema_name, + contract_major=changed_assignment_dto.schema_version, + assignment_id=execution.assignment_id, + assignment_digest=changed_assignment_dto.assignment_digest or "", + canonical_payload=changed_assignment_dto.to_dict(), + created_at=now + timedelta(seconds=2), + ) + ) + + handle_dto = RuntimeExecutionHandle( + runtime_execution_id=str(execution.id), + runtime_version_id=str(execution.runtime_version_id), + provider_execution_ref="opaque-ref", + assignment_id=str(execution.assignment_id), + assignment_digest=assignment_dto.assignment_digest or "", + created_at=now, + ) + handle = RuntimeHandleSnapshot( + id=uuid4(), + tenant_id=execution.tenant_id, + runtime_execution_id=execution.id, + handle_digest=canonical_digest(handle_dto.to_dict()), + canonical_payload=handle_dto.to_dict(), + created_at=now, + ) + assert repository.add_handle_snapshot(handle) == handle + assert repository.get_handle_snapshot( + execution.id, tenant_id=execution.tenant_id + ) == handle + assert repository.get_handle_snapshot(execution.id, tenant_id="other-tenant") is None + assert session.scalar( + select(func.count(RuntimeHandleSnapshotRecord.id)).where( + RuntimeHandleSnapshotRecord.runtime_execution_id == execution.id + ) + ) == 1 + handle_replay = RuntimeHandleSnapshot( + **{ + **handle.__dict__, + "id": uuid4(), + "created_at": now + timedelta(seconds=1), + } + ) + assert repository.add_handle_snapshot(handle_replay) == handle + + incident = RuntimeIntegrityIncident( + id=uuid4(), + tenant_id=execution.tenant_id, + runtime_execution_id=execution.id, + accepted_observation_id="accepted", + accepted_observation_digest="f" * 64, + accepted_phase=RuntimeExecutionPhase.SUCCEEDED, + conflicting_observation_id="conflict", + conflicting_observation_digest="1" * 64, + conflicting_phase=RuntimeExecutionPhase.FAILED, + status=RuntimeIntegrityIncidentStatus.OPEN, + reason="conflict", + created_at=now, + updated_at=now, + ) + assert repository.add_integrity_incident(incident) == incident + incident_replay = RuntimeIntegrityIncident( + **{ + **incident.__dict__, + "id": uuid4(), + "reason": "same evidence, retried", + "created_at": now + timedelta(seconds=1), + "updated_at": now + timedelta(seconds=2), + } + ) + assert repository.add_integrity_incident(incident_replay) == incident + with pytest.raises(RuntimeExecutionConflict, match="conflicting evidence"): + repository.add_integrity_incident( + RuntimeIntegrityIncident( + **{ + **incident.__dict__, + "id": uuid4(), + "conflicting_observation_id": "different-conflict", + "created_at": now + timedelta(seconds=3), + "updated_at": now + timedelta(seconds=3), + } + ) + ) + assert repository.get_integrity_incident( + incident.id, tenant_id=execution.tenant_id + ) == incident + assert repository.get_integrity_incident( + incident.id, tenant_id="other-tenant" + ) is None + assert repository.list_integrity_incidents( + execution.id, tenant_id=execution.tenant_id, limit=200, offset=0 + ) == [incident] + finally: + engine.dispose() diff --git a/tests/test_runtime_snapshot_domain.py b/tests/test_runtime_snapshot_domain.py new file mode 100644 index 0000000..d2c795c --- /dev/null +++ b/tests/test_runtime_snapshot_domain.py @@ -0,0 +1,179 @@ +from datetime import datetime, timedelta, timezone +from uuid import UUID, uuid4 + +import pytest + +from agentmesh.application.runtime_snapshots import ( + RuntimeAssignmentSnapshot, + RuntimeHandleSnapshot, + snapshot_payload, +) +from agentmesh.domain.errors import InvalidTaskInput +from agentmesh.domain.runtime_execution import ( + RuntimeExecutionPhase, + RuntimeIntegrityIncident, + RuntimeIntegrityIncidentStatus, +) +from agentmesh.runtime_sdk.assignment import RuntimeAssignment, RuntimeExecutionHandle +from agentmesh.runtime_sdk.canonical import canonical_digest + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _assignment() -> RuntimeAssignment: + return RuntimeAssignment( + assignment_id=str(uuid4()), + tenant_id="tenant-a", + task_id=str(uuid4()), + run_id=str(uuid4()), + agent_definition_id=str(uuid4()), + agent_version_id=str(uuid4()), + agent_version_digest="a" * 64, + runtime_version_id=str(uuid4()), + runtime_descriptor_digest="b" * 64, + execution_mode="managed_async", + run_role="EXECUTOR", + revision=0, + objective="bounded objective", + structured_input={"value": "safe"}, + deadline=_now() + timedelta(minutes=5), + ) + + +def _assignment_snapshot(assignment: RuntimeAssignment) -> RuntimeAssignmentSnapshot: + return RuntimeAssignmentSnapshot( + id=uuid4(), + tenant_id=assignment.tenant_id, + runtime_execution_id=uuid4(), + contract_name=assignment.schema_name, + contract_major=assignment.schema_version, + assignment_id=UUID(assignment.assignment_id), + assignment_digest=assignment.assignment_digest or "", + canonical_payload=assignment.to_dict(), + created_at=_now(), + ) + + +def _handle(assignment: RuntimeAssignment, execution_id: UUID) -> RuntimeExecutionHandle: + return RuntimeExecutionHandle( + runtime_execution_id=str(execution_id), + runtime_version_id=assignment.runtime_version_id, + provider_execution_ref="opaque-provider-ref", + assignment_id=assignment.assignment_id, + assignment_digest=assignment.assignment_digest or "", + created_at=_now(), + ) + + +def test_assignment_snapshot_accepts_real_runtime_dto_and_freezes_payload() -> None: + assignment = _assignment() + value = _assignment_snapshot(assignment) + assert value.canonical_payload["deadline"].endswith("Z") + assert value.canonical_payload["assignment_id"] == assignment.assignment_id + assert value.canonical_payload["structured_input"] == {"value": "safe"} + + +@pytest.mark.parametrize( + "payload", + [ + {"bad": float("nan")}, + {"bad": float("inf")}, + {"bad": "surrogate\ud800"}, + {"bad": 9_007_199_254_740_992}, + ["root must be object"], + {"schema_name": "agentmesh.runtime-assignment"}, + ], +) +def test_snapshot_rejects_non_jcs_or_partial_payload(payload: object) -> None: + with pytest.raises(InvalidTaskInput): + RuntimeHandleSnapshot( + id=uuid4(), + tenant_id="tenant-a", + runtime_execution_id=uuid4(), + handle_digest="b" * 64, + canonical_payload=payload, # type: ignore[arg-type] + created_at=_now(), + ) + + +def test_handle_snapshot_accepts_real_runtime_dto_and_checks_digest_identity() -> None: + assignment = _assignment() + execution_id = uuid4() + handle = _handle(assignment, execution_id) + value = RuntimeHandleSnapshot( + id=uuid4(), + tenant_id=assignment.tenant_id, + runtime_execution_id=execution_id, + handle_digest=canonical_digest(handle.to_dict()), + canonical_payload=handle.to_dict(), + created_at=_now(), + ) + assert value.canonical_payload["created_at"].endswith("Z") + with pytest.raises(InvalidTaskInput): + RuntimeHandleSnapshot( + **{**value.__dict__, "handle_digest": "b" * 64} + ) + with pytest.raises(InvalidTaskInput): + RuntimeHandleSnapshot( + **{ + **value.__dict__, + "runtime_execution_id": uuid4(), + } + ) + + +def test_snapshot_rejects_naive_or_oversize_handle_payload() -> None: + assignment = _assignment() + handle = _handle(assignment, uuid4()) + with pytest.raises(InvalidTaskInput): + RuntimeHandleSnapshot( + **{**RuntimeHandleSnapshot( + id=uuid4(), + tenant_id=assignment.tenant_id, + runtime_execution_id=UUID(handle.runtime_execution_id), + handle_digest=canonical_digest(handle.to_dict()), + canonical_payload=handle.to_dict(), + created_at=_now(), + ).__dict__, "created_at": datetime.now()} + ) + with pytest.raises(InvalidTaskInput): + snapshot_payload({"value": "x" * 70_000}, limit=65_536) + + +def test_integrity_incident_has_closed_status_and_phase_invariants() -> None: + now = _now() + value = RuntimeIntegrityIncident( + id=uuid4(), + tenant_id="tenant-a", + runtime_execution_id=uuid4(), + accepted_observation_id="accepted", + accepted_observation_digest="a" * 64, + accepted_phase=RuntimeExecutionPhase.SUCCEEDED, + conflicting_observation_id="conflict", + conflicting_observation_digest="b" * 64, + conflicting_phase=RuntimeExecutionPhase.LOST, + status=RuntimeIntegrityIncidentStatus.OPEN, + reason="late conflicting terminal", + created_at=now, + updated_at=now, + ) + assert value.status is RuntimeIntegrityIncidentStatus.OPEN + with pytest.raises(InvalidTaskInput): + RuntimeIntegrityIncident(**{**value.__dict__, "status": "RESOLVED"}) # type: ignore[arg-type] + with pytest.raises(InvalidTaskInput): + RuntimeIntegrityIncident( + **{**value.__dict__, "accepted_phase": RuntimeExecutionPhase.LOST} + ) + with pytest.raises(InvalidTaskInput): + RuntimeIntegrityIncident( + **{ + **value.__dict__, + "conflicting_observation_digest": "a" * 64, + } + ) + with pytest.raises(InvalidTaskInput): + RuntimeIntegrityIncident( + **{**value.__dict__, "updated_at": now - timedelta(seconds=1)} + )