import sqlite3 from contextlib import closing from pathlib import Path import pytest from agentci.engine.model import JobKind, JobStatus, Workflow, WorkflowKind, WorkflowStatus from agentci.engine.repository import Repository MIGRATIONS = Path(__file__).parents[1] / "src" / "agentci" / "migrations" PRE_V3_CREATED_AT = "2026-01-01T00:00:00+00:00" def create_schema_v2_database(database_path: Path, workspace: Path) -> None: with closing(sqlite3.connect(database_path)) as connection, connection: connection.executescript((MIGRATIONS / "001_initial.sql").read_text()) connection.execute( "INSERT INTO schema_migrations(version, applied_at) VALUES (?, ?)", (1, "2025-11-01T00:00:00+00:00"), ) connection.executescript((MIGRATIONS / "002_opencode_sessions.sql").read_text()) connection.execute( "INSERT INTO schema_migrations(version, applied_at) VALUES (?, ?)", (2, "2025-12-01T00:00:00+00:00"), ) connection.executemany( "INSERT INTO deliveries(delivery_id, comment_id, received_at) VALUES (?, ?, ?)", [ ("delivery-queued", 102, "2025-12-31T23:59:58+00:00"), ("delivery-unrelated", 999, "2025-12-31T23:59:59+00:00"), ], ) connection.executemany( """INSERT INTO workflows( id, kind, repo_owner, repo_name, issue_number, pr_number, base_sha, branch, workspace_path, primary_session_id, reviewer_session_id, artifact, review_json, status, created_at, updated_at, runtime ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", [ ( "pre-v3-plan-workflow", "plan", "alice", "repo", 7, None, "plan-base", "agent/plan", str(workspace / "plan"), "plan-primary", "plan-reviewer", "# Existing plan", '{"verdict":"approved"}', "active", "2025-12-30T00:00:00+00:00", "2025-12-31T00:00:00+00:00", "opencode", ), ( "pre-v3-implementation-workflow", "implement", "alice", "repo", 8, 44, "implementation-base", "agent/implementation", str(workspace / "implementation"), "implementation-primary", "implementation-reviewer", "# Existing implementation", '{"verdict":"changes_requested"}', "completed", "2025-12-29T00:00:00+00:00", "2025-12-31T12:00:00+00:00", "codex", ), ], ) connection.executemany( """INSERT INTO jobs( id, kind, target_key, repo_owner, repo_name, issue_number, pr_number, requester, message, comment_id, workflow_id, status, stage, error, accepted_comment_id, started_comment_id, created_at, started_at, finished_at, runtime_session_id ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", [ ( "pre-v3-job-a-terminal", "iterate_implement", "alice/repo:pr:44", "alice", "repo", 8, 44, "bob", "address review", 101, "pre-v3-implementation-workflow", "succeeded", "completed", None, 201, 202, PRE_V3_CREATED_AT, "2026-01-01T00:01:00+00:00", "2026-01-01T00:02:00+00:00", "implementation-runtime", ), ( "pre-v3-job-b-queued", "iterate_plan", "alice/repo:issue:7", "alice", "repo", 7, None, "alice", "refine plan", 102, "pre-v3-plan-workflow", "queued", "queued", None, 203, 204, PRE_V3_CREATED_AT, None, None, "plan-runtime", ), ], ) @pytest.fixture async def migrated_pre_v3_repository(tmp_path: Path) -> Repository: database_path = tmp_path / "schema-v2.sqlite3" create_schema_v2_database(database_path, tmp_path / "workspaces") repository = Repository(database_path, MIGRATIONS) await repository.initialize() return repository async def test_schema_v2_migration_preserves_jobs_and_reconstructs_state_machine_identity( migrated_pre_v3_repository: Repository, ) -> None: terminal = await migrated_pre_v3_repository.get_job("pre-v3-job-a-terminal") queued = await migrated_pre_v3_repository.get_job("pre-v3-job-b-queued") with ( closing(sqlite3.connect(migrated_pre_v3_repository.database_path)) as connection, connection, ): migrated = connection.execute( "SELECT id, delivery_id, receive_sequence, command_body FROM jobs " "ORDER BY receive_sequence" ).fetchall() storage_fields = connection.execute( """SELECT id, started_comment_id, created_at, started_at, finished_at FROM jobs ORDER BY receive_sequence""" ).fetchall() assert migrated == [ ( "pre-v3-job-a-terminal", "legacy:pre-v3-job-a-terminal", 1, "/agent iterate address review", ), ("pre-v3-job-b-queued", "delivery-queued", 2, "/agent iterate refine plan"), ] assert ( terminal and ( terminal.kind, terminal.target_key, terminal.repo_owner, terminal.repo_name, terminal.issue_number, terminal.pr_number, terminal.requester, terminal.message, terminal.comment_id, terminal.workflow_id, terminal.status, terminal.stage, terminal.error, terminal.runtime_session_id, terminal.accepted_comment_id, terminal.comment_body, ), queued and ( queued.kind, queued.target_key, queued.requester, queued.message, queued.comment_id, queued.workflow_id, queued.status, queued.stage, queued.runtime_session_id, queued.accepted_comment_id, ), storage_fields, ) == ( ( JobKind.ITERATE_IMPLEMENT, "alice/repo:pr:44", "alice", "repo", 8, 44, "bob", "address review", 101, "pre-v3-implementation-workflow", JobStatus.SUCCEEDED, "completed", None, "implementation-runtime", 201, None, ), ( JobKind.ITERATE_PLAN, "alice/repo:issue:7", "alice", "refine plan", 102, "pre-v3-plan-workflow", JobStatus.QUEUED, "queued", "plan-runtime", 203, ), [ ( "pre-v3-job-a-terminal", 202, PRE_V3_CREATED_AT, "2026-01-01T00:01:00+00:00", "2026-01-01T00:02:00+00:00", ), ("pre-v3-job-b-queued", 204, PRE_V3_CREATED_AT, None, None), ], ) async def test_schema_v2_migration_preserves_pre_v3_workflows( migrated_pre_v3_repository: Repository, tmp_path: Path, ) -> None: plan = await migrated_pre_v3_repository.get_workflow("pre-v3-plan-workflow") implementation = await migrated_pre_v3_repository.get_workflow("pre-v3-implementation-workflow") with ( closing(sqlite3.connect(migrated_pre_v3_repository.database_path)) as connection, connection, ): timestamps = connection.execute( "SELECT id, created_at, updated_at FROM workflows ORDER BY id" ).fetchall() assert ((plan, implementation), timestamps) == ( ( Workflow( id="pre-v3-plan-workflow", kind=WorkflowKind.PLAN, repo_owner="alice", repo_name="repo", issue_number=7, workspace_path=tmp_path / "workspaces" / "plan", base_sha="plan-base", runtime="opencode", branch="agent/plan", primary_session_id="plan-primary", reviewer_session_id="plan-reviewer", artifact="# Existing plan", review_json='{"verdict":"approved"}', status=WorkflowStatus.ACTIVE, ), Workflow( id="pre-v3-implementation-workflow", kind=WorkflowKind.IMPLEMENT, repo_owner="alice", repo_name="repo", issue_number=8, pr_number=44, workspace_path=tmp_path / "workspaces" / "implementation", base_sha="implementation-base", runtime="codex", branch="agent/implementation", primary_session_id="implementation-primary", reviewer_session_id="implementation-reviewer", artifact="# Existing implementation", review_json='{"verdict":"changes_requested"}', status=WorkflowStatus.COMPLETED, ), ), [ ( "pre-v3-implementation-workflow", "2025-12-29T00:00:00+00:00", "2025-12-31T12:00:00+00:00", ), ( "pre-v3-plan-workflow", "2025-12-30T00:00:00+00:00", "2025-12-31T00:00:00+00:00", ), ], ) async def test_schema_v2_migration_creates_audit_events_and_only_queued_tasks( migrated_pre_v3_repository: Repository, ) -> None: with ( closing(sqlite3.connect(migrated_pre_v3_repository.database_path)) as connection, connection, ): events = connection.execute( """SELECT event_id, job_id, event_type, payload_json, created_at FROM job_events ORDER BY job_id""" ).fetchall() tasks = connection.execute( """SELECT job_id, source_event_id, ordinal, listener, queue, status, available_at, created_at FROM listener_tasks ORDER BY ordinal""" ).fetchall() assert (events, tasks) == ( [ ( "delivery:legacy:pre-v3-job-a-terminal", "pre-v3-job-a-terminal", "legacy", "{}", PRE_V3_CREATED_AT, ), ( "delivery:delivery-queued", "pre-v3-job-b-queued", "legacy", "{}", PRE_V3_CREATED_AT, ), ], [ ( "pre-v3-job-b-queued", "delivery:delivery-queued", 0, "execute", "jobs", "pending", PRE_V3_CREATED_AT, PRE_V3_CREATED_AT, ), ( "pre-v3-job-b-queued", "delivery:delivery-queued", 1, "reconcile_comment", "control", "pending", PRE_V3_CREATED_AT, PRE_V3_CREATED_AT, ), ], ) def durable_snapshot(database_path: Path) -> tuple[object, ...]: with closing(sqlite3.connect(database_path)) as connection, connection: return ( connection.execute("SELECT version FROM schema_migrations ORDER BY version").fetchall(), connection.execute( """SELECT id, kind, target_key, repo_owner, repo_name, issue_number, pr_number, requester, message, comment_id, delivery_id, receive_sequence, command_body, workflow_id, status, stage, error, runtime_session_id, accepted_comment_id, started_comment_id, comment_body, created_at, started_at, finished_at FROM jobs ORDER BY id""" ).fetchall(), connection.execute( """SELECT id, kind, repo_owner, repo_name, issue_number, pr_number, base_sha, branch, workspace_path, primary_session_id, reviewer_session_id, artifact, review_json, status, runtime, created_at, updated_at FROM workflows ORDER BY id""" ).fetchall(), connection.execute( """SELECT event_id, job_id, event_type, payload_json FROM job_events ORDER BY event_id""" ).fetchall(), connection.execute( """SELECT job_id, source_event_id, ordinal, listener, queue, status FROM listener_tasks ORDER BY id""" ).fetchall(), ) async def test_reopening_migrated_schema_v2_database_is_idempotent( migrated_pre_v3_repository: Repository, ) -> None: before = durable_snapshot(migrated_pre_v3_repository.database_path) reopened = Repository(migrated_pre_v3_repository.database_path, MIGRATIONS) await reopened.initialize() assert (before[0], durable_snapshot(reopened.database_path)) == ( [(1,), (2,), (3,)], before, )