From 44cfca6210b023dfd2bf6485090d3fa33168eb2b Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:20:00 +0800 Subject: [PATCH 01/33] test: require result-aware migration head --- tests/integration/db/test_migrations.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/integration/db/test_migrations.py b/tests/integration/db/test_migrations.py index 5366de7..b122f5d 100644 --- a/tests/integration/db/test_migrations.py +++ b/tests/integration/db/test_migrations.py @@ -29,7 +29,7 @@ async def test_upgrade_head_succeeds_on_an_isolated_empty_database( try: with engine.connect() as connection: version = connection.scalar(text("SELECT version_num FROM alembic_version")) - assert version == "0012_version_identity_nonunique" + assert version == "0013_result_aware_agent_runs" finally: engine.dispose() @@ -45,6 +45,7 @@ async def test_upgrade_head_succeeds_on_an_isolated_empty_database( assert Path("migrations/versions/0010_attempt_physical_execution.py").exists() assert Path("migrations/versions/0011_result_plane.py").exists() assert Path("migrations/versions/0012_version_identity_nonunique.py").exists() + assert Path("migrations/versions/0013_result_aware_agent_runs.py").exists() @pytest.mark.integration From d69b904f982dd2721aca68f4dcf4898f799718cc Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:24:32 +0800 Subject: [PATCH 02/33] feat: add result-aware AgentRun provenance migration --- .../versions/0013_result_aware_agent_runs.py | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 migrations/versions/0013_result_aware_agent_runs.py diff --git a/migrations/versions/0013_result_aware_agent_runs.py b/migrations/versions/0013_result_aware_agent_runs.py new file mode 100644 index 0000000..4e8af6c --- /dev/null +++ b/migrations/versions/0013_result_aware_agent_runs.py @@ -0,0 +1,45 @@ +"""add result-aware AgentRun provenance + +Revision ID: 0013_result_aware_agent_runs +Revises: 0012_version_identity_nonunique +Create Date: 2026-09-20 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0013_result_aware_agent_runs" +down_revision: str | None = "0012_version_identity_nonunique" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "agent_runs", + sa.Column( + "source_batch_summary_revision_id", + postgresql.UUID(as_uuid=True), + nullable=True, + ), + ) + op.create_foreign_key( + "fk_agent_runs_source_summary_revision_id_batch_summary_revisions", + "agent_runs", + "batch_summary_revisions", + ["source_batch_summary_revision_id"], + ["batch_summary_revision_id"], + ondelete="RESTRICT", + ) + + +def downgrade() -> None: + op.drop_constraint( + "fk_agent_runs_source_summary_revision_id_batch_summary_revisions", + "agent_runs", + type_="foreignkey", + ) + op.drop_column("agent_runs", "source_batch_summary_revision_id") From 06f12f0b5f1d414fc0d14312b5f1474bf3a148dc Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:24:43 +0800 Subject: [PATCH 03/33] feat: persist AgentRun source summary provenance --- src/labpilot/agent_runtime/models.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/labpilot/agent_runtime/models.py b/src/labpilot/agent_runtime/models.py index 322cdb0..7c43d15 100644 --- a/src/labpilot/agent_runtime/models.py +++ b/src/labpilot/agent_runtime/models.py @@ -37,6 +37,12 @@ class AgentRunRow(Base): name="fk_agent_runs_produced_proposal_revision_id_proposal_revisions", ondelete="RESTRICT", ), + ForeignKeyConstraint( + ["source_batch_summary_revision_id"], + ["batch_summary_revisions.batch_summary_revision_id"], + name="fk_agent_runs_source_summary_revision_id_batch_summary_revisions", + ondelete="RESTRICT", + ), PrimaryKeyConstraint("agent_run_id", name="pk_agent_runs"), CheckConstraint( "status IN ('ready','running','awaiting_human','completed'," @@ -73,6 +79,9 @@ class AgentRunRow(Base): produced_proposal_revision_id: Mapped[UUID | None] = mapped_column( PGUUID(as_uuid=True), nullable=True ) + source_batch_summary_revision_id: Mapped[UUID | None] = mapped_column( + PGUUID(as_uuid=True), nullable=True + ) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) From 20ea6e45bec1f22e770808ab35ce7785011ea835 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:24:53 +0800 Subject: [PATCH 04/33] feat: expose AgentRun source summary provenance --- src/labpilot/agent_runtime/types.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/labpilot/agent_runtime/types.py b/src/labpilot/agent_runtime/types.py index 66d9a1b..fab5315 100644 --- a/src/labpilot/agent_runtime/types.py +++ b/src/labpilot/agent_runtime/types.py @@ -175,6 +175,7 @@ class AgentRunDetails: started_at: datetime | None completed_at: datetime | None row_version: int + source_batch_summary_revision_id: UUID | None = None @dataclass(frozen=True, slots=True) From 40856bfd8b4e42639110b6ba9a765f0f2cb89016 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:25:03 +0800 Subject: [PATCH 05/33] feat: round trip AgentRun source summary provenance --- src/labpilot/agent_runtime/repository.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/labpilot/agent_runtime/repository.py b/src/labpilot/agent_runtime/repository.py index deb84ba..f36d397 100644 --- a/src/labpilot/agent_runtime/repository.py +++ b/src/labpilot/agent_runtime/repository.py @@ -285,6 +285,7 @@ def _run_row(run: AgentRunDetails) -> AgentRunRow: total_tokens=run.total_tokens, next_step_ordinal=run.next_step_ordinal, produced_proposal_revision_id=run.produced_proposal_revision_id, + source_batch_summary_revision_id=run.source_batch_summary_revision_id, created_at=run.created_at, started_at=run.started_at, completed_at=run.completed_at, @@ -337,6 +338,7 @@ def _run_details(row: AgentRunRow) -> AgentRunDetails: next_step_ordinal=row.next_step_ordinal, produced_proposal_revision_id=row.produced_proposal_revision_id, created_at=row.created_at, + source_batch_summary_revision_id=row.source_batch_summary_revision_id, started_at=row.started_at, completed_at=row.completed_at, row_version=row.row_version, From 1e514fbdc5d6e2348bc65b3b41815682ef205ff7 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:25:15 +0800 Subject: [PATCH 06/33] feat: return AgentRun source summary provenance --- src/labpilot/api/schemas/agent_runs.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/labpilot/api/schemas/agent_runs.py b/src/labpilot/api/schemas/agent_runs.py index 05b3ba3..ed0ca8c 100644 --- a/src/labpilot/api/schemas/agent_runs.py +++ b/src/labpilot/api/schemas/agent_runs.py @@ -50,6 +50,7 @@ class AgentRunResponse(BaseModel): status: str termination_reason: str | None produced_proposal_revision_id: UUID | None + source_batch_summary_revision_id: UUID | None created_at: datetime row_version: int @@ -62,6 +63,7 @@ def from_details(cls, details: AgentRunDetails) -> "AgentRunResponse": status=details.status, termination_reason=details.termination_reason, produced_proposal_revision_id=details.produced_proposal_revision_id, + source_batch_summary_revision_id=details.source_batch_summary_revision_id, created_at=details.created_at, row_version=details.row_version, ) @@ -111,6 +113,7 @@ class AgentRunTraceResponse(BaseModel): agent_run_id: UUID project_id: UUID repository_snapshot_id: UUID + source_batch_summary_revision_id: UUID | None goal: str status: str termination_reason: str | None @@ -131,6 +134,7 @@ def from_details(cls, details: AgentRunTraceDetails) -> "AgentRunTraceResponse": agent_run_id=run.agent_run_id, project_id=run.project_id, repository_snapshot_id=run.repository_snapshot_id, + source_batch_summary_revision_id=run.source_batch_summary_revision_id, goal=run.goal, status=run.status, termination_reason=run.termination_reason, From 54baa68803a616e367d4e7e6cf423c7456fb4700 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:25:56 +0800 Subject: [PATCH 07/33] test: expect result-aware migration head --- tests/integration/db/test_migrations.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/integration/db/test_migrations.py b/tests/integration/db/test_migrations.py index b122f5d..30a3b4b 100644 --- a/tests/integration/db/test_migrations.py +++ b/tests/integration/db/test_migrations.py @@ -178,7 +178,7 @@ async def test_task3_migration_upgrade_downgrade_upgrade_round_trip( with engine.connect() as connection: assert ( connection.scalar(text("SELECT version_num FROM alembic_version")) - == "0012_version_identity_nonunique" + == "0013_result_aware_agent_runs" ) assert connection.scalar(text("SELECT to_regclass('public.projects')")) == "projects" @@ -194,7 +194,7 @@ async def test_task3_migration_upgrade_downgrade_upgrade_round_trip( with engine.connect() as connection: assert ( connection.scalar(text("SELECT version_num FROM alembic_version")) - == "0012_version_identity_nonunique" + == "0013_result_aware_agent_runs" ) assert connection.scalar(text("SELECT to_regclass('public.projects')")) == "projects" finally: @@ -214,7 +214,7 @@ async def test_task4_migration_upgrade_downgrade_upgrade_round_trip( with engine.connect() as connection: assert ( connection.scalar(text("SELECT version_num FROM alembic_version")) - == "0012_version_identity_nonunique" + == "0013_result_aware_agent_runs" ) assert ( connection.scalar(text("SELECT to_regclass('public.agent_steps')")) == "agent_steps" @@ -233,7 +233,7 @@ async def test_task4_migration_upgrade_downgrade_upgrade_round_trip( with engine.connect() as connection: assert ( connection.scalar(text("SELECT version_num FROM alembic_version")) - == "0012_version_identity_nonunique" + == "0013_result_aware_agent_runs" ) assert ( connection.scalar(text("SELECT to_regclass('public.agent_steps')")) == "agent_steps" @@ -562,7 +562,7 @@ async def test_result_plane_migration_upgrade_downgrade_upgrade_round_trip( with engine.connect() as connection: assert ( connection.scalar(text("SELECT version_num FROM alembic_version")) - == "0012_version_identity_nonunique" + == "0013_result_aware_agent_runs" ) assert { connection.scalar(text("SELECT to_regclass(:name)"), {"name": f"public.{table}"}) @@ -587,7 +587,7 @@ async def test_result_plane_migration_upgrade_downgrade_upgrade_round_trip( with engine.connect() as connection: assert ( connection.scalar(text("SELECT version_num FROM alembic_version")) - == "0012_version_identity_nonunique" + == "0013_result_aware_agent_runs" ) assert { connection.scalar(text("SELECT to_regclass(:name)"), {"name": f"public.{table}"}) From 0cfd7bdbeb6ce52022e280dc3cd218671074629c Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:26:03 +0800 Subject: [PATCH 08/33] test: expect result-aware migration head --- tests/integration/test_execution_schema.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/test_execution_schema.py b/tests/integration/test_execution_schema.py index cd2a095..9cc7b0d 100644 --- a/tests/integration/test_execution_schema.py +++ b/tests/integration/test_execution_schema.py @@ -45,7 +45,7 @@ async def test_migration_0009_registers_execution_preparation_schema( await database.dispose() assert {"batches", "trials", "attempts"} <= tables - assert revision == "0012_version_identity_nonunique" + assert revision == "0013_result_aware_agent_runs" assert "UNIQUE INDEX" in index_definition assert "(trial_id)" in index_definition assert "terminal_at IS NULL" in index_definition From e147f7b4860df5e3d13ad35435492fd1a6cac320 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:26:10 +0800 Subject: [PATCH 09/33] test: expect result-aware migration head --- tests/integration/test_attempt_physical_schema.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/test_attempt_physical_schema.py b/tests/integration/test_attempt_physical_schema.py index cb7d3d4..365ac62 100644 --- a/tests/integration/test_attempt_physical_schema.py +++ b/tests/integration/test_attempt_physical_schema.py @@ -50,7 +50,7 @@ async def test_migration_0010_adds_exact_attempt_physical_columns( "physical_boot_id", } } - assert revision == "0012_version_identity_nonunique" + assert revision == "0013_result_aware_agent_runs" assert physical == { "frozen_launch_spec_json": ("jsonb", None, "YES"), "execution_spec_hash": ("character", 64, "YES"), From d11ac328c75e55d6eb87cd045fa849274a92508a Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:26:25 +0800 Subject: [PATCH 10/33] test: expect result-aware migration head --- tests/integration/tools_policy/test_tool_persistence.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/tools_policy/test_tool_persistence.py b/tests/integration/tools_policy/test_tool_persistence.py index dce655a..e736c3c 100644 --- a/tests/integration/tools_policy/test_tool_persistence.py +++ b/tests/integration/tools_policy/test_tool_persistence.py @@ -31,7 +31,7 @@ def test_tool_tables_match_the_frozen_0008_contract(isolated_database_url: str) try: with engine.connect() as connection: assert connection.scalar(text("SELECT version_num FROM alembic_version")) == ( - "0012_version_identity_nonunique" + "0013_result_aware_agent_runs" ) tool_call_columns = set( connection.execute( From b74c02f970b32db99bdd1e28cb96358513cd69fa Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:26:37 +0800 Subject: [PATCH 11/33] test: include AgentRun source summary column --- tests/integration/agent_runtime/test_agent_service.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/integration/agent_runtime/test_agent_service.py b/tests/integration/agent_runtime/test_agent_service.py index 928dd86..2fed7de 100644 --- a/tests/integration/agent_runtime/test_agent_service.py +++ b/tests/integration/agent_runtime/test_agent_service.py @@ -156,6 +156,7 @@ def test_agent_tables_match_the_frozen_0008_contract( "total_tokens", "next_step_ordinal", "produced_proposal_revision_id", + "source_batch_summary_revision_id", "created_at", "started_at", "completed_at", From 7e0eb5da123d5e1a4e66b2a9c6b3e1ffd1786a40 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:26:53 +0800 Subject: [PATCH 12/33] test: cover AgentRun provenance migration contract --- tests/integration/db/test_migrations.py | 91 +++++++++++++++++++++++++ 1 file changed, 91 insertions(+) diff --git a/tests/integration/db/test_migrations.py b/tests/integration/db/test_migrations.py index 30a3b4b..432af87 100644 --- a/tests/integration/db/test_migrations.py +++ b/tests/integration/db/test_migrations.py @@ -48,6 +48,97 @@ async def test_upgrade_head_succeeds_on_an_isolated_empty_database( assert Path("migrations/versions/0013_result_aware_agent_runs.py").exists() + +@pytest.mark.integration +@pytest.mark.destructive +async def test_result_aware_agent_run_provenance_schema( + isolated_database_url: str, +) -> None: + sync_url = isolated_database_url.replace("+asyncpg", "+psycopg") + upgrade_database_to_head(isolated_database_url) + + engine = create_engine(sync_url) + try: + with engine.connect() as connection: + column = connection.execute( + text( + """ + SELECT data_type, is_nullable + FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = 'agent_runs' + AND column_name = 'source_batch_summary_revision_id' + """ + ) + ).mappings().one() + columns = set( + connection.scalars( + text( + """ + SELECT column_name + FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'agent_runs' + """ + ) + ) + ) + constraint = connection.scalar( + text( + """ + SELECT pg_get_constraintdef(oid) + FROM pg_constraint + WHERE conname = + 'fk_agent_runs_source_summary_revision_id_batch_summary_revisions' + """ + ) + ) + finally: + engine.dispose() + + assert dict(column) == {"data_type": "uuid", "is_nullable": "YES"} + assert "bound_experiment_id" not in columns + assert constraint is not None + assert ( + "FOREIGN KEY (source_batch_summary_revision_id) " + "REFERENCES batch_summary_revisions(batch_summary_revision_id) " + "ON DELETE RESTRICT" + ) in constraint + + +@pytest.mark.integration +@pytest.mark.destructive +async def test_result_aware_agent_run_migration_downgrade_removes_provenance_column( + isolated_database_url: str, +) -> None: + config = _alembic_config(isolated_database_url) + sync_url = isolated_database_url.replace("+asyncpg", "+psycopg") + engine = create_engine(sync_url) + try: + command.upgrade(config, "head") + command.downgrade(config, "0012_version_identity_nonunique") + with engine.connect() as connection: + assert ( + connection.scalar(text("SELECT version_num FROM alembic_version")) + == "0012_version_identity_nonunique" + ) + assert ( + connection.scalar( + text( + """ + SELECT count(*) + FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = 'agent_runs' + AND column_name = 'source_batch_summary_revision_id' + """ + ) + ) + == 0 + ) + finally: + engine.dispose() + + @pytest.mark.integration @pytest.mark.destructive async def test_experiment_version_identity_is_not_unique_after_head( From 799c34af03060aa8593f5c0d9b79cf6ab38b0d5e Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:27:20 +0800 Subject: [PATCH 13/33] test: expose null provenance for initial AgentRun --- tests/integration/api/test_slice1_api.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/integration/api/test_slice1_api.py b/tests/integration/api/test_slice1_api.py index 13916e8..814c288 100644 --- a/tests/integration/api/test_slice1_api.py +++ b/tests/integration/api/test_slice1_api.py @@ -151,7 +151,9 @@ async def _create(client: AsyncClient, snapshot_id: UUID) -> dict[str, object]: json={"repository_snapshot_id": str(snapshot_id), "goal": "Draft", "budget": BUDGET}, ) assert response.status_code == 201 - return response.json() + body = response.json() + assert body["source_batch_summary_revision_id"] is None + return body @pytest.mark.asyncio @@ -184,6 +186,7 @@ async def test_agent_trace_get_returns_persisted_steps( body = trace.json() assert body["agent_run_id"] == run["agent_run_id"] assert body["repository_snapshot_id"] == str(system.snapshot_id) + assert body["source_batch_summary_revision_id"] is None assert body["status"] == "completed" assert body["total_model_calls"] == 1 assert body["total_tokens"] == 2 From 4e253f922731ed38705f464a2810204ea13104b2 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:33:33 +0800 Subject: [PATCH 14/33] fix: shorten AgentRun source summary FK name --- migrations/versions/0013_result_aware_agent_runs.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/migrations/versions/0013_result_aware_agent_runs.py b/migrations/versions/0013_result_aware_agent_runs.py index 4e8af6c..e3153af 100644 --- a/migrations/versions/0013_result_aware_agent_runs.py +++ b/migrations/versions/0013_result_aware_agent_runs.py @@ -27,7 +27,7 @@ def upgrade() -> None: ), ) op.create_foreign_key( - "fk_agent_runs_source_summary_revision_id_batch_summary_revisions", + "fk_agent_runs_source_summary_revision", "agent_runs", "batch_summary_revisions", ["source_batch_summary_revision_id"], @@ -38,7 +38,7 @@ def upgrade() -> None: def downgrade() -> None: op.drop_constraint( - "fk_agent_runs_source_summary_revision_id_batch_summary_revisions", + "fk_agent_runs_source_summary_revision", "agent_runs", type_="foreignkey", ) From e83f6156c002b58345e1650dfa568171f0b9fc92 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:33:38 +0800 Subject: [PATCH 15/33] fix: shorten AgentRun source summary FK name --- src/labpilot/agent_runtime/models.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/labpilot/agent_runtime/models.py b/src/labpilot/agent_runtime/models.py index 7c43d15..c305080 100644 --- a/src/labpilot/agent_runtime/models.py +++ b/src/labpilot/agent_runtime/models.py @@ -40,7 +40,7 @@ class AgentRunRow(Base): ForeignKeyConstraint( ["source_batch_summary_revision_id"], ["batch_summary_revisions.batch_summary_revision_id"], - name="fk_agent_runs_source_summary_revision_id_batch_summary_revisions", + name="fk_agent_runs_source_summary_revision", ondelete="RESTRICT", ), PrimaryKeyConstraint("agent_run_id", name="pk_agent_runs"), From 51eb237af150e5ba01aca030b653b29f14c13818 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:33:45 +0800 Subject: [PATCH 16/33] fix: shorten AgentRun source summary FK name --- tests/integration/db/test_migrations.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/db/test_migrations.py b/tests/integration/db/test_migrations.py index 432af87..3415c89 100644 --- a/tests/integration/db/test_migrations.py +++ b/tests/integration/db/test_migrations.py @@ -88,7 +88,7 @@ async def test_result_aware_agent_run_provenance_schema( SELECT pg_get_constraintdef(oid) FROM pg_constraint WHERE conname = - 'fk_agent_runs_source_summary_revision_id_batch_summary_revisions' + 'fk_agent_runs_source_summary_revision' """ ) ) From 39cb9c3a1ac882c7c47fcc9fbb83b63dd0029b8e Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:38:52 +0800 Subject: [PATCH 17/33] test: define ResultContext compiler contract --- .../unit/agent_runtime/test_result_context.py | 226 ++++++++++++++++++ 1 file changed, 226 insertions(+) create mode 100644 tests/unit/agent_runtime/test_result_context.py diff --git a/tests/unit/agent_runtime/test_result_context.py b/tests/unit/agent_runtime/test_result_context.py new file mode 100644 index 0000000..cf78717 --- /dev/null +++ b/tests/unit/agent_runtime/test_result_context.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +import json +from dataclasses import replace +from uuid import uuid4 + +import pytest + +from labpilot.agent_runtime.context_compiler import ContextCompilationError, ContextCompiler +from labpilot.agent_runtime.result_context import ResultContext, ResultTrialCondition +from labpilot.context.repository_source import RepositorySourceDescriptor +from labpilot.experiments.proposal_types import ( + ArtifactDeclaration, + DatasetReference, + EnvironmentExpectation, + MetricDeclaration, + ProposalPayload, + ValidationRule, +) + + +def _descriptor() -> RepositorySourceDescriptor: + return RepositorySourceDescriptor( + repository_snapshot_id=uuid4(), + repository_snapshot_identity="b" * 64, + repository_registration_id=uuid4(), + project_id=uuid4(), + head_commit="c" * 40, + current_branch="main", + is_detached_head=False, + is_clean=True, + ) + + +def _proposal(snapshot: RepositorySourceDescriptor) -> ProposalPayload: + return ProposalPayload( + goal="Improve validation accuracy", + experiment_type="parameter_sweep", + repository_snapshot_id=snapshot.repository_snapshot_id, + repository_snapshot_identity=snapshot.repository_snapshot_identity, + train_entrypoint="train.py", + eval_entrypoint=None, + workdir=".", + parameter_matrix={"learning_rate": (0.001, 0.01)}, + seeds=(7,), + dataset_reference=DatasetReference( + reference_type="uri", + path_or_uri="s3://datasets/example", + stable_identity="dataset:v1", + reproducibility_level="strict", + ), + environment_expectation=EnvironmentExpectation( + python_profile="python-3.12", + framework_requirements=(), + device_requirement=None, + cuda_requirement=None, + ), + metrics=(MetricDeclaration("accuracy", "result", "test"),), + artifacts=(ArtifactDeclaration("result", "result.json", True),), + validation_rules=( + ValidationRule("artifact_exists", "result"), + ValidationRule("metric_finite", "accuracy"), + ), + timeout_seconds=30, + max_retries=1, + ) + + +def _result_context( + snapshot: RepositorySourceDescriptor, + *, + summary_revision_id=None, + numeric_value: str = "0.941", + warning_blob: str | None = None, +) -> ResultContext: + trial_id = uuid4() + included = { + "trial_id": str(trial_id), + "trial_ordinal": 1, + "selected_attempt_id": str(uuid4()), + "validation_report_id": str(uuid4()), + "metrics": [ + { + "metric_observation_id": str(uuid4()), + "artifact_id": str(uuid4()), + "metric_name": "accuracy", + "split": "test", + "numeric_value": numeric_value, + } + ], + } + warnings = () + if warning_blob is not None: + warnings = ({"code": "SYNTHETIC", "subject": "accuracy", "details": {"blob": warning_blob}},) + return ResultContext( + experiment_id=uuid4(), + source_experiment_version_id=uuid4(), + source_version_number=1, + source_proposal_revision_id=uuid4(), + source_batch_id=uuid4(), + source_batch_summary_revision_id=summary_revision_id or uuid4(), + source_summary_revision_number=1, + presentation_status="complete", + source_proposal=_proposal(snapshot), + trials=( + ResultTrialCondition( + trial_id=trial_id, + trial_ordinal=1, + resolved_parameters={"learning_rate": 0.01}, + seed=7, + ), + ), + included_results=(included,), + excluded_results=(), + warnings=warnings, + aggregates={"accuracy": {"count": 1, "mean": numeric_value}}, + ) + + +def _compile( + compiler: ContextCompiler, + snapshot: RepositorySourceDescriptor, + *, + result_context: ResultContext | None, +): + return compiler.compile( + provider_name="scripted", + model="test-model", + goal="decide the next experiment", + snapshot=snapshot, + definitions=(), + prior_tool_results=(), + prior_human_inputs=(), + result_context=result_context, + ) + + +def test_initial_context_none_preserves_existing_request_shape_and_hash() -> None: + compiler = ContextCompiler() + snapshot = _descriptor() + + implicit = compiler.compile( + provider_name="scripted", + model="test-model", + goal="decide the next experiment", + snapshot=snapshot, + definitions=(), + prior_tool_results=(), + prior_human_inputs=(), + ) + explicit = _compile(compiler, snapshot, result_context=None) + + assert explicit.request == implicit.request + assert explicit.canonical_request_hash == implicit.canonical_request_hash + user = json.loads(explicit.request.messages[1]["content"]) + assert "result_context" not in user + + +def test_follow_up_context_is_bounded_owned_input_and_hashes_exact_result_facts() -> None: + compiler = ContextCompiler() + snapshot = _descriptor() + summary_id = uuid4() + first_context = _result_context(snapshot, summary_revision_id=summary_id) + changed_metric = first_context.model_copy( + update={ + "included_results": ( + { + **first_context.included_results[0], + "metrics": [ + { + **first_context.included_results[0]["metrics"][0], + "numeric_value": "0.950", + } + ], + }, + ) + } + ) + changed_summary = first_context.model_copy( + update={"source_batch_summary_revision_id": uuid4()} + ) + + first = _compile(compiler, snapshot, result_context=first_context) + metric_changed = _compile(compiler, snapshot, result_context=changed_metric) + summary_changed = _compile(compiler, snapshot, result_context=changed_summary) + + user = json.loads(first.request.messages[1]["content"]) + assert user["result_context"]["source_batch_summary_revision_id"] == str(summary_id) + assert user["result_context"]["included_results"][0]["metrics"][0]["numeric_value"] == "0.941" + assert first.canonical_request_hash != metric_changed.canonical_request_hash + assert first.canonical_request_hash != summary_changed.canonical_request_hash + + +def test_follow_up_system_policy_preserves_result_authority_and_governance() -> None: + compiler = ContextCompiler() + snapshot = _descriptor() + + compiled = _compile(compiler, snapshot, result_context=_result_context(snapshot)) + system = json.loads(compiled.request.messages[0]["content"]) + policy = "\n".join(system["result_context_policy"]) + + assert "selected" in policy and "eligible" in policy + assert "exclusion" in policy + assert "invalid raw" in policy + assert "stop" in policy + assert "full ProposalPayload" in policy + assert "start" in policy and "approve" in policy + + +def test_result_context_has_individual_128_kib_bound_before_model_request() -> None: + compiler = ContextCompiler() + snapshot = _descriptor() + context = _result_context(snapshot, warning_blob="x" * (129 * 1024)) + + with pytest.raises(ContextCompilationError, match="result_context_too_large"): + _compile(compiler, snapshot, result_context=context) + + +def test_result_context_models_reject_extra_fields() -> None: + snapshot = _descriptor() + context = _result_context(snapshot) + + with pytest.raises(Exception): + ResultContext.model_validate( + {**context.model_dump(mode="json"), "unexpected": True} + ) From cfbde9a6a7ebe07530221dffe1cfd34d9566e78b Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:39:15 +0800 Subject: [PATCH 18/33] feat: add Agent-owned ResultContext contract --- src/labpilot/agent_runtime/result_context.py | 36 ++++++++++++++++++++ 1 file changed, 36 insertions(+) create mode 100644 src/labpilot/agent_runtime/result_context.py diff --git a/src/labpilot/agent_runtime/result_context.py b/src/labpilot/agent_runtime/result_context.py new file mode 100644 index 0000000..55eee7b --- /dev/null +++ b/src/labpilot/agent_runtime/result_context.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, JsonValue + +from labpilot.experiments.proposal_types import ProposalPayload + + +class ResultTrialCondition(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + trial_id: UUID + trial_ordinal: int = Field(gt=0) + resolved_parameters: dict[str, JsonValue] + seed: int + + +class ResultContext(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + experiment_id: UUID + source_experiment_version_id: UUID + source_version_number: int = Field(gt=0) + source_proposal_revision_id: UUID + source_batch_id: UUID + source_batch_summary_revision_id: UUID + source_summary_revision_number: int = Field(gt=0) + presentation_status: Literal["complete", "complete_with_invalid_results"] + source_proposal: ProposalPayload + trials: tuple[ResultTrialCondition, ...] + included_results: tuple[dict[str, JsonValue], ...] + excluded_results: tuple[dict[str, JsonValue], ...] + warnings: tuple[dict[str, JsonValue], ...] + aggregates: dict[str, JsonValue] From 9ab38b77157d0c9190db4ca7e6a07daf4027bebe Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:39:33 +0800 Subject: [PATCH 19/33] feat: compile bounded ResultContext for follow-up runs --- .../agent_runtime/context_compiler.py | 27 ++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/src/labpilot/agent_runtime/context_compiler.py b/src/labpilot/agent_runtime/context_compiler.py index 8da16a3..b275d78 100644 --- a/src/labpilot/agent_runtime/context_compiler.py +++ b/src/labpilot/agent_runtime/context_compiler.py @@ -9,6 +9,7 @@ from pydantic import TypeAdapter from labpilot.agent_runtime.provider import ModelRequest +from labpilot.agent_runtime.result_context import ResultContext from labpilot.agent_runtime.types import AgentDecision from labpilot.common.canonical_json import CanonicalValue, canonical_json_bytes from labpilot.common.hashing import namespaced_sha256 @@ -23,6 +24,7 @@ _MAX_PRIOR_RESULTS = 20 _MAX_TOOL_RESULT_BYTES = 64 * 1024 _MAX_HUMAN_INPUT_BYTES = 4000 +_MAX_RESULT_CONTEXT_BYTES = 128 * 1024 _MAX_MESSAGE_BYTES = 256 * 1024 @@ -54,6 +56,7 @@ def compile( definitions: tuple[ToolDefinition, ...], prior_tool_results: tuple[PriorToolResult, ...], prior_human_inputs: tuple[str, ...], + result_context: ResultContext | None = None, ) -> CompiledContext: _require_utf8(goal, field="goal", max_bytes=_MAX_GOAL_BYTES, non_blank=True) if len(definitions) > _MAX_TOOL_COUNT: @@ -120,6 +123,23 @@ def compile( "agent_decision_schema": output_schema_value, "tools": tools, } + if result_context is not None: + system_value["result_context_policy"] = [ + ( + "Treat selected and eligible evidence in result_context as authoritative " + "for the deterministic facts it contains." + ), + ( + "Use exclusion reason codes to diagnose rejected Trials, but do not " + "promote invalid raw results into accepted scientific evidence." + ), + "If the declared research goal is already satisfied, stop is a valid decision.", + ( + "If another experiment is warranted, propose must include the full " + "ProposalPayload required by the existing proposal contract." + ), + "Never start or approve the next experiment from the Agent decision.", + ] prior_results: list[CanonicalValue] = [] for item in prior_tool_results: value: CanonicalValue = { @@ -138,7 +158,7 @@ def compile( non_blank=True, ) human_inputs.append(value) - user_value: CanonicalValue = { + user_value: dict[str, CanonicalValue] = { "goal": goal, "snapshot": { "repository_snapshot_id": str(snapshot.repository_snapshot_id), @@ -153,6 +173,11 @@ def compile( "prior_tool_results": prior_results, "prior_human_inputs": human_inputs, } + if result_context is not None: + result_value = _canonical_value(result_context.model_dump(mode="json")) + if len(canonical_json_bytes(result_value)) > _MAX_RESULT_CONTEXT_BYTES: + raise ContextCompilationError("result_context_too_large") + user_value["result_context"] = result_value messages = ( {"role": "system", "content": _render(system_value)}, {"role": "user", "content": _render(user_value)}, From fbab4435fe3c39dec319dcd69e545c2e2ca5cd89 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:40:04 +0800 Subject: [PATCH 20/33] test: cover ResultContext global message bound --- .../unit/agent_runtime/test_result_context.py | 21 +++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/tests/unit/agent_runtime/test_result_context.py b/tests/unit/agent_runtime/test_result_context.py index cf78717..0781563 100644 --- a/tests/unit/agent_runtime/test_result_context.py +++ b/tests/unit/agent_runtime/test_result_context.py @@ -1,10 +1,10 @@ from __future__ import annotations import json -from dataclasses import replace from uuid import uuid4 import pytest +from pydantic import ValidationError from labpilot.agent_runtime.context_compiler import ContextCompilationError, ContextCompiler from labpilot.agent_runtime.result_context import ResultContext, ResultTrialCondition @@ -216,11 +216,28 @@ def test_result_context_has_individual_128_kib_bound_before_model_request() -> N _compile(compiler, snapshot, result_context=context) +def test_total_compiled_message_bound_still_applies_with_result_context() -> None: + compiler = ContextCompiler() + snapshot = _descriptor() + + with pytest.raises(ContextCompilationError, match="compiled_messages_too_large"): + compiler.compile( + provider_name="scripted", + model="test-model", + goal="decide the next experiment", + snapshot=snapshot, + definitions=(), + prior_tool_results=(), + prior_human_inputs=("h" * 4000,) * 70, + result_context=_result_context(snapshot), + ) + + def test_result_context_models_reject_extra_fields() -> None: snapshot = _descriptor() context = _result_context(snapshot) - with pytest.raises(Exception): + with pytest.raises(ValidationError): ResultContext.model_validate( {**context.model_dump(mode="json"), "unexpected": True} ) From c49dbb6f205ed050e684d173ea113c89bd539fe6 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:41:10 +0800 Subject: [PATCH 21/33] style: format ResultContext test fixture --- tests/unit/agent_runtime/test_result_context.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tests/unit/agent_runtime/test_result_context.py b/tests/unit/agent_runtime/test_result_context.py index 0781563..9a61b0a 100644 --- a/tests/unit/agent_runtime/test_result_context.py +++ b/tests/unit/agent_runtime/test_result_context.py @@ -91,7 +91,13 @@ def _result_context( } warnings = () if warning_blob is not None: - warnings = ({"code": "SYNTHETIC", "subject": "accuracy", "details": {"blob": warning_blob}},) + warnings = ( + { + "code": "SYNTHETIC", + "subject": "accuracy", + "details": {"blob": warning_blob}, + }, + ) return ResultContext( experiment_id=uuid4(), source_experiment_version_id=uuid4(), From 8c1834da7cafa39e729fd393637ca071aeef919d Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:42:32 +0800 Subject: [PATCH 22/33] fix: narrow compiled system context type --- src/labpilot/agent_runtime/context_compiler.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/labpilot/agent_runtime/context_compiler.py b/src/labpilot/agent_runtime/context_compiler.py index b275d78..9fd8d37 100644 --- a/src/labpilot/agent_runtime/context_compiler.py +++ b/src/labpilot/agent_runtime/context_compiler.py @@ -68,7 +68,7 @@ def compile( output_schema_value = _canonical_value(output_schema) output_schema_hash = namespaced_sha256(_OUTPUT_SCHEMA_NAMESPACE, output_schema_value) tools = [_tool_contract(definition) for definition in definitions] - system_value: CanonicalValue = { + system_value: dict[str, CanonicalValue] = { "instruction": ( "Return exactly one JSON object matching agent_decision_schema. " "Do not include rationale, reasoning, or chain_of_thought." From 266f67a978f2a27a24b07018c7cdef75a01587ca Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:44:29 +0800 Subject: [PATCH 23/33] fix: serialize frozen Proposal mappings in ResultContext --- src/labpilot/agent_runtime/context_compiler.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/src/labpilot/agent_runtime/context_compiler.py b/src/labpilot/agent_runtime/context_compiler.py index 9fd8d37..483a02b 100644 --- a/src/labpilot/agent_runtime/context_compiler.py +++ b/src/labpilot/agent_runtime/context_compiler.py @@ -174,7 +174,13 @@ def compile( "prior_human_inputs": human_inputs, } if result_context is not None: - result_value = _canonical_value(result_context.model_dump(mode="json")) + result_value = _canonical_value( + result_context.model_dump( + mode="json", + warnings="none", + fallback=_serialization_fallback, + ) + ) if len(canonical_json_bytes(result_value)) > _MAX_RESULT_CONTEXT_BYTES: raise ContextCompilationError("result_context_too_large") user_value["result_context"] = result_value @@ -239,6 +245,12 @@ def _render(value: CanonicalValue) -> str: return canonical_json_bytes(value).decode("utf-8") +def _serialization_fallback(value: object) -> object: + if isinstance(value, Mapping): + return dict(value) + raise TypeError(f"unsupported result context value: {type(value).__name__}") + + def _require_utf8(value: str, *, field: str, max_bytes: int, non_blank: bool) -> None: if non_blank and not value.strip(): raise ContextCompilationError(f"{field}_blank") From b17ff797c380af841d54bc2f285d15dd407fe5ce Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:44:45 +0800 Subject: [PATCH 24/33] test: serialize frozen Proposal mapping for validation case --- tests/unit/agent_runtime/test_result_context.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tests/unit/agent_runtime/test_result_context.py b/tests/unit/agent_runtime/test_result_context.py index 9a61b0a..263cc0d 100644 --- a/tests/unit/agent_runtime/test_result_context.py +++ b/tests/unit/agent_runtime/test_result_context.py @@ -245,5 +245,11 @@ def test_result_context_models_reject_extra_fields() -> None: with pytest.raises(ValidationError): ResultContext.model_validate( - {**context.model_dump(mode="json"), "unexpected": True} + { + **context.model_dump( + mode="json", + fallback=lambda value: dict(value), + ), + "unexpected": True, + } ) From c817ed752bcfedb1d0941fade75306d3f7132b66 Mon Sep 17 00:00:00 2001 From: Air000000 <1113476369@qq.com> Date: Sun, 20 Sep 2026 15:37:48 +0800 Subject: [PATCH 25/33] feat: add result-aware follow-up binding --- src/labpilot/agent_runtime/repository.py | 10 +- src/labpilot/agent_runtime/result_context.py | 11 +- .../api/result_aware_agent_adapter.py | 189 ++++++++++++ .../test_follow_up_run_binding.py | 282 ++++++++++++++++++ 4 files changed, 488 insertions(+), 4 deletions(-) create mode 100644 src/labpilot/api/result_aware_agent_adapter.py create mode 100644 tests/integration/agent_runtime/test_follow_up_run_binding.py diff --git a/src/labpilot/agent_runtime/repository.py b/src/labpilot/agent_runtime/repository.py index f36d397..3decf09 100644 --- a/src/labpilot/agent_runtime/repository.py +++ b/src/labpilot/agent_runtime/repository.py @@ -83,11 +83,15 @@ def __init__(self, session_context_factory: SessionContextFactory) -> None: self._session_context_factory = session_context_factory async def add_run(self, run: AgentRunDetails) -> AgentRunDetails: - row = _run_row(run) async with UnitOfWork(self._session_context_factory) as uow: - uow.session.add(row) - await uow.session.flush() + details = await self.add_run_in_uow(uow, run) await uow.commit() + return details + + async def add_run_in_uow(self, uow: UnitOfWork, run: AgentRunDetails) -> AgentRunDetails: + row = _run_row(run) + uow.session.add(row) + await uow.session.flush() return _run_details(row) async def get_run(self, agent_run_id: UUID) -> AgentRunDetails | None: diff --git a/src/labpilot/agent_runtime/result_context.py b/src/labpilot/agent_runtime/result_context.py index 55eee7b..81cadd6 100644 --- a/src/labpilot/agent_runtime/result_context.py +++ b/src/labpilot/agent_runtime/result_context.py @@ -1,10 +1,11 @@ from __future__ import annotations -from typing import Literal +from typing import Literal, Protocol from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, JsonValue +from labpilot.agent_runtime.types import AgentRunDetails from labpilot.experiments.proposal_types import ProposalPayload @@ -34,3 +35,11 @@ class ResultContext(BaseModel): excluded_results: tuple[dict[str, JsonValue], ...] warnings: tuple[dict[str, JsonValue], ...] aggregates: dict[str, JsonValue] + + +class FollowUpRunStore(Protocol): + async def create_bound_run(self, run: AgentRunDetails) -> AgentRunDetails: ... + + +class ResultContextSource(Protocol): + async def load(self, source_batch_summary_revision_id: UUID) -> ResultContext: ... diff --git a/src/labpilot/api/result_aware_agent_adapter.py b/src/labpilot/api/result_aware_agent_adapter.py new file mode 100644 index 0000000..adcff5b --- /dev/null +++ b/src/labpilot/api/result_aware_agent_adapter.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +from collections.abc import Iterable +from typing import cast +from uuid import UUID + +from pydantic import JsonValue, ValidationError +from sqlalchemy import func, select + +from labpilot.agent_runtime.repository import SqlAgentRepository +from labpilot.agent_runtime.result_context import ( + FollowUpRunStore, + ResultContext, + ResultContextSource, + ResultTrialCondition, +) +from labpilot.agent_runtime.types import AgentRunDetails, AgentRuntimeError +from labpilot.context.models import RepositoryRegistrationRow, RepositorySnapshotRow +from labpilot.db.uow import SessionContextFactory, UnitOfWork +from labpilot.execution.models import BatchRow, TrialRow +from labpilot.experiments.proposal_models import ( + ExperimentRow, + ExperimentVersionRow, + ProposalRevisionRow, +) +from labpilot.experiments.proposal_service import _payload_from_json +from labpilot.results_validation.models import BatchSummaryRevisionRow + + +class SqlResultAwareAgentAdapter(FollowUpRunStore, ResultContextSource): + def __init__( + self, + session_context_factory: SessionContextFactory, + *, + agent_repository: SqlAgentRepository, + ) -> None: + self._session_context_factory = session_context_factory + self._agent_repository = agent_repository + + async def create_bound_run(self, run: AgentRunDetails) -> AgentRunDetails: + summary_id = run.source_batch_summary_revision_id + if summary_id is None: + raise AgentRuntimeError("source_result_lineage_invalid") + async with UnitOfWork(self._session_context_factory) as uow: + initial = await uow.session.get(BatchSummaryRevisionRow, summary_id) + if initial is None: + raise AgentRuntimeError("source_summary_not_found") + batch = await uow.session.scalar( + select(BatchRow).where(BatchRow.batch_id == initial.batch_id).with_for_update() + ) + if batch is None: + raise AgentRuntimeError("source_batch_not_found") + summary = await uow.session.scalar( + select(BatchSummaryRevisionRow).where( + BatchSummaryRevisionRow.batch_summary_revision_id == summary_id + ) + ) + if summary is None or summary.batch_id != batch.batch_id: + raise AgentRuntimeError("source_result_lineage_invalid") + latest = await uow.session.scalar( + select(func.max(BatchSummaryRevisionRow.revision_number)).where( + BatchSummaryRevisionRow.batch_id == batch.batch_id + ) + ) + if latest is None or summary.revision_number != latest: + raise AgentRuntimeError("source_summary_not_latest") + version = await uow.session.get(ExperimentVersionRow, batch.experiment_version_id) + if version is None: + raise AgentRuntimeError("source_result_lineage_invalid") + proposal = await uow.session.get(ProposalRevisionRow, version.proposal_revision_id) + experiment = await uow.session.get(ExperimentRow, version.experiment_id) + snapshot = await uow.session.get(RepositorySnapshotRow, run.repository_snapshot_id) + if proposal is None or experiment is None or snapshot is None: + raise AgentRuntimeError("source_result_lineage_invalid") + registration = await uow.session.get( + RepositoryRegistrationRow, snapshot.repository_registration_id + ) + if registration is None: + raise AgentRuntimeError("source_result_lineage_invalid") + if ( + batch.project_id != experiment.project_id + or experiment.project_id != registration.project_id + or registration.project_id != run.project_id + ): + raise AgentRuntimeError("source_project_mismatch") + if not snapshot.is_clean: + raise AgentRuntimeError("source_snapshot_not_clean") + if ( + run.repository_snapshot_id != version.repository_snapshot_id + or snapshot.repository_snapshot_identity != version.repository_snapshot_identity + ): + raise AgentRuntimeError("source_snapshot_mismatch") + if ( + proposal.experiment_id != experiment.experiment_id + or version.experiment_id != experiment.experiment_id + ): + raise AgentRuntimeError("source_result_lineage_invalid") + details = await self._agent_repository.add_run_in_uow(uow, run) + await uow.commit() + return details + + async def load(self, source_batch_summary_revision_id: UUID) -> ResultContext: + async with self._session_context_factory() as session: + summary = await session.get(BatchSummaryRevisionRow, source_batch_summary_revision_id) + if summary is None: + raise AgentRuntimeError("source_summary_not_found") + batch = await session.get(BatchRow, summary.batch_id) + if batch is None: + raise AgentRuntimeError("source_result_lineage_invalid") + version = await session.get(ExperimentVersionRow, batch.experiment_version_id) + if version is None: + raise AgentRuntimeError("source_result_lineage_invalid") + proposal = await session.get(ProposalRevisionRow, version.proposal_revision_id) + experiment = await session.get(ExperimentRow, version.experiment_id) + if proposal is None or experiment is None: + raise AgentRuntimeError("source_result_lineage_invalid") + if ( + proposal.experiment_id != experiment.experiment_id + or version.experiment_id != experiment.experiment_id + ): + raise AgentRuntimeError("source_result_lineage_invalid") + trials = tuple( + ( + await session.scalars( + select(TrialRow) + .where(TrialRow.batch_id == batch.batch_id) + .order_by(TrialRow.trial_ordinal) + ) + ).all() + ) + included = _json_objects(summary.included_result_refs_json) + excluded = _json_objects(summary.excluded_reasons_json) + warnings = _json_objects(summary.warning_json.get("warnings", [])) + _require_exact_trial_coverage(trials, included=included, excluded=excluded) + try: + return ResultContext( + experiment_id=experiment.experiment_id, + source_experiment_version_id=version.experiment_version_id, + source_version_number=version.version_number, + source_proposal_revision_id=proposal.proposal_revision_id, + source_batch_id=batch.batch_id, + source_batch_summary_revision_id=summary.batch_summary_revision_id, + source_summary_revision_number=summary.revision_number, + presentation_status=("complete_with_invalid_results" if excluded else "complete"), + source_proposal=_payload_from_json(proposal.payload_json), + trials=tuple( + ResultTrialCondition( + trial_id=trial.trial_id, + trial_ordinal=trial.trial_ordinal, + resolved_parameters=cast( + dict[str, JsonValue], trial.resolved_parameters_json + ), + seed=trial.seed, + ) + for trial in trials + ), + included_results=cast(tuple[dict[str, JsonValue], ...], included), + excluded_results=cast(tuple[dict[str, JsonValue], ...], excluded), + warnings=cast(tuple[dict[str, JsonValue], ...], warnings), + aggregates=cast(dict[str, JsonValue], summary.aggregates_json), + ) + except (TypeError, ValidationError, ValueError) as exc: + raise AgentRuntimeError("source_result_lineage_invalid") from exc + + +def _json_objects(values: object) -> tuple[dict[str, object], ...]: + if not isinstance(values, list) or any(not isinstance(value, dict) for value in values): + raise AgentRuntimeError("source_result_lineage_invalid") + return tuple(dict(cast(dict[str, object], value)) for value in values) + + +def _require_exact_trial_coverage( + trials: Iterable[TrialRow], + *, + included: tuple[dict[str, object], ...], + excluded: tuple[dict[str, object], ...], +) -> None: + expected = {trial.trial_id for trial in trials} + covered: list[UUID] = [] + for result in (*included, *excluded): + trial_id = result.get("trial_id") + if not isinstance(trial_id, str): + raise AgentRuntimeError("source_result_lineage_invalid") + try: + covered.append(UUID(trial_id)) + except ValueError as exc: + raise AgentRuntimeError("source_result_lineage_invalid") from exc + if len(covered) != len(expected) or set(covered) != expected: + raise AgentRuntimeError("source_result_lineage_invalid") diff --git a/tests/integration/agent_runtime/test_follow_up_run_binding.py b/tests/integration/agent_runtime/test_follow_up_run_binding.py new file mode 100644 index 0000000..f940714 --- /dev/null +++ b/tests/integration/agent_runtime/test_follow_up_run_binding.py @@ -0,0 +1,282 @@ +from __future__ import annotations + +import asyncio +import importlib +from copy import deepcopy +from dataclasses import replace +from datetime import UTC, datetime +from decimal import Decimal +from pathlib import Path +from uuid import UUID, uuid4 + +import pytest +from sqlalchemy import func, select +from tests.integration.test_batch_summary_service import _process +from tests.integration.test_result_processing_service import _seed_result_attempt + +from labpilot.agent_runtime.models import AgentRunRow +from labpilot.agent_runtime.repository import SqlAgentRepository +from labpilot.agent_runtime.types import AgentRunDetails, AgentRuntimeError +from labpilot.api.result_aware_agent_adapter import SqlResultAwareAgentAdapter +from labpilot.context.models import RepositorySnapshotRow +from labpilot.coordinator.execution_adapters import VersionExecutionSemanticsAdapter +from labpilot.db.uow import UnitOfWork +from labpilot.execution.models import BatchRow +from labpilot.execution.trial_outcome import TrialExecutionOutcomeService +from labpilot.experiments.proposal_models import ExperimentVersionRow +from labpilot.results_validation.models import ( + BatchSummaryRevisionRow, + MetricObservationRow, + ValidationReportRow, +) + + +def _adapter(system: object) -> SqlResultAwareAgentAdapter: + database = system.database + return SqlResultAwareAgentAdapter( + database.session, agent_repository=SqlAgentRepository(database.session) + ) + + +async def _refresh(system: object, batch_id: UUID) -> object: + summary = importlib.import_module("labpilot.results_validation.summary_service") + database = system.database + outcomes = TrialExecutionOutcomeService( + database.session, version_semantics=VersionExecutionSemanticsAdapter(database.session) + ) + return await summary.BatchSummaryService( + database.session, trial_execution_outcome_service=outcomes + ).refresh_summary(batch_id) + + +async def _state( + system: object, batch_id: UUID +) -> tuple[BatchSummaryRevisionRow, BatchRow, ExperimentVersionRow]: + async with system.database.session() as session: + summary = await session.scalar( + select(BatchSummaryRevisionRow) + .where(BatchSummaryRevisionRow.batch_id == batch_id) + .order_by(BatchSummaryRevisionRow.revision_number.desc()) + ) + batch = await session.get(BatchRow, batch_id) + assert summary is not None and batch is not None + version = await session.get(ExperimentVersionRow, batch.experiment_version_id) + assert version is not None + return summary, batch, version + + +def _run( + summary: BatchSummaryRevisionRow, batch: BatchRow, version: ExperimentVersionRow +) -> AgentRunDetails: + return AgentRunDetails( + uuid4(), + batch.project_id, + version.repository_snapshot_id, + "Improve the experiment.", + "ready", + None, + "local_only", + "test-provider", + "test-model", + 1, + 1, + 1, + 1, + 0, + 0, + 1, + None, + datetime.now(UTC), + None, + None, + 1, + summary.batch_summary_revision_id, + ) + + +async def _count(system: object) -> int: + async with system.database.session() as session: + result = await session.scalar(select(func.count()).select_from(AgentRunRow)) + assert result is not None + return result + + +async def _change_metric(system: object) -> None: + async with system.database.session() as session: + metric = await session.scalar( + select(MetricObservationRow).order_by(MetricObservationRow.created_at) + ) + report = await session.scalar( + select(ValidationReportRow).order_by(ValidationReportRow.created_at) + ) + assert metric is not None and report is not None + snapshot = deepcopy(report.input_snapshot_json) + metrics = snapshot["metrics"] + assert isinstance(metrics, list) and isinstance(metrics[0], dict) + metrics[0]["numeric_value"] = "0.83" + report.input_snapshot_json = snapshot + metric.numeric_value = Decimal("0.83") + await session.commit() + + +@pytest.mark.integration +async def test_missing_and_stale_summary_write_no_follow_up_run( + isolated_database_url: str, tmp_path: Path +) -> None: + async with _seed_result_attempt(isolated_database_url, tmp_path) as ( + system, + batch_id, + _trial_id, + attempt_id, + _workspace, + ): + await _process(system.database, attempt_id) + await _refresh(system, batch_id) + summary, batch, version = await _state(system, batch_id) + run, adapter = _run(summary, batch, version), _adapter(system) + with pytest.raises(AgentRuntimeError, match="source_summary_not_found"): + await adapter.create_bound_run(replace(run, source_batch_summary_revision_id=uuid4())) + await _change_metric(system) + await _refresh(system, batch_id) + with pytest.raises(AgentRuntimeError, match="source_summary_not_latest"): + await adapter.create_bound_run(run) + assert await _count(system) == 0 + + +@pytest.mark.integration +async def test_batch_lock_linearizes_refresh_before_follow_up( + isolated_database_url: str, tmp_path: Path +) -> None: + async with _seed_result_attempt(isolated_database_url, tmp_path) as ( + system, + batch_id, + _trial_id, + attempt_id, + _workspace, + ): + await _process(system.database, attempt_id) + await _refresh(system, batch_id) + summary, batch, version = await _state(system, batch_id) + await _change_metric(system) + async with UnitOfWork(system.database.session) as lock: + await lock.session.scalar( + select(BatchRow).where(BatchRow.batch_id == batch_id).with_for_update() + ) + refresh_task = asyncio.create_task(_refresh(system, batch_id)) + await asyncio.sleep(0.05) + follow_up_task = asyncio.create_task( + _adapter(system).create_bound_run(_run(summary, batch, version)) + ) + await asyncio.sleep(0.05) + assert not refresh_task.done() and not follow_up_task.done() + await lock.commit() + assert (await refresh_task).created is True + with pytest.raises(AgentRuntimeError, match="source_summary_not_latest"): + await follow_up_task + assert await _count(system) == 0 + + +@pytest.mark.integration +async def test_bound_run_pins_exact_summary_not_later_raw_result_rows( + isolated_database_url: str, tmp_path: Path +) -> None: + async with _seed_result_attempt(isolated_database_url, tmp_path) as ( + system, + batch_id, + _trial_id, + attempt_id, + _workspace, + ): + await _process(system.database, attempt_id) + await _refresh(system, batch_id) + summary, batch, version = await _state(system, batch_id) + created = await _adapter(system).create_bound_run(_run(summary, batch, version)) + await _change_metric(system) + await _refresh(system, batch_id) + context = await _adapter(system).load(summary.batch_summary_revision_id) + assert created.source_batch_summary_revision_id == summary.batch_summary_revision_id + assert context.source_batch_summary_revision_id == summary.batch_summary_revision_id + assert context.source_summary_revision_number == 1 + assert context.presentation_status == "complete" + assert context.aggregates == {"score": {"count": 1, "mean": "0.82"}} + assert "0.83" not in str(context.included_results) + + +@pytest.mark.integration +@pytest.mark.parametrize("mutation", ("project", "dirty", "snapshot_id", "snapshot_identity")) +async def test_bound_run_rejects_project_and_snapshot_mismatches( + isolated_database_url: str, tmp_path: Path, mutation: str +) -> None: + async with _seed_result_attempt(isolated_database_url, tmp_path) as ( + system, + batch_id, + _trial_id, + attempt_id, + _workspace, + ): + await _process(system.database, attempt_id) + await _refresh(system, batch_id) + summary, batch, version = await _state(system, batch_id) + run = _run(summary, batch, version) + expected = { + "project": "source_project_mismatch", + "dirty": "source_snapshot_not_clean", + }.get(mutation, "source_snapshot_mismatch") + if mutation == "project": + run = replace(run, project_id=uuid4()) + else: + async with system.database.session() as session: + current = await session.get(ExperimentVersionRow, version.experiment_version_id) + snapshot = await session.get(RepositorySnapshotRow, version.repository_snapshot_id) + assert current is not None and snapshot is not None + if mutation == "dirty": + snapshot.is_clean = False + elif mutation == "snapshot_id": + current.repository_snapshot_id = uuid4() + else: + current.repository_snapshot_identity = "0" * 64 + await session.commit() + with pytest.raises(AgentRuntimeError, match=expected): + await _adapter(system).create_bound_run(run) + assert await _count(system) == 0 + + +@pytest.mark.integration +async def test_context_derives_invalid_presentation_from_persisted_summary( + isolated_database_url: str, tmp_path: Path +) -> None: + async with _seed_result_attempt( + isolated_database_url, + tmp_path, + output=b'{"schema_version":"v1","metrics":{"score":0.61}}', + ) as (system, batch_id, _trial_id, attempt_id, _workspace): + await _process(system.database, attempt_id) + await _refresh(system, batch_id) + summary, _batch, _version = await _state(system, batch_id) + context = await _adapter(system).load(summary.batch_summary_revision_id) + + assert context.presentation_status == "complete_with_invalid_results" + assert context.included_results == () + assert context.excluded_results == tuple(summary.excluded_reasons_json) + assert context.aggregates == summary.aggregates_json + + +@pytest.mark.integration +async def test_context_uses_persisted_summary_warnings_and_aggregates( + isolated_database_url: str, tmp_path: Path +) -> None: + async with _seed_result_attempt(isolated_database_url, tmp_path, best_effort_dataset=True) as ( + system, + batch_id, + _trial_id, + attempt_id, + _workspace, + ): + await _process(system.database, attempt_id) + await _refresh(system, batch_id) + summary, _batch, _version = await _state(system, batch_id) + context = await _adapter(system).load(summary.batch_summary_revision_id) + + assert context.presentation_status == "complete" + assert context.warnings == tuple(summary.warning_json["warnings"]) + assert context.aggregates == summary.aggregates_json From e3116ecb7f5799e550317a02ea35d868ffbb23b0 Mon Sep 17 00:00:00 2001 From: Air000000 <1113476369@qq.com> Date: Mon, 21 Sep 2026 16:15:56 +0800 Subject: [PATCH 26/33] feat: add follow-up AgentRun service API --- src/labpilot/agent_runtime/service.py | 66 ++++ src/labpilot/api/routers/agent_runs.py | 27 ++ src/labpilot/api/schemas/agent_runs.py | 9 + src/labpilot/main.py | 10 +- .../api/test_result_aware_agent_api.py | 233 +++++++++++++ .../unit/agent_runtime/test_agent_service.py | 315 +++++++++++++++++- 6 files changed, 646 insertions(+), 14 deletions(-) create mode 100644 tests/integration/api/test_result_aware_agent_api.py diff --git a/src/labpilot/agent_runtime/service.py b/src/labpilot/agent_runtime/service.py index 9337d94..5207d64 100644 --- a/src/labpilot/agent_runtime/service.py +++ b/src/labpilot/agent_runtime/service.py @@ -21,6 +21,10 @@ assert_provider_egress, ) from labpilot.agent_runtime.repository import AgentRepository +from labpilot.agent_runtime.result_context import ( + FollowUpRunStore, + ResultContextSource, +) from labpilot.agent_runtime.types import ( AgentBudget, AgentDecision, @@ -95,6 +99,8 @@ def __init__( tool_service_factory: AgentToolServiceFactory, proposal_service: AgentProposalService, data_egress_policy: DataEgressPolicy, + follow_up_run_store: FollowUpRunStore | None = None, + result_context_source: ResultContextSource | None = None, ) -> None: self._repository = repository self._repository_source = repository_source @@ -104,6 +110,8 @@ def __init__( self._tool_service_factory = tool_service_factory self._proposals = proposal_service self._egress_policy = data_egress_policy + self._follow_up_run_store = follow_up_run_store + self._result_context_source = result_context_source self._decision_adapter: TypeAdapter[AgentDecision] = TypeAdapter(AgentDecision) async def create_run( @@ -149,6 +157,48 @@ async def create_run( _log("agent_run_created", context, stored) return stored + async def create_follow_up_run( + self, + *, + context: CommandContext, + project_id: UUID, + source_batch_summary_revision_id: UUID, + repository_snapshot_id: UUID, + goal: str, + budget: AgentBudget, + ) -> AgentRunDetails: + _require_text(goal, "goal", 4000) + if self._follow_up_run_store is None or self._result_context_source is None: + raise AgentRuntimeError("follow_up_runtime_unconfigured") + now = datetime.now(UTC) + run = AgentRunDetails( + agent_run_id=uuid4(), + project_id=project_id, + repository_snapshot_id=repository_snapshot_id, + goal=goal, + status="ready", + termination_reason=None, + egress_policy=self._egress_policy, + provider_name=self._provider.provider_name, + model_name=self._model_name, + max_steps=budget.max_steps, + max_model_calls=budget.max_model_calls, + max_wall_clock_seconds=budget.max_wall_clock_seconds, + max_tokens=budget.max_tokens, + total_model_calls=0, + total_tokens=0, + next_step_ordinal=1, + produced_proposal_revision_id=None, + created_at=now, + started_at=None, + completed_at=None, + row_version=1, + source_batch_summary_revision_id=source_batch_summary_revision_id, + ) + stored = await self._follow_up_run_store.create_bound_run(run) + _log("follow_up_agent_run_created", context, stored) + return stored + async def advance_run( self, *, @@ -163,6 +213,8 @@ async def advance_run( raise AgentRuntimeError("agent_run_terminal") if run.status == "awaiting_human": raise AgentRuntimeError("human_input_required") + if run.source_batch_summary_revision_id is not None and self._result_context_source is None: + raise AgentRuntimeError("follow_up_runtime_unconfigured") steps = await self._repository.list_steps(agent_run_id) if steps and steps[-1].outcome == "reserved": return await self._repository.finalize_model_step( @@ -196,6 +248,12 @@ async def advance_run( except ProviderPolicyError as exc: raise AgentRuntimeError(exc.reason_code) from None descriptor = await self._repository_source.describe(run.repository_snapshot_id) + result_context = None + if run.source_batch_summary_revision_id is not None: + assert self._result_context_source is not None + result_context = await self._result_context_source.load( + run.source_batch_summary_revision_id + ) prior_results, human_inputs = await self._history(agent_run_id, tools) compiled = self._compiler.compile( provider_name=run.provider_name, @@ -205,6 +263,7 @@ async def advance_run( definitions=tools.definitions, prior_tool_results=prior_results, prior_human_inputs=human_inputs, + result_context=result_context, ) step = AgentStepDetails( agent_step_id=uuid4(), @@ -454,6 +513,13 @@ async def _materialize_proposal( step: AgentStepDetails, decision: ProposeDecision | None = None, ) -> AgentRunDetails: + if run.source_batch_summary_revision_id is not None: + return await self._repository.update_run_state( + run.agent_run_id, + expected_row_version=run.row_version, + status="failed", + termination_reason="follow_up_proposal_not_supported", + ) if decision is None: assert step.decision_json is not None parsed = self._decision_adapter.validate_python(step.decision_json) diff --git a/src/labpilot/api/routers/agent_runs.py b/src/labpilot/api/routers/agent_runs.py index 5b1bf93..3da6c40 100644 --- a/src/labpilot/api/routers/agent_runs.py +++ b/src/labpilot/api/routers/agent_runs.py @@ -11,6 +11,7 @@ AgentRunResponse, AgentRunTraceResponse, CreateAgentRunRequest, + CreateFollowUpAgentRunRequest, HumanInputRequest, ) from labpilot.common.commands import CommandContext @@ -43,6 +44,32 @@ async def create_agent_run( return AgentRunResponse.from_details(details) +@router.post( + "/projects/{project_id}/agent-runs/follow-up", + response_model=AgentRunResponse, + status_code=201, +) +async def create_follow_up_agent_run( + project_id: UUID, + body: CreateFollowUpAgentRunRequest, + context: Annotated[CommandContext, Depends(get_command_context)], + service: Annotated[AgentApplicationService, Depends(get_agent_run_service)], +) -> AgentRunResponse: + try: + budget = body.budget.to_domain() + except ValueError as exc: + raise RequestDomainValidationError("invalid AgentRun request") from exc + details = await service.create_follow_up_run( + context=context, + project_id=project_id, + source_batch_summary_revision_id=body.source_batch_summary_revision_id, + repository_snapshot_id=body.repository_snapshot_id, + goal=body.goal, + budget=budget, + ) + return AgentRunResponse.from_details(details) + + @router.get("/agent-runs/{agent_run_id}", response_model=AgentRunResponse) async def get_agent_run( agent_run_id: UUID, diff --git a/src/labpilot/api/schemas/agent_runs.py b/src/labpilot/api/schemas/agent_runs.py index ed0ca8c..7cd9bf9 100644 --- a/src/labpilot/api/schemas/agent_runs.py +++ b/src/labpilot/api/schemas/agent_runs.py @@ -31,6 +31,15 @@ class CreateAgentRunRequest(BaseModel): budget: AgentBudgetRequest +class CreateFollowUpAgentRunRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + source_batch_summary_revision_id: UUID + repository_snapshot_id: UUID + goal: str + budget: AgentBudgetRequest + + class AdvanceAgentRunRequest(BaseModel): model_config = ConfigDict(extra="forbid") diff --git a/src/labpilot/main.py b/src/labpilot/main.py index 30ad5f5..94ddb52 100644 --- a/src/labpilot/main.py +++ b/src/labpilot/main.py @@ -14,6 +14,7 @@ from labpilot.api.actor import ServerConfiguredActorProvider from labpilot.api.app import create_app from labpilot.api.dependencies import ApplicationServices +from labpilot.api.result_aware_agent_adapter import SqlResultAwareAgentAdapter from labpilot.context.content_store import ContentAddressedStore from labpilot.context.git_adapter import GitAdapter from labpilot.context.proposal_repository_adapter import ProposalRepositoryFactAdapter @@ -79,8 +80,13 @@ def tool_service_for_snapshot(repository_snapshot_id: UUID) -> ToolApplicationSe provider = UnconfiguredModelProvider() model_name = "unconfigured" +agent_repository = SqlAgentRepository(database.session) +result_aware_agent = SqlResultAwareAgentAdapter( + database.session, + agent_repository=agent_repository, +) agent_runs = AgentApplicationService( - repository=SqlAgentRepository(database.session), + repository=agent_repository, repository_source=repository_source, provider=provider, model_name=model_name, @@ -88,6 +94,8 @@ def tool_service_for_snapshot(repository_snapshot_id: UUID) -> ToolApplicationSe tool_service_factory=tool_service_for_snapshot, proposal_service=cast(AgentProposalService, proposals), data_egress_policy=settings.data_egress_policy, + follow_up_run_store=result_aware_agent, + result_context_source=result_aware_agent, ) services = ApplicationServices( projects=ProjectApplicationService(database.session), diff --git a/tests/integration/api/test_result_aware_agent_api.py b/tests/integration/api/test_result_aware_agent_api.py new file mode 100644 index 0000000..5bc5334 --- /dev/null +++ b/tests/integration/api/test_result_aware_agent_api.py @@ -0,0 +1,233 @@ +from __future__ import annotations + +import importlib +from pathlib import Path +from uuid import UUID + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import func, select +from tests.integration.test_batch_summary_service import _process +from tests.integration.test_result_processing_service import _seed_result_attempt + +from labpilot.agent_runtime.context_compiler import ContextCompiler +from labpilot.agent_runtime.provider import ModelRequest, ModelResponse +from labpilot.agent_runtime.repository import SqlAgentRepository +from labpilot.agent_runtime.service import AgentApplicationService +from labpilot.api.actor import ServerConfiguredActorProvider +from labpilot.api.app import create_app +from labpilot.api.dependencies import ApplicationServices +from labpilot.api.result_aware_agent_adapter import SqlResultAwareAgentAdapter +from labpilot.context.content_store import ContentAddressedStore +from labpilot.context.git_adapter import GitAdapter +from labpilot.context.repository_source import RepositorySnapshotSourceService +from labpilot.coordinator.execution_adapters import VersionExecutionSemanticsAdapter +from labpilot.execution.models import BatchRow +from labpilot.execution.trial_outcome import TrialExecutionOutcomeService +from labpilot.experiments.project_service import ProjectApplicationService +from labpilot.experiments.proposal_models import ( + ExperimentRow, + ExperimentVersionRow, + ProposalRevisionRow, +) +from labpilot.results_validation.models import BatchSummaryRevisionRow +from labpilot.tools_policy.repository import SqlToolRepository +from labpilot.tools_policy.repository_tools import repository_tool_definitions +from labpilot.tools_policy.service import ToolApplicationService + +pytestmark = pytest.mark.integration + +_BUDGET = { + "max_steps": 4, + "max_model_calls": 4, + "max_wall_clock_seconds": 60, + "max_tokens": 100, +} + + +class ScriptedProvider: + provider_name = "scripted" + egress = "local" + + def __init__(self, script: list[object]) -> None: + self.script = script + self.calls: list[ModelRequest] = [] + + async def complete(self, request: ModelRequest) -> ModelResponse: + self.calls.append(request) + return ModelResponse(payload=self.script.pop(0), usage={"total_tokens": 1}) + + +def _app(system: object, provider: ScriptedProvider): + database = system.database # type: ignore[attr-defined] + source = RepositorySnapshotSourceService( + database.session, + git_adapter=GitAdapter(git_executable=Path("/usr/bin/git"), timeout_seconds=2), + content_store=ContentAddressedStore(system.repository.parent / "content"), # type: ignore[attr-defined] + ) + agent_repository = SqlAgentRepository(database.session) + result_aware_agent = SqlResultAwareAgentAdapter( + database.session, agent_repository=agent_repository + ) + + def tool_service_factory(repository_snapshot_id: UUID) -> ToolApplicationService: + return ToolApplicationService( + repository_tool_definitions(source, repository_snapshot_id), + SqlToolRepository(database.session), + ) + + agent_runs = AgentApplicationService( + repository=agent_repository, + repository_source=source, + provider=provider, + model_name="scripted-model", + context_compiler=ContextCompiler(), + tool_service_factory=tool_service_factory, + proposal_service=system.service, # type: ignore[attr-defined] + data_egress_policy="local_only", + follow_up_run_store=result_aware_agent, + result_context_source=result_aware_agent, + ) + return create_app( + services=ApplicationServices( + projects=ProjectApplicationService(database.session), + repository_source=source, + proposals=system.service, # type: ignore[attr-defined] + agent_runs=agent_runs, + ), + actor_provider=ServerConfiguredActorProvider( + installation_identity="result-aware-api", actor_id="operator" + ), + ) + + +async def _summary_id(system: object, batch_id: UUID) -> tuple[UUID, UUID]: + database = system.database # type: ignore[attr-defined] + summary_module = importlib.import_module("labpilot.results_validation.summary_service") + outcomes = TrialExecutionOutcomeService( + database.session, + version_semantics=VersionExecutionSemanticsAdapter(database.session), + ) + await summary_module.BatchSummaryService( + database.session, trial_execution_outcome_service=outcomes + ).refresh_summary(batch_id) + async with database.session() as session: + batch = await session.get(BatchRow, batch_id) + summary = await session.scalar( + select(BatchSummaryRevisionRow) + .where(BatchSummaryRevisionRow.batch_id == batch_id) + .order_by(BatchSummaryRevisionRow.revision_number.desc()) + ) + assert summary is not None and batch is not None + return summary.batch_summary_revision_id, batch.project_id + + +async def _counts(system: object) -> tuple[int, int, int, int]: + database = system.database # type: ignore[attr-defined] + async with database.session() as session: + values = [] + for row in (ProposalRevisionRow, ExperimentVersionRow, BatchRow, ExperimentRow): + value = await session.scalar(select(func.count()).select_from(row)) + assert value is not None + values.append(value) + return tuple(values) # type: ignore[return-value] + + +@pytest.mark.asyncio +async def test_follow_up_api_binds_exact_summary_context_and_stops_without_writes( + isolated_database_url: str, tmp_path: Path +) -> None: + async with _seed_result_attempt(isolated_database_url, tmp_path) as ( + system, + batch_id, + _trial_id, + attempt_id, + _workspace, + ): + await _process(system.database, attempt_id) + summary_id, project_id = await _summary_id(system, batch_id) + provider = ScriptedProvider( + [{"kind": "stop", "reason_code": "goal_satisfied", "summary": "done"}] + ) + async with AsyncClient( + transport=ASGITransport(app=_app(system, provider)), + base_url="http://test", + ) as client: + created = await client.post( + f"/projects/{project_id}/agent-runs/follow-up", + json={ + "source_batch_summary_revision_id": str(summary_id), + "repository_snapshot_id": str(system.snapshot_id), + "goal": "Interpret this completed batch.", + "budget": _BUDGET, + }, + ) + assert created.status_code == 201 + run = created.json() + assert run["status"] == "ready" + assert run["source_batch_summary_revision_id"] == str(summary_id) + assert run["repository_snapshot_id"] == str(system.snapshot_id) + counts_before = await _counts(system) + + advanced = await client.post( + f"/agent-runs/{run['agent_run_id']}/advance", + json={"expected_row_version": run["row_version"]}, + ) + + assert advanced.status_code == 200 + assert advanced.json()["status"] == "completed" + assert advanced.json()["termination_reason"] == "goal_satisfied" + assert str(summary_id) in str(provider.calls[0].messages) + assert await _counts(system) == counts_before + + +async def _source_proposal(system: object) -> dict[str, object]: + database = system.database # type: ignore[attr-defined] + async with database.session() as session: + revision = await session.scalar(select(ProposalRevisionRow)) + assert revision is not None + return dict(revision.payload_json) + + +@pytest.mark.asyncio +async def test_follow_up_api_propose_fails_closed_without_new_domain_rows( + isolated_database_url: str, tmp_path: Path +) -> None: + async with _seed_result_attempt(isolated_database_url, tmp_path) as ( + system, + batch_id, + _trial_id, + attempt_id, + _workspace, + ): + await _process(system.database, attempt_id) + summary_id, project_id = await _summary_id(system, batch_id) + provider = ScriptedProvider( + [{"kind": "propose", "proposal": await _source_proposal(system)}] + ) + async with AsyncClient( + transport=ASGITransport(app=_app(system, provider)), + base_url="http://test", + ) as client: + created = await client.post( + f"/projects/{project_id}/agent-runs/follow-up", + json={ + "source_batch_summary_revision_id": str(summary_id), + "repository_snapshot_id": str(system.snapshot_id), + "goal": "Propose a next step.", + "budget": _BUDGET, + }, + ) + assert created.status_code == 201 + run = created.json() + counts_before = await _counts(system) + + advanced = await client.post( + f"/agent-runs/{run['agent_run_id']}/advance", + json={"expected_row_version": run["row_version"]}, + ) + + assert advanced.status_code == 200 + assert advanced.json()["status"] == "failed" + assert advanced.json()["termination_reason"] == "follow_up_proposal_not_supported" + assert await _counts(system) == counts_before diff --git a/tests/unit/agent_runtime/test_agent_service.py b/tests/unit/agent_runtime/test_agent_service.py index 680da28..938c5ce 100644 --- a/tests/unit/agent_runtime/test_agent_service.py +++ b/tests/unit/agent_runtime/test_agent_service.py @@ -12,10 +12,12 @@ from labpilot.agent_runtime.context_compiler import ContextCompiler from labpilot.agent_runtime.provider import ModelRequest, ModelResponse, ProviderError +from labpilot.agent_runtime.result_context import ResultContext from labpilot.agent_runtime.service import AgentApplicationService, AgentRuntimeError from labpilot.agent_runtime.types import AgentBudget, AgentRunDetails, AgentStepDetails from labpilot.common.commands import CommandActor, CommandContext from labpilot.context.repository_source import RepositorySourceDescriptor +from labpilot.experiments.proposal_service import _payload_from_json from labpilot.tools_policy.types import ToolCallDetails, ToolDefinition @@ -617,9 +619,7 @@ async def test_get_trace_returns_persisted_state_without_side_effects() -> None: persisted_steps = await repository.list_steps(run.agent_run_id) provider_calls_before = len(provider.calls) tool_factory_calls_before = len(factory.calls) - tool_requests_before = sum( - tools.events.count("request_call") for tools in factory.services - ) + tool_requests_before = sum(tools.events.count("request_call") for tools in factory.services) tool_executions_before = sum(tools.executions for tools in factory.services) trace = await service.get_trace(run.agent_run_id) @@ -719,9 +719,7 @@ async def test_one_agent_service_isolates_two_run_snapshots() -> None: ] ) factory = RecordingToolFactory(repository) - service = _factory_service( - repository, SnapshotCatalog(first, second), provider, factory - ) + service = _factory_service(repository, SnapshotCatalog(first, second), provider, factory) first_run = await service.create_run( context=_context(), project_id=project_id, @@ -812,9 +810,7 @@ async def test_tool_arguments_cannot_override_persisted_snapshot_binding() -> No ] ) factory = RecordingToolFactory(repository) - service = _factory_service( - repository, SnapshotCatalog(persisted, attempted), provider, factory - ) + service = _factory_service(repository, SnapshotCatalog(persisted, attempted), provider, factory) run = await service.create_run( context=_context(), project_id=project_id, @@ -830,10 +826,7 @@ async def test_tool_arguments_cannot_override_persisted_snapshot_binding() -> No ) assert factory.calls == [persisted.repository_snapshot_id] - assert all( - tools.snapshot_id == persisted.repository_snapshot_id - for tools in factory.services - ) + assert all(tools.snapshot_id == persisted.repository_snapshot_id for tools in factory.services) assert factory.services[0].events.count("request_call") == 1 assert factory.services[0].events.count("execute_call") == 1 @@ -1595,3 +1588,299 @@ async def test_structured_logs_have_correlations_and_no_raw_content(caplog) -> N assert all(record.request_id == "agent-request-1" for record in lifecycle) assert all(record.project_id == str(descriptor.project_id) for record in lifecycle) assert all(record.agent_run_id == str(run.agent_run_id) for record in lifecycle) + + +class MemoryFollowUpRunStore: + def __init__(self, repository: MemoryAgentRepository) -> None: + self.repository = repository + self.runs: list[AgentRunDetails] = [] + + async def create_bound_run(self, run: AgentRunDetails) -> AgentRunDetails: + self.runs.append(run) + return await self.repository.add_run(run) + + +class MemoryResultContextSource: + def __init__(self, context: ResultContext) -> None: + self.context = context + self.calls: list[UUID] = [] + + async def load(self, source_batch_summary_revision_id: UUID) -> ResultContext: + self.calls.append(source_batch_summary_revision_id) + return self.context + + +def _result_context( + descriptor: RepositorySourceDescriptor, source_summary_id: UUID +) -> ResultContext: + return ResultContext( + experiment_id=uuid4(), + source_experiment_version_id=uuid4(), + source_version_number=1, + source_proposal_revision_id=uuid4(), + source_batch_id=uuid4(), + source_batch_summary_revision_id=source_summary_id, + source_summary_revision_number=1, + presentation_status="complete", + source_proposal=_payload_from_json(_proposal(descriptor)), + trials=(), + included_results=(), + excluded_results=(), + warnings=(), + aggregates={}, + ) + + +def _follow_up_service( + repository: MemoryAgentRepository, + descriptor: RepositorySourceDescriptor, + provider: ScriptedProvider, + *, + store: MemoryFollowUpRunStore | None, + source: MemoryResultContextSource | None, + egress_policy: str = "local_only", +) -> tuple[AgentApplicationService, MemoryToolService, MemoryProposalService]: + tools = MemoryToolService(repository) + proposals = MemoryProposalService(repository) + service = AgentApplicationService( + repository=repository, + repository_source=SnapshotSource(descriptor), + provider=provider, + model_name="test-model", + context_compiler=ContextCompiler(), + tool_service_factory=lambda snapshot_id: tools, + proposal_service=proposals, + data_egress_policy=egress_policy, + follow_up_run_store=store, + result_context_source=source, + ) + return service, tools, proposals + + +@pytest.mark.asyncio +async def test_initial_constructor_without_follow_up_ports_completes_normally() -> None: + descriptor = _descriptor() + repository = MemoryAgentRepository() + provider = ScriptedProvider([{"kind": "stop", "reason_code": "done"}]) + service, _, _ = _service(repository, descriptor, provider) + run = await service.create_run( + context=_context(), + project_id=descriptor.project_id, + repository_snapshot_id=descriptor.repository_snapshot_id, + goal="initial", + budget=_budget(), + ) + + completed = await service.advance_run( + context=_context(), + agent_run_id=run.agent_run_id, + expected_row_version=run.row_version, + ) + + assert completed.status == "completed" + assert completed.termination_reason == "done" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "configured_store,configured_source", [(False, False), (True, False), (False, True)] +) +async def test_follow_up_create_requires_both_runtime_ports( + configured_store: bool, configured_source: bool +) -> None: + descriptor = _descriptor() + repository = MemoryAgentRepository() + provider = ScriptedProvider([]) + summary_id = uuid4() + store = MemoryFollowUpRunStore(repository) if configured_store else None + source = ( + MemoryResultContextSource(_result_context(descriptor, summary_id)) + if configured_source + else None + ) + service, tools, proposals = _follow_up_service( + repository, descriptor, provider, store=store, source=source + ) + + with pytest.raises(AgentRuntimeError, match="follow_up_runtime_unconfigured"): + await service.create_follow_up_run( + context=_context(), + project_id=descriptor.project_id, + source_batch_summary_revision_id=summary_id, + repository_snapshot_id=descriptor.repository_snapshot_id, + goal="follow up", + budget=_budget(), + ) + + assert repository.runs == {} + assert provider.calls == [] + assert tools.events == [] + assert proposals.calls == [] + + +@pytest.mark.asyncio +async def test_follow_up_advance_requires_context_and_checks_egress_before_load() -> None: + descriptor = _descriptor() + repository = MemoryAgentRepository() + summary_id = uuid4() + provider = ScriptedProvider([]) + service, _, _ = _follow_up_service(repository, descriptor, provider, store=None, source=None) + initial = await service.create_run( + context=_context(), + project_id=descriptor.project_id, + repository_snapshot_id=descriptor.repository_snapshot_id, + goal="follow up", + budget=_budget(), + ) + repository.runs[initial.agent_run_id] = replace( + initial, source_batch_summary_revision_id=summary_id + ) + + with pytest.raises(AgentRuntimeError, match="follow_up_runtime_unconfigured"): + await service.advance_run( + context=_context(), + agent_run_id=initial.agent_run_id, + expected_row_version=initial.row_version, + ) + assert provider.calls == [] + + remote_provider = ScriptedProvider([]) + remote_provider.egress = "remote" + remote_repository = MemoryAgentRepository() + source = MemoryResultContextSource(_result_context(descriptor, summary_id)) + store = MemoryFollowUpRunStore(remote_repository) + remote_service, _, _ = _follow_up_service( + remote_repository, + descriptor, + remote_provider, + store=store, + source=source, + ) + remote_run = await remote_service.create_follow_up_run( + context=_context(), + project_id=descriptor.project_id, + source_batch_summary_revision_id=summary_id, + repository_snapshot_id=descriptor.repository_snapshot_id, + goal="follow up", + budget=_budget(), + ) + + with pytest.raises(AgentRuntimeError, match="remote_provider_blocked"): + await remote_service.advance_run( + context=_context(), + agent_run_id=remote_run.agent_run_id, + expected_row_version=remote_run.row_version, + ) + assert source.calls == [] + assert remote_provider.calls == [] + + +@pytest.mark.asyncio +async def test_remote_follow_up_receives_exact_context_and_stop_does_not_propose() -> None: + descriptor = _descriptor() + repository = MemoryAgentRepository() + summary_id = uuid4() + provider = ScriptedProvider([{"kind": "stop", "reason_code": "goal_satisfied"}]) + provider.egress = "remote" + source = MemoryResultContextSource(_result_context(descriptor, summary_id)) + store = MemoryFollowUpRunStore(repository) + service, _, proposals = _follow_up_service( + repository, + descriptor, + provider, + store=store, + source=source, + egress_policy="remote_provider_allowed", + ) + run = await service.create_follow_up_run( + context=_context(), + project_id=descriptor.project_id, + source_batch_summary_revision_id=summary_id, + repository_snapshot_id=descriptor.repository_snapshot_id, + goal="follow up", + budget=_budget(), + ) + + completed = await service.advance_run( + context=_context(), + agent_run_id=run.agent_run_id, + expected_row_version=run.row_version, + ) + + assert completed.status == "completed" + assert completed.termination_reason == "goal_satisfied" + assert source.calls == [summary_id] + assert str(summary_id) in str(provider.calls[0].messages) + assert proposals.calls == [] + + +@pytest.mark.asyncio +async def test_follow_up_propose_and_pending_replay_fail_closed() -> None: + descriptor = _descriptor() + repository = MemoryAgentRepository() + summary_id = uuid4() + source = MemoryResultContextSource(_result_context(descriptor, summary_id)) + store = MemoryFollowUpRunStore(repository) + provider = ScriptedProvider([{"kind": "propose", "proposal": _proposal(descriptor)}]) + service, _, proposals = _follow_up_service( + repository, descriptor, provider, store=store, source=source + ) + run = await service.create_follow_up_run( + context=_context(), + project_id=descriptor.project_id, + source_batch_summary_revision_id=summary_id, + repository_snapshot_id=descriptor.repository_snapshot_id, + goal="follow up", + budget=_budget(), + ) + + failed = await service.advance_run( + context=_context(), + agent_run_id=run.agent_run_id, + expected_row_version=run.row_version, + ) + assert failed.status == "failed" + assert failed.termination_reason == "follow_up_proposal_not_supported" + assert proposals.calls == [] + + replay_run = await service.create_follow_up_run( + context=_context(), + project_id=descriptor.project_id, + source_batch_summary_revision_id=summary_id, + repository_snapshot_id=descriptor.repository_snapshot_id, + goal="replay", + budget=_budget(), + ) + replay_step = AgentStepDetails( + agent_step_id=uuid4(), + agent_run_id=replay_run.agent_run_id, + ordinal=1, + step_kind="model_call", + tool_call_id=None, + provider_name=provider.provider_name, + model_name="test-model", + canonical_request_hash="a" * 64, + bound_toolset_hash="b" * 64, + semantic_parameters={}, + output_schema_hash="c" * 64, + started_at=datetime.now(UTC), + completed_at=datetime.now(UTC), + outcome="succeeded", + token_usage={}, + decision_kind="propose", + decision_json={"kind": "propose", "proposal": _proposal(descriptor)}, + human_input_text=None, + ) + repository.steps[replay_run.agent_run_id] = [replay_step] + repository.runs[replay_run.agent_run_id] = replace( + replay_run, status="running", next_step_ordinal=2, row_version=2 + ) + + replay_failed = await service.advance_run( + context=_context(), + agent_run_id=replay_run.agent_run_id, + expected_row_version=2, + ) + assert replay_failed.status == "failed" + assert replay_failed.termination_reason == "follow_up_proposal_not_supported" + assert proposals.calls == [] From fdb9738614c36ffa29d995ecc7593bb5366fcb65 Mon Sep 17 00:00:00 2001 From: Air000000 <1113476369@qq.com> Date: Mon, 21 Sep 2026 16:55:41 +0800 Subject: [PATCH 27/33] feat: add same-experiment Agent proposal revision --- src/labpilot/experiments/proposal_service.py | 85 +++++++++++ .../experiments/test_proposal_service.py | 144 ++++++++++++++++++ 2 files changed, 229 insertions(+) diff --git a/src/labpilot/experiments/proposal_service.py b/src/labpilot/experiments/proposal_service.py index e18c5bb..460af6b 100644 --- a/src/labpilot/experiments/proposal_service.py +++ b/src/labpilot/experiments/proposal_service.py @@ -275,6 +275,86 @@ async def create_agent_revision( ) raise AssertionError("UnitOfWork unexpectedly suppressed control flow") + async def create_follow_up_agent_revision( + self, + *, + context: CommandContext, + project_id: UUID, + experiment_id: UUID, + origin_agent_step_id: UUID, + payload: ProposalPayload, + ) -> ProposalRevisionDetails: + payload_hash = proposal_payload_hash(payload) + async with self._session_context_factory() as session: + existing = await self._repository.get_revision_by_origin_step( + session, origin_agent_step_id + ) + if existing is not None: + return await self._require_exact_agent_replay( + session, existing, payload_hash=payload_hash, project_id=project_id, + expected_experiment_id=experiment_id, + ) + + await self._require_snapshot_binding(project_id=project_id, payload=payload) + actor_id = _actor_id(context) + try: + async with UnitOfWork(self._session_context_factory) as uow: + existing = await self._repository.get_revision_by_origin_step( + uow.session, origin_agent_step_id + ) + if existing is not None: + details = await self._require_exact_agent_replay( + uow.session, existing, payload_hash=payload_hash, project_id=project_id, + expected_experiment_id=experiment_id, + ) + await uow.commit() + return details + + experiment = await self._repository.get_experiment_for_update( + uow.session, experiment_id + ) + if experiment is None: + raise ProposalNotFound(f"experiment not found: {experiment_id}") + if experiment.project_id != project_id: + raise ProposalInvariantViolation("experiment Project mismatch") + revision_number = ( + await self._repository.max_revision_number(uow.session, experiment_id) + ) + 1 + row = ProposalRevisionRow( + proposal_revision_id=uuid4(), + experiment_id=experiment_id, + revision_number=revision_number, + origin_agent_step_id=origin_agent_step_id, + repository_snapshot_id=payload.repository_snapshot_id, + repository_snapshot_identity=payload.repository_snapshot_identity, + payload_json=_payload_to_json(payload), + payload_hash=payload_hash, + created_by_actor_type=context.actor.actor_type, + created_by_actor_id=actor_id, + created_at=await _database_now(uow.session), + ) + await self._repository.add_revision(uow, row) + details = _revision_details(row) + await uow.commit() + return details + except IntegrityError as exc: + if not _is_origin_agent_step_conflict(exc): + raise + + async with self._session_context_factory() as session: + existing = await self._repository.get_revision_by_origin_step( + session, origin_agent_step_id + ) + if existing is None: + raise ProposalInvariantViolation( + "origin_agent_step_id conflict has no winning ProposalRevision" + ) + return await self._require_exact_agent_replay( + session, existing, payload_hash=payload_hash, project_id=project_id, + expected_experiment_id=experiment_id, + ) + raise AssertionError("UnitOfWork unexpectedly suppressed control flow") + async def get_revision( self, proposal_revision_id: UUID, @@ -680,6 +760,7 @@ async def _require_exact_agent_replay( *, payload_hash: str, project_id: UUID, + expected_experiment_id: UUID | None = None, ) -> ProposalRevisionDetails: if row.payload_hash != payload_hash: raise ProposalInvariantViolation( @@ -690,6 +771,10 @@ async def _require_exact_agent_replay( raise ProposalInvariantViolation( "origin_agent_step_id already binds a different Project" ) + if expected_experiment_id is not None and row.experiment_id != expected_experiment_id: + raise ProposalInvariantViolation( + "origin_agent_step_id already binds a different Experiment" + ) return _revision_details(row) diff --git a/tests/integration/experiments/test_proposal_service.py b/tests/integration/experiments/test_proposal_service.py index b348851..844f1c3 100644 --- a/tests/integration/experiments/test_proposal_service.py +++ b/tests/integration/experiments/test_proposal_service.py @@ -935,3 +935,147 @@ async def test_threshold_validation_rule_persists_and_reloads_canonically( assert row.payload_json["validation_rules"][-1]["threshold"] == "0.8" assert "threshold" not in row.payload_json["validation_rules"][0] assert reloaded.payload.validation_rules[-1].threshold == Decimal("0.8") + + +async def _follow_up_revision( + system: _ProposalSystem, + *, + experiment_id: UUID, + origin_agent_step_id: UUID | None = None, + payload: ProposalPayload | None = None, + project_id: UUID = PROJECT_ID, +): + return await system.service.create_follow_up_agent_revision( + context=_context(), + project_id=project_id, + experiment_id=experiment_id, + origin_agent_step_id=origin_agent_step_id or uuid4(), + payload=payload or _payload(system), + ) + + +@pytest.mark.integration +async def test_follow_up_agent_revisions_reuse_experiment_and_sequence( + isolated_database_url: str, + tmp_path: Path, +) -> None: + async with _proposal_system(isolated_database_url, tmp_path) as system: + initial = await _agent_revision(system) + experiments_before = await _table_count(system.database, "experiments") + second = await _follow_up_revision(system, experiment_id=initial.experiment_id) + third = await _follow_up_revision(system, experiment_id=initial.experiment_id) + + assert initial.revision_number == 1 + assert (second.experiment_id, second.revision_number) == (initial.experiment_id, 2) + assert (third.experiment_id, third.revision_number) == (initial.experiment_id, 3) + assert await _table_count(system.database, "experiments") == experiments_before + + +@pytest.mark.integration +async def test_follow_up_agent_revision_replay_is_exact_and_experiment_bound( + isolated_database_url: str, + tmp_path: Path, +) -> None: + from labpilot.experiments.proposal_service import ProposalInvariantViolation + + async with _proposal_system(isolated_database_url, tmp_path) as system: + initial = await _agent_revision(system) + origin = uuid4() + first = await _follow_up_revision( + system, + experiment_id=initial.experiment_id, + origin_agent_step_id=origin, + ) + replay = await _follow_up_revision( + system, + experiment_id=initial.experiment_id, + origin_agent_step_id=origin, + ) + with pytest.raises(ProposalInvariantViolation): + await _follow_up_revision( + system, + experiment_id=initial.experiment_id, + origin_agent_step_id=origin, + payload=_payload(system, goal="Changed semantics"), + ) + other = await _agent_revision(system) + with pytest.raises(ProposalInvariantViolation): + await _follow_up_revision( + system, + experiment_id=other.experiment_id, + origin_agent_step_id=origin, + ) + + assert replay.proposal_revision_id == first.proposal_revision_id + assert replay.revision_number == 2 + + +@pytest.mark.integration +async def test_follow_up_agent_revision_rejects_wrong_project_without_write( + isolated_database_url: str, + tmp_path: Path, +) -> None: + from labpilot.experiments.proposal_service import ProposalInvariantViolation + + async with _proposal_system(isolated_database_url, tmp_path) as system: + initial = await _agent_revision(system) + revisions_before = await _table_count( + system.database, "experiment_proposal_revisions" + ) + with pytest.raises(ProposalInvariantViolation): + await _follow_up_revision( + system, + experiment_id=initial.experiment_id, + project_id=UUID("10000000-0000-4000-8000-000000000099"), + ) + + assert await _table_count( + system.database, "experiment_proposal_revisions" + ) == revisions_before + + +@pytest.mark.integration +async def test_concurrent_distinct_follow_up_agent_revisions_are_consecutive( + isolated_database_url: str, + tmp_path: Path, +) -> None: + async with _proposal_system(isolated_database_url, tmp_path) as system: + initial = await _agent_revision(system) + revisions = await asyncio.gather( + _follow_up_revision(system, experiment_id=initial.experiment_id), + _follow_up_revision(system, experiment_id=initial.experiment_id), + ) + + assert {revision.experiment_id for revision in revisions} == {initial.experiment_id} + assert {revision.revision_number for revision in revisions} == {2, 3} + assert await _table_count(system.database, "experiments") == 1 + + +@pytest.mark.integration +async def test_concurrent_same_follow_up_origin_recovers_one_winner( + isolated_database_url: str, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + async with _proposal_system(isolated_database_url, tmp_path) as system: + initial = await _agent_revision(system) + origin = uuid4() + _synchronize_concurrent_origin_inserts(system, monkeypatch) + revisions = await asyncio.gather( + _follow_up_revision( + system, + experiment_id=initial.experiment_id, + origin_agent_step_id=origin, + ), + _follow_up_revision( + system, + experiment_id=initial.experiment_id, + origin_agent_step_id=origin, + ), + ) + + assert len({revision.proposal_revision_id for revision in revisions}) == 1 + assert {revision.revision_number for revision in revisions} == {2} + assert await _table_count( + system.database, "experiment_proposal_revisions" + ) == 2 From ae079be252fd62eff415ef040af4228ab45e2d61 Mon Sep 17 00:00:00 2001 From: Air000000 <1113476369@qq.com> Date: Mon, 21 Sep 2026 18:50:19 +0800 Subject: [PATCH 28/33] feat: materialize follow-up Agent proposals --- src/labpilot/agent_runtime/service.py | 49 +++++-- .../agent_runtime/test_agent_service.py | 127 +++++++++++++++++- .../api/test_result_aware_agent_api.py | 33 ++++- .../unit/agent_runtime/test_agent_service.py | 82 +++++++++-- 4 files changed, 265 insertions(+), 26 deletions(-) diff --git a/src/labpilot/agent_runtime/service.py b/src/labpilot/agent_runtime/service.py index 5207d64..509c726 100644 --- a/src/labpilot/agent_runtime/service.py +++ b/src/labpilot/agent_runtime/service.py @@ -86,6 +86,16 @@ async def create_agent_revision( payload: ProposalPayload, ) -> ProposalRevisionResult: ... + async def create_follow_up_agent_revision( + self, + *, + context: CommandContext, + project_id: UUID, + experiment_id: UUID, + origin_agent_step_id: UUID, + payload: ProposalPayload, + ) -> ProposalRevisionResult: ... + class AgentApplicationService: def __init__( @@ -513,25 +523,38 @@ async def _materialize_proposal( step: AgentStepDetails, decision: ProposeDecision | None = None, ) -> AgentRunDetails: - if run.source_batch_summary_revision_id is not None: - return await self._repository.update_run_state( - run.agent_run_id, - expected_row_version=run.row_version, - status="failed", - termination_reason="follow_up_proposal_not_supported", - ) if decision is None: assert step.decision_json is not None parsed = self._decision_adapter.validate_python(step.decision_json) if not isinstance(parsed, ProposeDecision): raise AgentRuntimeError("proposal_replay_invalid") decision = parsed - revision = await self._proposals.create_agent_revision( - context=context, - project_id=run.project_id, - origin_agent_step_id=step.agent_step_id, - payload=decision.proposal, - ) + if run.source_batch_summary_revision_id is None: + revision = await self._proposals.create_agent_revision( + context=context, + project_id=run.project_id, + origin_agent_step_id=step.agent_step_id, + payload=decision.proposal, + ) + else: + if decision.proposal.repository_snapshot_id != run.repository_snapshot_id: + return await self._repository.update_run_state( + run.agent_run_id, + expected_row_version=run.row_version, + status="failed", + termination_reason="follow_up_proposal_snapshot_mismatch", + ) + assert self._result_context_source is not None + result_context = await self._result_context_source.load( + run.source_batch_summary_revision_id + ) + revision = await self._proposals.create_follow_up_agent_revision( + context=context, + project_id=run.project_id, + experiment_id=result_context.experiment_id, + origin_agent_step_id=step.agent_step_id, + payload=decision.proposal, + ) completed = await self._repository.update_run_state( run.agent_run_id, expected_row_version=run.row_version, diff --git a/tests/integration/agent_runtime/test_agent_service.py b/tests/integration/agent_runtime/test_agent_service.py index 2fed7de..55af9df 100644 --- a/tests/integration/agent_runtime/test_agent_service.py +++ b/tests/integration/agent_runtime/test_agent_service.py @@ -8,7 +8,7 @@ from uuid import UUID import pytest -from sqlalchemy import create_engine, text +from sqlalchemy import create_engine, select, text from tests.integration.context.test_repository_snapshot import _seed_capture_service from tests.integration.db_helpers import upgrade_database_to_head from tests.integration.experiments.test_proposal_service import ( @@ -17,17 +17,26 @@ _payload, _proposal_system, ) +from tests.integration.test_batch_summary_service import _process +from tests.integration.test_result_processing_service import _seed_result_attempt from labpilot.agent_runtime.context_compiler import ContextCompiler from labpilot.agent_runtime.provider import ModelRequest, ModelResponse, ProviderError from labpilot.agent_runtime.repository import SqlAgentRepository from labpilot.agent_runtime.service import AgentApplicationService, AgentRuntimeError from labpilot.agent_runtime.types import AgentBudget +from labpilot.api.result_aware_agent_adapter import SqlResultAwareAgentAdapter from labpilot.common.commands import CommandActor, CommandContext from labpilot.context.content_store import ContentAddressedStore from labpilot.context.git_adapter import GitAdapter from labpilot.context.repository_source import RepositorySnapshotSourceService from labpilot.context.repository_types import CaptureRepositorySnapshot +from labpilot.coordinator.execution_adapters import VersionExecutionSemanticsAdapter +from labpilot.execution.models import BatchRow +from labpilot.execution.trial_outcome import TrialExecutionOutcomeService +from labpilot.experiments.proposal_models import ProposalRevisionRow +from labpilot.results_validation.models import BatchSummaryRevisionRow +from labpilot.results_validation.summary_service import BatchSummaryService from labpilot.tools_policy.repository import SqlToolRepository from labpilot.tools_policy.repository_tools import repository_tool_definitions from labpilot.tools_policy.service import ToolApplicationService @@ -111,6 +120,13 @@ async def create_agent_revision(self, **kwargs): raise RuntimeError("crash after committed ProposalRevision") return result + async def create_follow_up_agent_revision(self, **kwargs): + self.calls += 1 + result = await self.service.create_follow_up_agent_revision(**kwargs) + if self.calls == 1: + raise RuntimeError("crash after committed ProposalRevision") + return result + def test_agent_tables_match_the_frozen_0008_contract( isolated_database_url: str, @@ -611,3 +627,112 @@ async def before_call(request: ModelRequest) -> None: assert completed.status == "completed" assert len(provider.calls) == 1 + + +@pytest.mark.asyncio +async def test_follow_up_proposal_crash_replay_keeps_same_r2( + isolated_database_url: str, tmp_path: Path +) -> None: + async with _seed_result_attempt(isolated_database_url, tmp_path) as ( + system, + batch_id, + _trial_id, + attempt_id, + _workspace, + ): + await _process(system.database, attempt_id) + outcomes = TrialExecutionOutcomeService( + system.database.session, + version_semantics=VersionExecutionSemanticsAdapter(system.database.session), + ) + await BatchSummaryService( + system.database.session, trial_execution_outcome_service=outcomes + ).refresh_summary(batch_id) + async with system.database.session() as session: + batch = await session.get(BatchRow, batch_id) + summary = await session.scalar( + select(BatchSummaryRevisionRow) + .where(BatchSummaryRevisionRow.batch_id == batch_id) + .order_by(BatchSummaryRevisionRow.revision_number.desc()) + ) + source_revision = await session.scalar( + select(ProposalRevisionRow).order_by(ProposalRevisionRow.revision_number) + ) + assert batch is not None and summary is not None and source_revision is not None + + source = RepositorySnapshotSourceService( + system.database.session, + git_adapter=GitAdapter(git_executable=Path('/usr/bin/git'), timeout_seconds=2), + content_store=ContentAddressedStore(tmp_path / 'content'), + ) + repository = SqlAgentRepository(system.database.session) + result_adapter = SqlResultAwareAgentAdapter( + system.database.session, agent_repository=repository + ) + provider = ScriptedProvider( + [{'kind': 'propose', 'proposal': dict(source_revision.payload_json)}] + ) + crashing = CrashAfterCreateProposalService(system.service) + tools = ToolApplicationService( + repository_tool_definitions(source, system.snapshot_id), + SqlToolRepository(system.database.session), + ) + service = AgentApplicationService( + repository=repository, + repository_source=source, + provider=provider, + model_name='test-model', + context_compiler=ContextCompiler(), + tool_service_factory=_single_snapshot_tool_factory(system.snapshot_id, tools), + proposal_service=crashing, + data_egress_policy='local_only', + follow_up_run_store=result_adapter, + result_context_source=result_adapter, + ) + run = await service.create_follow_up_run( + context=_context(), + project_id=batch.project_id, + source_batch_summary_revision_id=summary.batch_summary_revision_id, + repository_snapshot_id=system.snapshot_id, + goal='propose the next revision', + budget=_budget(), + ) + + with pytest.raises(RuntimeError, match='crash after committed'): + await service.advance_run( + context=_context(), + agent_run_id=run.agent_run_id, + expected_row_version=run.row_version, + ) + after_crash = await service.get_run(run.agent_run_id) + assert after_crash.produced_proposal_revision_id is None + assert after_crash.status == 'running' + + async with system.database.session() as session: + revisions_after_crash = ( + await session.scalars( + select(ProposalRevisionRow).order_by(ProposalRevisionRow.revision_number) + ) + ).all() + assert len(revisions_after_crash) == 2 + assert revisions_after_crash[1].experiment_id == revisions_after_crash[0].experiment_id + assert revisions_after_crash[1].revision_number == 2 + + completed = await service.advance_run( + context=_context(), + agent_run_id=run.agent_run_id, + expected_row_version=after_crash.row_version, + ) + + assert completed.status == 'completed' + assert completed.termination_reason == 'proposal_created' + assert completed.produced_proposal_revision_id == revisions_after_crash[1].proposal_revision_id + assert len(provider.calls) == 1 + assert str(summary.batch_summary_revision_id) in str(provider.calls[0].messages) + async with system.database.session() as session: + revisions_after_replay = ( + await session.scalars( + select(ProposalRevisionRow).order_by(ProposalRevisionRow.revision_number) + ) + ).all() + assert [revision.revision_number for revision in revisions_after_replay] == [1, 2] diff --git a/tests/integration/api/test_result_aware_agent_api.py b/tests/integration/api/test_result_aware_agent_api.py index 5bc5334..1d21fb0 100644 --- a/tests/integration/api/test_result_aware_agent_api.py +++ b/tests/integration/api/test_result_aware_agent_api.py @@ -26,9 +26,11 @@ from labpilot.execution.trial_outcome import TrialExecutionOutcomeService from labpilot.experiments.project_service import ProjectApplicationService from labpilot.experiments.proposal_models import ( + DesignDecisionRow, ExperimentRow, ExperimentVersionRow, ProposalRevisionRow, + ProposalVerificationRow, ) from labpilot.results_validation.models import BatchSummaryRevisionRow from labpilot.tools_policy.repository import SqlToolRepository @@ -133,6 +135,17 @@ async def _counts(system: object) -> tuple[int, int, int, int]: return tuple(values) # type: ignore[return-value] +async def _governance_counts(system: object) -> tuple[int, int]: + database = system.database # type: ignore[attr-defined] + async with database.session() as session: + values = [] + for row in (ProposalVerificationRow, DesignDecisionRow): + value = await session.scalar(select(func.count()).select_from(row)) + assert value is not None + values.append(value) + return tuple(values) # type: ignore[return-value] + + @pytest.mark.asyncio async def test_follow_up_api_binds_exact_summary_context_and_stops_without_writes( isolated_database_url: str, tmp_path: Path @@ -190,7 +203,7 @@ async def _source_proposal(system: object) -> dict[str, object]: @pytest.mark.asyncio -async def test_follow_up_api_propose_fails_closed_without_new_domain_rows( +async def test_follow_up_api_propose_creates_same_experiment_revision( isolated_database_url: str, tmp_path: Path ) -> None: async with _seed_result_attempt(isolated_database_url, tmp_path) as ( @@ -221,6 +234,7 @@ async def test_follow_up_api_propose_fails_closed_without_new_domain_rows( assert created.status_code == 201 run = created.json() counts_before = await _counts(system) + governance_before = await _governance_counts(system) advanced = await client.post( f"/agent-runs/{run['agent_run_id']}/advance", @@ -228,6 +242,17 @@ async def test_follow_up_api_propose_fails_closed_without_new_domain_rows( ) assert advanced.status_code == 200 - assert advanced.json()["status"] == "failed" - assert advanced.json()["termination_reason"] == "follow_up_proposal_not_supported" - assert await _counts(system) == counts_before + assert advanced.json()["status"] == "completed" + assert advanced.json()["termination_reason"] == "proposal_created" + counts_after = await _counts(system) + assert counts_after == (counts_before[0] + 1, *counts_before[1:]) + async with system.database.session() as session: # type: ignore[attr-defined] + revisions = ( + await session.scalars( + select(ProposalRevisionRow).order_by(ProposalRevisionRow.revision_number) + ) + ).all() + assert len(revisions) == 2 + assert revisions[1].experiment_id == revisions[0].experiment_id + assert revisions[1].revision_number == revisions[0].revision_number + 1 + assert await _governance_counts(system) == governance_before diff --git a/tests/unit/agent_runtime/test_agent_service.py b/tests/unit/agent_runtime/test_agent_service.py index 938c5ce..d8f982d 100644 --- a/tests/unit/agent_runtime/test_agent_service.py +++ b/tests/unit/agent_runtime/test_agent_service.py @@ -451,6 +451,7 @@ def __init__(self, repository: MemoryAgentRepository) -> None: self.repository = repository self.calls: list[UUID] = [] self.revisions: dict[UUID, Revision] = {} + self.follow_up_calls: list[tuple[UUID, UUID, UUID]] = [] self.crash_after_first_create = False async def create_agent_revision( @@ -473,6 +474,24 @@ async def create_agent_revision( raise RuntimeError("simulated materialization crash") return revision + async def create_follow_up_agent_revision( + self, *, context, project_id, experiment_id, origin_agent_step_id, payload + ) -> Revision: + steps = await self.repository.list_steps( + next( + run_id + for run_id, steps in self.repository.steps.items() + if any(step.agent_step_id == origin_agent_step_id for step in steps) + ) + ) + assert ( + next(step for step in steps if step.agent_step_id == origin_agent_step_id).outcome + == "succeeded" + ) + self.follow_up_calls.append((project_id, experiment_id, origin_agent_step_id)) + return self.revisions.setdefault(origin_agent_step_id, Revision(uuid4())) + + def _context() -> CommandContext: return CommandContext( @@ -1377,6 +1396,7 @@ async def test_propose_is_persisted_before_materialization_and_replays_after_cra assert len(provider.calls) == 1 assert len(proposals.calls) == 2 assert proposals.calls[0] == proposals.calls[1] + assert proposals.follow_up_calls == [] @pytest.mark.asyncio @@ -1815,11 +1835,12 @@ async def test_remote_follow_up_receives_exact_context_and_stop_does_not_propose @pytest.mark.asyncio -async def test_follow_up_propose_and_pending_replay_fail_closed() -> None: +async def test_follow_up_propose_materializes_same_experiment_and_replays() -> None: descriptor = _descriptor() repository = MemoryAgentRepository() summary_id = uuid4() - source = MemoryResultContextSource(_result_context(descriptor, summary_id)) + result_context = _result_context(descriptor, summary_id) + source = MemoryResultContextSource(result_context) store = MemoryFollowUpRunStore(repository) provider = ScriptedProvider([{"kind": "propose", "proposal": _proposal(descriptor)}]) service, _, proposals = _follow_up_service( @@ -1834,14 +1855,20 @@ async def test_follow_up_propose_and_pending_replay_fail_closed() -> None: budget=_budget(), ) - failed = await service.advance_run( + completed = await service.advance_run( context=_context(), agent_run_id=run.agent_run_id, expected_row_version=run.row_version, ) - assert failed.status == "failed" - assert failed.termination_reason == "follow_up_proposal_not_supported" + assert completed.status == "completed" + assert completed.termination_reason == "proposal_created" assert proposals.calls == [] + assert len(proposals.follow_up_calls) == 1 + project_id, experiment_id, step_id = proposals.follow_up_calls[0] + assert project_id == descriptor.project_id + assert experiment_id == result_context.experiment_id + assert step_id == repository.steps[run.agent_run_id][-1].agent_step_id + assert source.calls == [summary_id, summary_id] replay_run = await service.create_follow_up_run( context=_context(), @@ -1876,11 +1903,50 @@ async def test_follow_up_propose_and_pending_replay_fail_closed() -> None: replay_run, status="running", next_step_ordinal=2, row_version=2 ) - replay_failed = await service.advance_run( + replay_completed = await service.advance_run( context=_context(), agent_run_id=replay_run.agent_run_id, expected_row_version=2, ) - assert replay_failed.status == "failed" - assert replay_failed.termination_reason == "follow_up_proposal_not_supported" + assert replay_completed.status == "completed" + assert replay_completed.termination_reason == "proposal_created" + assert proposals.calls == [] + assert proposals.follow_up_calls == [ + (descriptor.project_id, result_context.experiment_id, step_id), + (descriptor.project_id, result_context.experiment_id, replay_step.agent_step_id) + ] + assert source.calls == [summary_id, summary_id, summary_id] + + +@pytest.mark.asyncio +async def test_follow_up_propose_rejects_snapshot_mismatch_without_initial_fallback() -> None: + descriptor = _descriptor() + repository = MemoryAgentRepository() + summary_id = uuid4() + source = MemoryResultContextSource(_result_context(descriptor, summary_id)) + store = MemoryFollowUpRunStore(repository) + proposal = _proposal(descriptor) + proposal["repository_snapshot_id"] = str(uuid4()) + provider = ScriptedProvider([{"kind": "propose", "proposal": proposal}]) + service, _, proposals = _follow_up_service( + repository, descriptor, provider, store=store, source=source + ) + run = await service.create_follow_up_run( + context=_context(), + project_id=descriptor.project_id, + source_batch_summary_revision_id=summary_id, + repository_snapshot_id=descriptor.repository_snapshot_id, + goal="follow up", + budget=_budget(), + ) + + failed = await service.advance_run( + context=_context(), + agent_run_id=run.agent_run_id, + expected_row_version=run.row_version, + ) + + assert failed.status == "failed" + assert failed.termination_reason == "follow_up_proposal_snapshot_mismatch" assert proposals.calls == [] + assert proposals.follow_up_calls == [] From 4e8bcab0c04922d8ce55ad14b8fde4bc036a5d25 Mon Sep 17 00:00:00 2001 From: Air000000 <1113476369@qq.com> Date: Mon, 21 Sep 2026 21:45:01 +0800 Subject: [PATCH 29/33] test: prove result-aware two-iteration closure --- .../p1/test_result_aware_iteration.py | 254 ++++++++++++++++++ 1 file changed, 254 insertions(+) create mode 100644 tests/integration/p1/test_result_aware_iteration.py diff --git a/tests/integration/p1/test_result_aware_iteration.py b/tests/integration/p1/test_result_aware_iteration.py new file mode 100644 index 0000000..ed7364f --- /dev/null +++ b/tests/integration/p1/test_result_aware_iteration.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +import json +from copy import deepcopy +from pathlib import Path +from uuid import UUID, uuid4 + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import func, select +from tests.integration.api.test_result_aware_agent_api import ( + _BUDGET, + ScriptedProvider, + _app, + _source_proposal, + _summary_id, +) +from tests.integration.experiments.test_proposal_service import _context +from tests.integration.test_batch_summary_service import _process +from tests.integration.test_result_processing_service import _seed_result_attempt + +from labpilot.execution.models import AttemptRow, BatchRow, TrialRow +from labpilot.experiments.proposal_models import ( + DesignDecisionRow, + ExperimentRow, + ExperimentVersionRow, + ProposalRevisionRow, + ProposalVerificationRow, +) +from labpilot.results_validation.models import ValidationReportRow + +pytestmark = pytest.mark.integration + +_SENTINEL = 'SHOULD_NOT_REACH_AGENT_CONTEXT_invalid_metric_0.01_stdout_checkpoint' + + +async def _add_excluded_trial( + system: object, batch_id: UUID, first_attempt_id: UUID +) -> UUID: + database = system.database # type: ignore[attr-defined] + async with database.session() as session: + first_attempt = await session.get(AttemptRow, first_attempt_id) + first_trial = await session.get(TrialRow, first_attempt.trial_id) + assert first_attempt is not None and first_trial is not None + excluded_trial_id = uuid4() + session.add( + TrialRow( + trial_id=excluded_trial_id, + batch_id=batch_id, + trial_ordinal=2, + trial_identity=uuid4().hex * 2, + resolved_parameters_json={'learning_rate': 0.2}, + seed=first_trial.seed, + ) + ) + await session.flush() + session.add( + AttemptRow( + attempt_id=uuid4(), + trial_id=excluded_trial_id, + attempt_number=1, + status='terminal', + supervisor_execution_identity=f'excluded-{excluded_trial_id}', + workspace_path=first_attempt.workspace_path, + frozen_execution_spec_json=deepcopy(first_attempt.frozen_execution_spec_json), + frozen_execution_spec_hash=first_attempt.frozen_execution_spec_hash, + frozen_launch_spec_json=deepcopy(first_attempt.frozen_launch_spec_json), + execution_spec_hash=first_attempt.execution_spec_hash, + launch_intent_committed_at=first_attempt.launch_intent_committed_at, + observed_environment_json=deepcopy(first_attempt.observed_environment_json), + observed_environment_recorded_at=first_attempt.observed_environment_recorded_at, + physical_started_at=first_attempt.physical_started_at, + physical_ended_at=first_attempt.physical_ended_at, + terminal_kind='process_exited', + exit_code=1, + terminal_at=first_attempt.terminal_at, + ) + ) + await session.commit() + return excluded_trial_id + + +async def _counts(system: object) -> tuple[int, int, int]: + database = system.database # type: ignore[attr-defined] + async with database.session() as session: + values = [] + for row in (ProposalRevisionRow, ExperimentVersionRow, BatchRow): + value = await session.scalar(select(func.count()).select_from(row)) + assert value is not None + values.append(value) + return tuple(values) # type: ignore[return-value] + + +@pytest.mark.asyncio +async def test_result_aware_two_iteration_closure( + isolated_database_url: str, tmp_path: Path +) -> None: + async with _seed_result_attempt(isolated_database_url, tmp_path) as ( + system, + batch_id, + included_trial_id, + attempt_id, + workspace, + ): + (workspace / 'runtime' / 'output' / 'raw-debug.txt').write_text( + _SENTINEL, encoding='utf-8' + ) + await _process(system.database, attempt_id) + excluded_trial_id = await _add_excluded_trial(system, batch_id, attempt_id) + summary_id, project_id = await _summary_id(system, batch_id) + async with system.database.session() as session: + batch = await session.get(BatchRow, batch_id) + assert batch is not None + v1 = await session.get(ExperimentVersionRow, batch.experiment_version_id) + report = await session.scalar(select(ValidationReportRow)) + assert v1 is not None and report is not None + r1 = await session.get(ProposalRevisionRow, v1.proposal_revision_id) + assert r1 is not None + + r2_payload = deepcopy(await _source_proposal(system)) + r2_payload['parameter_matrix'] = {'learning_rate': [0.2]} + provider = ScriptedProvider([{'kind': 'propose', 'proposal': r2_payload}]) + async with AsyncClient( + transport=ASGITransport(app=_app(system, provider)), + base_url='http://test', + ) as client: + created = await client.post( + f'/projects/{project_id}/agent-runs/follow-up', + json={ + 'source_batch_summary_revision_id': str(summary_id), + 'repository_snapshot_id': str(system.snapshot_id), + 'goal': 'Improve this experiment from the exact results.', + 'budget': _BUDGET, + }, + ) + assert created.status_code == 201 + run = created.json() + assert run['source_batch_summary_revision_id'] == str(summary_id) + + # Phase A: the Agent may materialize R2, but nothing governed after it. + advanced = await client.post( + f"/agent-runs/{run['agent_run_id']}/advance", + json={'expected_row_version': run['row_version']}, + ) + + assert advanced.status_code == 200 + advanced_body = advanced.json() + assert advanced_body['status'] == 'completed' + assert advanced_body['termination_reason'] == 'proposal_created' + r2_id = UUID(advanced_body['produced_proposal_revision_id']) + + request = json.loads(provider.calls[0].messages[1]['content']) + result_context = request['result_context'] + assert result_context['source_batch_summary_revision_id'] == str(summary_id) + assert result_context['source_experiment_version_id'] == str(v1.experiment_version_id) + assert result_context['experiment_id'] == str(r1.experiment_id) + assert result_context['source_batch_id'] == str(batch_id) + assert any( + item['trial_id'] == str(included_trial_id) + and item['metrics'][0]['numeric_value'] == '0.82' + and item['validation_report_id'] == str(report.validation_report_id) + for item in result_context['included_results'] + ) + assert any( + item['trial_id'] == str(excluded_trial_id) + and item['exclusion_code'] == 'EXECUTION_EXHAUSTED' + for item in result_context['excluded_results'] + ) + assert any( + item['trial_id'] == str(excluded_trial_id) + and item['trial_ordinal'] == 2 + and item['resolved_parameters'] == {'learning_rate': 0.2} + and item['seed'] == 7 + for item in result_context['trials'] + ) + assert _SENTINEL not in provider.calls[0].messages[1]['content'] + + async with system.database.session() as session: + r2 = await session.get(ProposalRevisionRow, r2_id) + verification_count = await session.scalar( + select(func.count()) + .select_from(ProposalVerificationRow) + .where(ProposalVerificationRow.proposal_revision_id == r2_id) + ) + decision_count = await session.scalar( + select(func.count()) + .select_from(DesignDecisionRow) + .where(DesignDecisionRow.proposal_revision_id == r2_id) + ) + v2_before_human = await session.scalar( + select(ExperimentVersionRow).where( + ExperimentVersionRow.proposal_revision_id == r2_id + ) + ) + experiment_count = await session.scalar(select(func.count()).select_from(ExperimentRow)) + assert r2 is not None + assert r2.experiment_id == r1.experiment_id + assert r2.revision_number == r1.revision_number + 1 + assert verification_count == decision_count == 0 + assert v2_before_human is None + assert experiment_count == 1 + + # Phase B: deterministic verification is explicit and outside Agent runtime. + verification = await system.service.verify_revision( + context=_context(), proposal_revision_id=r2_id + ) + assert verification.result.passed is True + + # Phase C: approval is an explicit human command, never an Agent side effect. + v2 = await system.service.approve_revision( + context=_context(), + proposal_revision_id=r2_id, + verification_id=verification.proposal_verification_id, + ) + assert v2.experiment_id == v1.experiment_id + assert v2.version_number == v1.version_number + 1 + assert v2.proposal_revision_id == r2_id + async with system.database.session() as session: + v2_batches = await session.scalar( + select(func.count()) + .select_from(BatchRow) + .where(BatchRow.experiment_version_id == v2.experiment_version_id) + ) + assert v2_batches == 0 + + # A second exact-source run that stops creates no R3, V3, or Batch. + counts_before_stop = await _counts(system) + stop_provider = ScriptedProvider( + [{'kind': 'stop', 'reason_code': 'goal_satisfied', 'summary': 'done'}] + ) + async with AsyncClient( + transport=ASGITransport(app=_app(system, stop_provider)), + base_url='http://test', + ) as client: + stopped_created = await client.post( + f'/projects/{project_id}/agent-runs/follow-up', + json={ + 'source_batch_summary_revision_id': str(summary_id), + 'repository_snapshot_id': str(system.snapshot_id), + 'goal': 'Confirm the exact batch goal is satisfied.', + 'budget': _BUDGET, + }, + ) + assert stopped_created.status_code == 201 + stopped = stopped_created.json() + stopped_advanced = await client.post( + f"/agent-runs/{stopped['agent_run_id']}/advance", + json={'expected_row_version': stopped['row_version']}, + ) + assert stopped_advanced.status_code == 200 + assert stopped_advanced.json()['status'] == 'completed' + assert stopped_advanced.json()['termination_reason'] == 'goal_satisfied' + assert str(summary_id) in stop_provider.calls[0].messages[1]['content'] + assert await _counts(system) == counts_before_stop From 5e1c735adb6d3bb12a1be8028960a8c5015abbb0 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Mon, 21 Sep 2026 21:48:27 +0800 Subject: [PATCH 30/33] test: prove raw metric stays outside result-aware context --- .../p1/test_result_aware_iteration.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/tests/integration/p1/test_result_aware_iteration.py b/tests/integration/p1/test_result_aware_iteration.py index ed7364f..948c776 100644 --- a/tests/integration/p1/test_result_aware_iteration.py +++ b/tests/integration/p1/test_result_aware_iteration.py @@ -2,6 +2,7 @@ import json from copy import deepcopy +from decimal import Decimal from pathlib import Path from uuid import UUID, uuid4 @@ -27,7 +28,7 @@ ProposalRevisionRow, ProposalVerificationRow, ) -from labpilot.results_validation.models import ValidationReportRow +from labpilot.results_validation.models import MetricObservationRow, ValidationReportRow pytestmark = pytest.mark.integration @@ -91,6 +92,19 @@ async def _counts(system: object) -> tuple[int, int, int]: return tuple(values) # type: ignore[return-value] +async def _mutate_raw_metric_after_summary(system: object) -> str: + raw_value = "0.01" + database = system.database # type: ignore[attr-defined] + async with database.session() as session: + metric = await session.scalar( + select(MetricObservationRow).order_by(MetricObservationRow.created_at) + ) + assert metric is not None + metric.numeric_value = Decimal(raw_value) + await session.commit() + return raw_value + + @pytest.mark.asyncio async def test_result_aware_two_iteration_closure( isolated_database_url: str, tmp_path: Path @@ -108,6 +122,7 @@ async def test_result_aware_two_iteration_closure( await _process(system.database, attempt_id) excluded_trial_id = await _add_excluded_trial(system, batch_id, attempt_id) summary_id, project_id = await _summary_id(system, batch_id) + raw_metric_after_summary = await _mutate_raw_metric_after_summary(system) async with system.database.session() as session: batch = await session.get(BatchRow, batch_id) assert batch is not None @@ -174,6 +189,7 @@ async def test_result_aware_two_iteration_closure( for item in result_context['trials'] ) assert _SENTINEL not in provider.calls[0].messages[1]['content'] + assert raw_metric_after_summary not in provider.calls[0].messages[1]['content'] async with system.database.session() as session: r2 = await session.get(ProposalRevisionRow, r2_id) From ab73e2c5a96a9b661b418ebf59d1169c23afad11 Mon Sep 17 00:00:00 2001 From: Air000000 <73988854+Air000000@users.noreply.github.com> Date: Mon, 21 Sep 2026 21:48:50 +0800 Subject: [PATCH 31/33] test: use unique raw metric sentinel --- tests/integration/p1/test_result_aware_iteration.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/p1/test_result_aware_iteration.py b/tests/integration/p1/test_result_aware_iteration.py index 948c776..1bee6c8 100644 --- a/tests/integration/p1/test_result_aware_iteration.py +++ b/tests/integration/p1/test_result_aware_iteration.py @@ -93,7 +93,7 @@ async def _counts(system: object) -> tuple[int, int, int]: async def _mutate_raw_metric_after_summary(system: object) -> str: - raw_value = "0.01" + raw_value = "-12345.6789012345" database = system.database # type: ignore[attr-defined] async with database.session() as session: metric = await session.scalar( From e8caaf668fed2b5e5878393fa0c47c001ba218e9 Mon Sep 17 00:00:00 2001 From: Air000000 <1113476369@qq.com> Date: Tue, 22 Sep 2026 13:09:22 +0800 Subject: [PATCH 32/33] test: synchronize result-aware batch lock race --- .../test_follow_up_run_binding.py | 36 +++++++++++-------- 1 file changed, 22 insertions(+), 14 deletions(-) diff --git a/tests/integration/agent_runtime/test_follow_up_run_binding.py b/tests/integration/agent_runtime/test_follow_up_run_binding.py index f940714..902ea76 100644 --- a/tests/integration/agent_runtime/test_follow_up_run_binding.py +++ b/tests/integration/agent_runtime/test_follow_up_run_binding.py @@ -20,7 +20,6 @@ from labpilot.api.result_aware_agent_adapter import SqlResultAwareAgentAdapter from labpilot.context.models import RepositorySnapshotRow from labpilot.coordinator.execution_adapters import VersionExecutionSemanticsAdapter -from labpilot.db.uow import UnitOfWork from labpilot.execution.models import BatchRow from labpilot.execution.trial_outcome import TrialExecutionOutcomeService from labpilot.experiments.proposal_models import ExperimentVersionRow @@ -145,7 +144,9 @@ async def test_missing_and_stale_summary_write_no_follow_up_run( @pytest.mark.integration async def test_batch_lock_linearizes_refresh_before_follow_up( - isolated_database_url: str, tmp_path: Path + isolated_database_url: str, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, ) -> None: async with _seed_result_attempt(isolated_database_url, tmp_path) as ( system, @@ -158,18 +159,25 @@ async def test_batch_lock_linearizes_refresh_before_follow_up( await _refresh(system, batch_id) summary, batch, version = await _state(system, batch_id) await _change_metric(system) - async with UnitOfWork(system.database.session) as lock: - await lock.session.scalar( - select(BatchRow).where(BatchRow.batch_id == batch_id).with_for_update() - ) - refresh_task = asyncio.create_task(_refresh(system, batch_id)) - await asyncio.sleep(0.05) - follow_up_task = asyncio.create_task( - _adapter(system).create_bound_run(_run(summary, batch, version)) - ) - await asyncio.sleep(0.05) - assert not refresh_task.done() and not follow_up_task.done() - await lock.commit() + summary_service = importlib.import_module("labpilot.results_validation.summary_service") + original_build_state = summary_service._build_state + refresh_has_batch_lock = asyncio.Event() + allow_refresh_to_continue = asyncio.Event() + + async def pause_after_batch_lock(*args: object, **kwargs: object) -> object: + refresh_has_batch_lock.set() + await allow_refresh_to_continue.wait() + return await original_build_state(*args, **kwargs) + + monkeypatch.setattr(summary_service, "_build_state", pause_after_batch_lock) + refresh_task = asyncio.create_task(_refresh(system, batch_id)) + await asyncio.wait_for(refresh_has_batch_lock.wait(), timeout=1) + follow_up_task = asyncio.create_task( + _adapter(system).create_bound_run(_run(summary, batch, version)) + ) + await asyncio.sleep(0) + assert not follow_up_task.done() + allow_refresh_to_continue.set() assert (await refresh_task).created is True with pytest.raises(AgentRuntimeError, match="source_summary_not_latest"): await follow_up_task From 631b76190934ed7f6566f49998ae35b7fc274cef Mon Sep 17 00:00:00 2001 From: Air000000 <1113476369@qq.com> Date: Tue, 22 Sep 2026 13:52:40 +0800 Subject: [PATCH 33/33] docs: close P1 result-aware iteration --- DOCUMENT-STATUS.md | 56 +++------------ ...ilot-p1-result-aware-iteration-closeout.md | 69 +++++++++++++++++++ 2 files changed, 80 insertions(+), 45 deletions(-) create mode 100644 docs/superpowers/evidence/2026-09-22-labpilot-p1-result-aware-iteration-closeout.md diff --git a/DOCUMENT-STATUS.md b/DOCUMENT-STATUS.md index aaada1d..bc525f8 100644 --- a/DOCUMENT-STATUS.md +++ b/DOCUMENT-STATUS.md @@ -1,6 +1,6 @@ # LabPilot 文档状态 -> 当前状态：R0 仍为 V1 rebaseline authority root；Slice 1 已 Accepted；Slice 2 bounded MVP 已验证并集成到 `main`，production hardening deferred；Slice 3 Result Plane implementation 已完成。Competition Core Demo Runtime/E2E 已在产品基线 `c92bf45b63053b1ae00209660ed351cf309f91c8` 上验收并冻结，competition console 已合并回 `main`，并行 Core Demo 工作中已验证的 exact read-only API workspace acceptance contract 也已收敛。比赛后 remaining-V1 已重新进入 planning gate：当前只授权 P1 Result-Aware Experiment Iteration 的设计 focused review，尚未授权实现。 +> 当前状态：R0 仍为 V1 rebaseline authority root；Slice 1 已 Accepted；Slice 2 bounded MVP 已验证并集成到 `main`，production hardening deferred；Slice 3 Result Plane implementation 已完成。Competition Core Demo Runtime/E2E 已验收并冻结。P1 Result-Aware Experiment Iteration implementation completed and accepted; deterministic two-iteration acceptance + regression/full-CI evidence closed。 | 文档 | 状态 | 用途 | |---|---|---| @@ -112,7 +112,12 @@ Slice-specific detailed plans must record their exact planning baseline commit a - P1 Result-Aware Iteration detailed implementation plan: **APPROVED / CURRENT** - P1 Result-Aware Iteration plan doc: `docs/superpowers/plans/2026-09-20-labpilot-p1-result-aware-iteration.md` - P1 Result-Aware Iteration focused plan review: **APPROVE**; P1-PLAN-001 CLOSED; direct regressions NONE -- P1 Result-Aware Iteration Product implementation: **AUTHORIZED under approved detailed plan** +- P1 Result-Aware Iteration implementation: **COMPLETED / ACCEPTED** +- P1 last production-code commit: `ae079be252fd62eff415ef040af4228ab45e2d61` +- P1 tranche acceptance/test baseline: `e8caaf668fed2b5e5878393fa0c47c001ba218e9` +- P1 deterministic two-iteration acceptance: **PASS** +- P1 Task 8 focused regression: **PASS** +- P1 Task 8 GitHub CI #312: **PASS** Under Sequencing Amendment v0.3.1, the bounded Slice 2 MVP closeout is sufficient to unlock Slice 3. It does not retroactively complete the deferred Slice 2 production-hardening campaign or establish Slice 2 production acceptance. @@ -122,51 +127,12 @@ The Git commit containing this status transition is the exact Slice3 implementat The competition workstream is closed. Its Runtime/E2E evidence and UI are frozen unless a concrete regression or demo-critical defect is demonstrated. -The detailed plan is approved. The next authorized remaining-V1 action is **implementation under the approved task sequence**: +P1 Result-Aware Experiment Iteration implementation is accepted. Repository integration is completed through the normal reviewed PR process; any further remaining-V1 tranche requires its own planning/authority gate. ```text -Approved P1 Result-Aware Experiment Iteration Design -→ approved detailed implementation plan -→ P1-A Tasks 1–4 -→ focused P1-A implementation review -→ P1-B Tasks 5–7 -→ focused P1-B implementation review -→ final regression/full-CI gate -``` - -Current proposed design: - -```text -exact BatchSummaryRevision -→ deterministic ResultContext -→ Follow-up AgentRun -→ Stop | Request Human Input | Follow-up Proposal -→ existing deterministic ProposalVerification -→ Human Approval -→ next immutable ExperimentVersion in the same Experiment -``` - -Focused design and focused plan review are closed. Product implementation is authorized only within the approved P1-A/B task sequence and review gates. - -Explicitly not authorized in this first P1 tranche: - -- Slice 4 as a new roadmap label; -- Multi-Agent; -- controlled/autonomous code modification; -- editable Worktree; -- cross-Experiment history search; -- automatic Batch start after Agent proposal; -- LLM semantic verifier; -- new Runtime-hardening campaign; -- competition UI redesign. - -The proposed design remains subordinate to: - -```text -V1 Scope v0.3 -> Architecture 10 v0.3 -> Master Roadmap v0.3 -> approved P1 Result-Aware authority after review +implementation accepted +→ reviewed PR integration +→ separate planning/authority gate for any further tranche ``` Competition closure still does not complete deferred Slice 2 production hardening or establish Slice 2 production acceptance. diff --git a/docs/superpowers/evidence/2026-09-22-labpilot-p1-result-aware-iteration-closeout.md b/docs/superpowers/evidence/2026-09-22-labpilot-p1-result-aware-iteration-closeout.md new file mode 100644 index 0000000..9956be9 --- /dev/null +++ b/docs/superpowers/evidence/2026-09-22-labpilot-p1-result-aware-iteration-closeout.md @@ -0,0 +1,69 @@ +# LabPilot P1 Result-Aware Experiment Iteration — Implementation Closeout + +Date: 2026-09-22 + +Status: Accepted + +## 1. Authority + +- Design: `docs/superpowers/specs/2026-09-20-labpilot-p1-result-aware-iteration-design.md` +- Implementation plan: `docs/superpowers/plans/2026-09-20-labpilot-p1-result-aware-iteration.md` +- PR: #9 +- Last production-code commit: `ae079be252fd62eff415ef040af4228ab45e2d61` +- Final pre-closeout acceptance/test baseline: `e8caaf668fed2b5e5878393fa0c47c001ba218e9` + +## 2. Implemented loop + +```text +Experiment E +→ ProposalRevision R1 +→ deterministic verification +→ Human Approval +→ ExperimentVersion V1 +→ Batch B1 +→ deterministic Result processing +→ exact BatchSummaryRevision S1 +→ Follow-up AgentRun A2 +→ Stop + OR +→ Follow-up ProposalRevision R2 in the same Experiment +→ deterministic verification +→ Human Approval +→ ExperimentVersion V2 +``` + +No automatic Batch is created for V2. + +## 3. Accepted invariants + +- Follow-up AgentRun pins the exact BatchSummaryRevision; admission accepts only latest-at-create under the BatchRow lock. +- A run already created against S(N) remains pinned when S(N+1) is later persisted. +- ResultContext loads the exact pinned summary and exposes selected valid evidence, exclusions, and Trial conditions. +- Raw file contents do not enter ResultContext; later raw MetricObservation mutation does not override the pinned summary. +- Follow-up Proposal stays in the same Experiment; R2 and V2 numbers increment. +- Proposal crash/retry returns the same R2 and does not create R3. +- Agent cannot auto-verify, auto-approve, or auto-create/start a Batch. +- The initial Agent path remains compatible and remote provider local-only egress remains blocked. + +## 4. Test and CI evidence + +- Task 7 deterministic acceptance: PASS. +- Task 8 focused regression: PASS. +- Task 8 concurrency barrier repeated 5x: PASS. +- Task 8 GitHub CI #312: SUCCESS. +- `./scripts/check_ci.sh`: repository-equivalent full gate PASS. +- Ruff: PASS; Mypy: PASS; unit/architecture: PASS; integration: PASS; dependency shell guard: PASS. + +## 5. Not established + +This closeout does NOT establish: + +- autonomous research Agent; +- automatic approval or next-Batch execution; +- controlled/autonomous code modification or editable Agent Worktree; +- Multi-Agent or cross-Experiment historical search; +- global exactly-once execution; +- Slice 2 production acceptance or production-scale scheduling/readiness; +- measured researcher productivity improvement. + +The product claim remains: bounded Agent-assisted Experiment Runtime / Harness.