From 94284b55509f5f04c31a6bf8b446fbb38b0c9d49 Mon Sep 17 00:00:00 2001 From: Curry Date: Sun, 26 Jul 2026 19:54:58 +0800 Subject: [PATCH 1/3] feat(backend): add workspace Source + immutable revisions, project SourceBinding + pinned revisions Implements ADR-0041 vertical slice: Workspace-owned Source/SourceRevision and Project-owned SourceBinding/SourceBindingRevision, distinct from the legacy DataSource and FeedProvider concepts (untouched). Bindings pin an exact SourceRevision and only drift on an explicit re-pin. New RBAC-gated endpoints under /workspaces/{id}/sources and /workspaces/{id}/projects/{id}/source-bindings. Single new Alembic head f3g4h5i6j7k8 on top of 1901f6da7138. --- backend/api/v1/__init__.py | 4 + backend/api/v1/project_source_bindings.py | 213 ++++++++++++++++++ backend/api/v1/workspace_sources.py | 148 ++++++++++++ ...4h5i6j7k8_add_source_and_source_binding.py | 119 ++++++++++ backend/models/__init__.py | 12 + backend/models/source_binding.py | 119 ++++++++++ backend/schemas/source_binding.py | 109 +++++++++ tests/unit/api/test_source_binding.py | 212 +++++++++++++++++ tests/unit/test_migration_heads.py | 2 +- 9 files changed, 937 insertions(+), 1 deletion(-) create mode 100644 backend/api/v1/project_source_bindings.py create mode 100644 backend/api/v1/workspace_sources.py create mode 100644 backend/migrations/versions/f3g4h5i6j7k8_add_source_and_source_binding.py create mode 100644 backend/models/source_binding.py create mode 100644 backend/schemas/source_binding.py create mode 100644 tests/unit/api/test_source_binding.py diff --git a/backend/api/v1/__init__.py b/backend/api/v1/__init__.py index 4c14d05..1e24187 100644 --- a/backend/api/v1/__init__.py +++ b/backend/api/v1/__init__.py @@ -24,6 +24,7 @@ plans, plugins, presets, + project_source_bindings, providers, records, schedules, @@ -37,6 +38,7 @@ webhooks, workers, workflows, + workspace_sources, workspaces, ) @@ -77,3 +79,5 @@ v1_router.include_router(system.router) v1_router.include_router(identity.router) v1_router.include_router(workspaces.router) +v1_router.include_router(workspace_sources.router) +v1_router.include_router(project_source_bindings.router) diff --git a/backend/api/v1/project_source_bindings.py b/backend/api/v1/project_source_bindings.py new file mode 100644 index 0000000..61124b4 --- /dev/null +++ b/backend/api/v1/project_source_bindings.py @@ -0,0 +1,213 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from backend.database import get_db +from backend.models.source_binding import Source, SourceBinding, SourceBindingRevision, SourceRevision +from backend.models.workflow import Project +from backend.schemas.common import ApiResponse +from backend.schemas.source_binding import ( + SourceBindingCreate, + SourceBindingRead, + SourceBindingRevisionCreate, + SourceBindingRevisionRead, + SourceBindingUpdate, +) +from backend.security.identity import RequestIdentity, get_request_identity +from backend.security.workspace_rbac import WorkspacePermission, get_workspace_access, require_permission + +router = APIRouter( + prefix="/workspaces/{workspace_id}/projects/{project_id}/source-bindings", + tags=["source-bindings"], +) + + +async def _get_project(db: AsyncSession, workspace_id: str, project_id: str) -> Project: + project = await db.scalar( + select(Project).where(Project.workspace_id == workspace_id, Project.id == project_id) + ) + if project is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Project not found") + return project + + +async def _get_source_in_workspace(db: AsyncSession, workspace_id: str, source_id: str) -> Source: + # Scoping the lookup to workspace_id (taken from THIS project's own path, + # never from the caller-supplied source_id) is the cross-workspace guard + # from ADR-0041: a Project can only bind Sources owned by its own Workspace. + source = await db.scalar( + select(Source).where(Source.workspace_id == workspace_id, Source.id == source_id) + ) + if source is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Source not found") + return source + + +async def _get_source_revision(db: AsyncSession, source_id: str, revision_number: int) -> SourceRevision: + revision = await db.scalar( + select(SourceRevision).where( + SourceRevision.source_id == source_id, + SourceRevision.revision_number == revision_number, + ) + ) + if revision is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Source revision not found") + return revision + + +async def _get_binding( + db: AsyncSession, project_id: str, binding_id: str, for_update: bool = False +) -> SourceBinding: + query = select(SourceBinding).where( + SourceBinding.project_id == project_id, SourceBinding.id == binding_id + ) + if for_update: + query = query.with_for_update() + binding = await db.scalar(query) + if binding is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Source binding not found") + return binding + + +@router.get("", response_model=ApiResponse[list[SourceBindingRead]]) +async def list_source_bindings( + workspace_id: str, + project_id: str, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.READ) + await _get_project(db, workspace_id, project_id) + rows = (await db.execute( + select(SourceBinding) + .where(SourceBinding.project_id == project_id) + .order_by(SourceBinding.created_at) + )).scalars().all() + return ApiResponse.ok([SourceBindingRead.model_validate(row) for row in rows]) + + +@router.post("", response_model=ApiResponse[SourceBindingRead], status_code=201) +async def create_source_binding( + workspace_id: str, + project_id: str, + body: SourceBindingCreate, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.MANAGE_CONFIGURATION) + await _get_project(db, workspace_id, project_id) + source = await _get_source_in_workspace(db, workspace_id, body.source_id) + pinned_revision = await _get_source_revision(db, source.id, body.source_revision_number) + + binding = SourceBinding( + project_id=project_id, + source_id=source.id, + name=body.name, + slug=body.slug, + current_revision_number=1, + created_by_user_id=access.user_id, + ) + db.add(binding) + await db.flush() + db.add( + SourceBindingRevision( + source_binding_id=binding.id, + revision_number=1, + pinned_source_revision_id=pinned_revision.id, + scope_config=body.scope_config, + created_by_user_id=access.user_id, + ) + ) + await db.flush() + return ApiResponse.ok(SourceBindingRead.model_validate(binding)) + + +@router.get("/{binding_id}", response_model=ApiResponse[SourceBindingRead]) +async def get_source_binding( + workspace_id: str, + project_id: str, + binding_id: str, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.READ) + await _get_project(db, workspace_id, project_id) + binding = await _get_binding(db, project_id, binding_id) + return ApiResponse.ok(SourceBindingRead.model_validate(binding)) + + +@router.patch("/{binding_id}", response_model=ApiResponse[SourceBindingRead]) +async def update_source_binding( + workspace_id: str, + project_id: str, + binding_id: str, + body: SourceBindingUpdate, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.MANAGE_CONFIGURATION) + await _get_project(db, workspace_id, project_id) + binding = await _get_binding(db, project_id, binding_id, for_update=True) + for field, value in body.model_dump(exclude_unset=True).items(): + setattr(binding, field, value) + await db.flush() + return ApiResponse.ok(SourceBindingRead.model_validate(binding)) + + +@router.get( + "/{binding_id}/revisions", response_model=ApiResponse[list[SourceBindingRevisionRead]] +) +async def list_source_binding_revisions( + workspace_id: str, + project_id: str, + binding_id: str, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.READ) + await _get_project(db, workspace_id, project_id) + await _get_binding(db, project_id, binding_id) + rows = (await db.execute( + select(SourceBindingRevision) + .where(SourceBindingRevision.source_binding_id == binding_id) + .order_by(SourceBindingRevision.revision_number) + )).scalars().all() + return ApiResponse.ok([SourceBindingRevisionRead.model_validate(row) for row in rows]) + + +@router.post( + "/{binding_id}/revisions", + response_model=ApiResponse[SourceBindingRevisionRead], + status_code=201, +) +async def create_source_binding_revision( + workspace_id: str, + project_id: str, + binding_id: str, + body: SourceBindingRevisionCreate, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.MANAGE_CONFIGURATION) + await _get_project(db, workspace_id, project_id) + binding = await _get_binding(db, project_id, binding_id, for_update=True) + pinned_revision = await _get_source_revision(db, binding.source_id, body.source_revision_number) + + next_revision = binding.current_revision_number + 1 + revision = SourceBindingRevision( + source_binding_id=binding.id, + revision_number=next_revision, + pinned_source_revision_id=pinned_revision.id, + scope_config=body.scope_config, + created_by_user_id=access.user_id, + ) + db.add(revision) + binding.current_revision_number = next_revision + await db.flush() + return ApiResponse.ok(SourceBindingRevisionRead.model_validate(revision)) diff --git a/backend/api/v1/workspace_sources.py b/backend/api/v1/workspace_sources.py new file mode 100644 index 0000000..fb6ad03 --- /dev/null +++ b/backend/api/v1/workspace_sources.py @@ -0,0 +1,148 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from backend.database import get_db +from backend.models.source_binding import Source, SourceRevision +from backend.schemas.common import ApiResponse +from backend.schemas.source_binding import ( + SourceCreate, + SourceRead, + SourceRevisionCreate, + SourceRevisionRead, + SourceUpdate, +) +from backend.security.identity import RequestIdentity, get_request_identity +from backend.security.workspace_rbac import WorkspacePermission, get_workspace_access, require_permission + +router = APIRouter(prefix="/workspaces/{workspace_id}/sources", tags=["sources"]) + + +async def _get_source(db: AsyncSession, workspace_id: str, source_id: str, for_update: bool = False) -> Source: + query = select(Source).where(Source.workspace_id == workspace_id, Source.id == source_id) + if for_update: + query = query.with_for_update() + source = await db.scalar(query) + if source is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Source not found") + return source + + +@router.get("", response_model=ApiResponse[list[SourceRead]]) +async def list_sources( + workspace_id: str, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.READ) + rows = (await db.execute( + select(Source).where(Source.workspace_id == workspace_id).order_by(Source.created_at) + )).scalars().all() + return ApiResponse.ok([SourceRead.model_validate(row) for row in rows]) + + +@router.post("", response_model=ApiResponse[SourceRead], status_code=201) +async def create_source( + workspace_id: str, + body: SourceCreate, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.MANAGE_CONFIGURATION) + source = Source( + workspace_id=workspace_id, + name=body.name, + slug=body.slug, + adapter_type=body.adapter_type, + description=body.description, + current_revision_number=1, + created_by_user_id=access.user_id, + ) + db.add(source) + await db.flush() + db.add( + SourceRevision( + source_id=source.id, + revision_number=1, + adapter_config=body.adapter_config, + created_by_user_id=access.user_id, + ) + ) + await db.flush() + return ApiResponse.ok(SourceRead.model_validate(source)) + + +@router.get("/{source_id}", response_model=ApiResponse[SourceRead]) +async def get_source( + workspace_id: str, + source_id: str, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.READ) + source = await _get_source(db, workspace_id, source_id) + return ApiResponse.ok(SourceRead.model_validate(source)) + + +@router.patch("/{source_id}", response_model=ApiResponse[SourceRead]) +async def update_source( + workspace_id: str, + source_id: str, + body: SourceUpdate, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.MANAGE_CONFIGURATION) + source = await _get_source(db, workspace_id, source_id, for_update=True) + for field, value in body.model_dump(exclude_unset=True).items(): + setattr(source, field, value) + await db.flush() + return ApiResponse.ok(SourceRead.model_validate(source)) + + +@router.get("/{source_id}/revisions", response_model=ApiResponse[list[SourceRevisionRead]]) +async def list_source_revisions( + workspace_id: str, + source_id: str, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.READ) + await _get_source(db, workspace_id, source_id) + rows = (await db.execute( + select(SourceRevision) + .where(SourceRevision.source_id == source_id) + .order_by(SourceRevision.revision_number) + )).scalars().all() + return ApiResponse.ok([SourceRevisionRead.model_validate(row) for row in rows]) + + +@router.post( + "/{source_id}/revisions", response_model=ApiResponse[SourceRevisionRead], status_code=201 +) +async def create_source_revision( + workspace_id: str, + source_id: str, + body: SourceRevisionCreate, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, WorkspacePermission.MANAGE_CONFIGURATION) + source = await _get_source(db, workspace_id, source_id, for_update=True) + next_revision = source.current_revision_number + 1 + revision = SourceRevision( + source_id=source.id, + revision_number=next_revision, + adapter_config=body.adapter_config, + created_by_user_id=access.user_id, + ) + db.add(revision) + source.current_revision_number = next_revision + await db.flush() + return ApiResponse.ok(SourceRevisionRead.model_validate(revision)) diff --git a/backend/migrations/versions/f3g4h5i6j7k8_add_source_and_source_binding.py b/backend/migrations/versions/f3g4h5i6j7k8_add_source_and_source_binding.py new file mode 100644 index 0000000..a6e264d --- /dev/null +++ b/backend/migrations/versions/f3g4h5i6j7k8_add_source_and_source_binding.py @@ -0,0 +1,119 @@ +"""add workspace sources and project source bindings + +Revision ID: f3g4h5i6j7k8 +Revises: 1901f6da7138 +Create Date: 2026-07-26 + +ADR-0041: a Workspace-owned Source (identity + mutable operational/health +status) gets immutable SourceRevisions for semantic adapter/endpoint/config +edits. A Project narrows a Source into a SourceBinding (authorization + +collection scope, also mutable for immediate credential/health/revocation +changes), which gets immutable SourceBindingRevisions that each pin an exact +SourceRevision explicitly — never the latest one implicitly. Adds only these +four tables; the legacy unscoped `data_sources` table (and its /sources API) +is untouched and remains the compatibility path for existing callers. +""" + +import sqlalchemy as sa +from alembic import op + +revision = "f3g4h5i6j7k8" +down_revision = "1901f6da7138" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "sources", + sa.Column("workspace_id", sa.String(36), nullable=False), + sa.Column("name", sa.String(255), nullable=False), + sa.Column("slug", sa.String(100), nullable=False), + sa.Column("adapter_type", sa.String(64), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column( + "status", + sa.Enum("active", "disabled", "revoked", name="source_lifecycle_status"), + nullable=False, + ), + sa.Column("current_revision_number", sa.Integer(), nullable=False), + sa.Column("created_by_user_id", sa.String(36), nullable=False), + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["workspace_id"], ["workspaces.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["created_by_user_id"], ["users.id"], ondelete="RESTRICT"), + sa.UniqueConstraint("workspace_id", "slug"), + ) + op.create_index("ix_sources_workspace_id", "sources", ["workspace_id"]) + + op.create_table( + "source_revisions", + sa.Column("source_id", sa.String(36), nullable=False), + sa.Column("revision_number", sa.Integer(), nullable=False), + sa.Column("adapter_config", sa.JSON(), nullable=False), + sa.Column("created_by_user_id", sa.String(36), nullable=False), + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["source_id"], ["sources.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["created_by_user_id"], ["users.id"], ondelete="RESTRICT"), + sa.UniqueConstraint("source_id", "revision_number"), + ) + op.create_index("ix_source_revisions_source_id", "source_revisions", ["source_id"]) + + op.create_table( + "source_bindings", + sa.Column("project_id", sa.String(36), nullable=False), + sa.Column("source_id", sa.String(36), nullable=False), + sa.Column("name", sa.String(255), nullable=False), + sa.Column("slug", sa.String(100), nullable=False), + sa.Column( + "status", + sa.Enum("active", "disabled", "revoked", name="source_binding_lifecycle_status"), + nullable=False, + ), + sa.Column("current_revision_number", sa.Integer(), nullable=False), + sa.Column("created_by_user_id", sa.String(36), nullable=False), + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["project_id"], ["projects.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["source_id"], ["sources.id"], ondelete="RESTRICT"), + sa.ForeignKeyConstraint(["created_by_user_id"], ["users.id"], ondelete="RESTRICT"), + sa.UniqueConstraint("project_id", "slug"), + ) + op.create_index("ix_source_bindings_project_id", "source_bindings", ["project_id"]) + op.create_index("ix_source_bindings_source_id", "source_bindings", ["source_id"]) + + op.create_table( + "source_binding_revisions", + sa.Column("source_binding_id", sa.String(36), nullable=False), + sa.Column("revision_number", sa.Integer(), nullable=False), + sa.Column("pinned_source_revision_id", sa.String(36), nullable=False), + sa.Column("scope_config", sa.JSON(), nullable=False), + sa.Column("created_by_user_id", sa.String(36), nullable=False), + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["source_binding_id"], ["source_bindings.id"], ondelete="CASCADE" + ), + sa.ForeignKeyConstraint( + ["pinned_source_revision_id"], ["source_revisions.id"], ondelete="RESTRICT" + ), + sa.ForeignKeyConstraint(["created_by_user_id"], ["users.id"], ondelete="RESTRICT"), + sa.UniqueConstraint("source_binding_id", "revision_number"), + ) + op.create_index( + "ix_source_binding_revisions_source_binding_id", + "source_binding_revisions", + ["source_binding_id"], + ) + + +def downgrade() -> None: + op.drop_table("source_binding_revisions") + op.drop_table("source_bindings") + op.drop_table("source_revisions") + op.drop_table("sources") diff --git a/backend/models/__init__.py b/backend/models/__init__.py index 3dd43e6..b1822a5 100644 --- a/backend/models/__init__.py +++ b/backend/models/__init__.py @@ -46,6 +46,13 @@ from backend.models.schedule import CronSchedule from backend.models.skill import Skill from backend.models.source import DataSource +from backend.models.source_binding import ( + Source, + SourceBinding, + SourceBindingRevision, + SourceLifecycleStatus, + SourceRevision, +) from backend.models.source_credential import SourceCredential from backend.models.source_cursor import SourceCursor from backend.models.source_measurement import SourceMeasurement @@ -102,6 +109,11 @@ "PlanSourceIndex", "PluginInstallation", "DataSource", + "Source", + "SourceRevision", + "SourceBinding", + "SourceBindingRevision", + "SourceLifecycleStatus", "SourceCredential", "SourceCursor", "SourceMeasurement", diff --git a/backend/models/source_binding.py b/backend/models/source_binding.py new file mode 100644 index 0000000..1926f15 --- /dev/null +++ b/backend/models/source_binding.py @@ -0,0 +1,119 @@ +"""Workspace-owned Source/SourceRevision and Project-owned SourceBinding/SourceBindingRevision. + +Per ADR-0041: Source is a reusable, Workspace-owned external endpoint; a Project +narrows it into a SourceBinding that scopes authorization and collection. Semantic +edits to either create a new immutable revision — a SourceBindingRevision always +pins an exact SourceRevision explicitly, it never silently follows the latest one. +Status changes (credential rotation, health, safety revocation) mutate the parent +row directly and do not require a new revision. + +This is a distinct concept from the legacy, unscoped `DataSource` +(backend/models/source.py, table `data_sources`, exposed at /sources) and from +`FeedProvider` (backend/models/feed_provider.py, the multi-project "Data Feed"). +Neither is touched or redefined by this module; both remain as-is. +""" + +from enum import StrEnum + +from sqlalchemy import Enum, ForeignKey, Integer, JSON, String, Text, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from backend.models.base import TimestampMixin + + +class SourceLifecycleStatus(StrEnum): + ACTIVE = "active" + DISABLED = "disabled" + REVOKED = "revoked" + + +class Source(TimestampMixin): + """Workspace-owned identity for a reusable external endpoint.""" + + __tablename__ = "sources" + __table_args__ = (UniqueConstraint("workspace_id", "slug"),) + + workspace_id: Mapped[str] = mapped_column( + ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False, index=True + ) + name: Mapped[str] = mapped_column(String(255), nullable=False) + slug: Mapped[str] = mapped_column(String(100), nullable=False) + adapter_type: Mapped[str] = mapped_column(String(64), nullable=False) + description: Mapped[str | None] = mapped_column(Text, nullable=True) + status: Mapped[SourceLifecycleStatus] = mapped_column( + Enum( + SourceLifecycleStatus, + name="source_lifecycle_status", + values_callable=lambda values: [v.value for v in values], + ), + nullable=False, + default=SourceLifecycleStatus.ACTIVE, + ) + current_revision_number: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + created_by_user_id: Mapped[str] = mapped_column( + ForeignKey("users.id", ondelete="RESTRICT"), nullable=False + ) + + +class SourceRevision(TimestampMixin): + """Immutable snapshot of a Source's adapter/endpoint/connection config.""" + + __tablename__ = "source_revisions" + __table_args__ = (UniqueConstraint("source_id", "revision_number"),) + + source_id: Mapped[str] = mapped_column( + ForeignKey("sources.id", ondelete="CASCADE"), nullable=False, index=True + ) + revision_number: Mapped[int] = mapped_column(Integer, nullable=False) + adapter_config: Mapped[dict] = mapped_column(JSON, nullable=False) + created_by_user_id: Mapped[str] = mapped_column( + ForeignKey("users.id", ondelete="RESTRICT"), nullable=False + ) + + +class SourceBinding(TimestampMixin): + """Project-owned authorization + collection scope over a Workspace Source.""" + + __tablename__ = "source_bindings" + __table_args__ = (UniqueConstraint("project_id", "slug"),) + + project_id: Mapped[str] = mapped_column( + ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, index=True + ) + source_id: Mapped[str] = mapped_column( + ForeignKey("sources.id", ondelete="RESTRICT"), nullable=False, index=True + ) + name: Mapped[str] = mapped_column(String(255), nullable=False) + slug: Mapped[str] = mapped_column(String(100), nullable=False) + status: Mapped[SourceLifecycleStatus] = mapped_column( + Enum( + SourceLifecycleStatus, + name="source_binding_lifecycle_status", + values_callable=lambda values: [v.value for v in values], + ), + nullable=False, + default=SourceLifecycleStatus.ACTIVE, + ) + current_revision_number: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + created_by_user_id: Mapped[str] = mapped_column( + ForeignKey("users.id", ondelete="RESTRICT"), nullable=False + ) + + +class SourceBindingRevision(TimestampMixin): + """Immutable pin of an exact SourceRevision plus the frozen scope it authorizes.""" + + __tablename__ = "source_binding_revisions" + __table_args__ = (UniqueConstraint("source_binding_id", "revision_number"),) + + source_binding_id: Mapped[str] = mapped_column( + ForeignKey("source_bindings.id", ondelete="CASCADE"), nullable=False, index=True + ) + revision_number: Mapped[int] = mapped_column(Integer, nullable=False) + pinned_source_revision_id: Mapped[str] = mapped_column( + ForeignKey("source_revisions.id", ondelete="RESTRICT"), nullable=False + ) + scope_config: Mapped[dict] = mapped_column(JSON, nullable=False) + created_by_user_id: Mapped[str] = mapped_column( + ForeignKey("users.id", ondelete="RESTRICT"), nullable=False + ) diff --git a/backend/schemas/source_binding.py b/backend/schemas/source_binding.py new file mode 100644 index 0000000..2f2a6b8 --- /dev/null +++ b/backend/schemas/source_binding.py @@ -0,0 +1,109 @@ +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field + +from backend.schemas.common import UTCModel + + +class SourceCreate(BaseModel): + model_config = ConfigDict(extra="forbid") + + name: str = Field(min_length=1, max_length=255) + slug: str = Field(min_length=1, max_length=100) + adapter_type: str = Field(min_length=1, max_length=64) + description: str | None = Field(default=None, max_length=4000) + adapter_config: dict = Field(default_factory=dict) + + +class SourceUpdate(BaseModel): + model_config = ConfigDict(extra="forbid") + + name: str | None = Field(default=None, min_length=1, max_length=255) + description: str | None = Field(default=None, max_length=4000) + status: str | None = None + + +class SourceRead(UTCModel): + id: str + workspace_id: str + name: str + slug: str + adapter_type: str + description: str | None + status: str + current_revision_number: int + created_by_user_id: str + created_at: datetime + updated_at: datetime + + model_config = {"from_attributes": True} + + +class SourceRevisionCreate(BaseModel): + model_config = ConfigDict(extra="forbid") + + adapter_config: dict = Field(default_factory=dict) + + +class SourceRevisionRead(UTCModel): + id: str + source_id: str + revision_number: int + adapter_config: dict + created_by_user_id: str + created_at: datetime + updated_at: datetime + + model_config = {"from_attributes": True} + + +class SourceBindingCreate(BaseModel): + model_config = ConfigDict(extra="forbid") + + source_id: str + name: str = Field(min_length=1, max_length=255) + slug: str = Field(min_length=1, max_length=100) + source_revision_number: int = Field(ge=1) + scope_config: dict = Field(default_factory=dict) + + +class SourceBindingUpdate(BaseModel): + model_config = ConfigDict(extra="forbid") + + name: str | None = Field(default=None, min_length=1, max_length=255) + status: str | None = None + + +class SourceBindingRead(UTCModel): + id: str + project_id: str + source_id: str + name: str + slug: str + status: str + current_revision_number: int + created_by_user_id: str + created_at: datetime + updated_at: datetime + + model_config = {"from_attributes": True} + + +class SourceBindingRevisionCreate(BaseModel): + model_config = ConfigDict(extra="forbid") + + source_revision_number: int = Field(ge=1) + scope_config: dict = Field(default_factory=dict) + + +class SourceBindingRevisionRead(UTCModel): + id: str + source_binding_id: str + revision_number: int + pinned_source_revision_id: str + scope_config: dict + created_by_user_id: str + created_at: datetime + updated_at: datetime + + model_config = {"from_attributes": True} diff --git a/tests/unit/api/test_source_binding.py b/tests/unit/api/test_source_binding.py new file mode 100644 index 0000000..32fc641 --- /dev/null +++ b/tests/unit/api/test_source_binding.py @@ -0,0 +1,212 @@ +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient + +from backend.api.v1.project_source_bindings import router as source_bindings_router +from backend.api.v1.workspace_sources import router as sources_router +from backend.database import get_db +from backend.models.identity import User, Workspace, WorkspaceMembership, WorkspaceRole +from backend.models.workflow import Project +from backend.security.identity import RequestIdentity, get_request_identity + + +def _build_client(db_session, user): + app = FastAPI() + app.include_router(sources_router) + app.include_router(source_bindings_router) + + async def override_db(): + yield db_session + + async def override_identity(): + return RequestIdentity(subject=user.subject) + + app.dependency_overrides[get_db] = override_db + app.dependency_overrides[get_request_identity] = override_identity + return AsyncClient(transport=ASGITransport(app=app), base_url="http://test") + + +async def _seed_workspace_admin(db_session, user, workspace_name: str, workspace_slug: str) -> Workspace: + workspace = Workspace(name=workspace_name, slug=workspace_slug) + db_session.add(workspace) + await db_session.flush() + db_session.add( + WorkspaceMembership(workspace_id=workspace.id, user_id=user.id, role=WorkspaceRole.ADMIN) + ) + await db_session.commit() + return workspace + + +async def test_source_is_workspace_owned_and_not_visible_from_another_workspace(db_session): + user = User(subject="source-owner") + db_session.add(user) + await db_session.flush() + workspace_a = await _seed_workspace_admin(db_session, user, "Workspace A", "workspace-a") + workspace_b = await _seed_workspace_admin(db_session, user, "Workspace B", "workspace-b") + + async with _build_client(db_session, user) as client: + created = await client.post( + f"/workspaces/{workspace_a.id}/sources", + json={ + "name": "RSS Feed", + "slug": "rss-feed", + "adapter_type": "rss", + "adapter_config": {"feed_url": "https://example.com/feed.xml"}, + }, + ) + assert created.status_code == 201 + source_id = created.json()["data"]["id"] + + found_in_owner = await client.get(f"/workspaces/{workspace_a.id}/sources/{source_id}") + assert found_in_owner.status_code == 200 + assert found_in_owner.json()["data"]["workspace_id"] == workspace_a.id + + not_found_elsewhere = await client.get(f"/workspaces/{workspace_b.id}/sources/{source_id}") + assert not_found_elsewhere.status_code == 404 + + +async def test_source_revisions_are_immutable_and_accumulate(db_session): + user = User(subject="revision-owner") + db_session.add(user) + await db_session.flush() + workspace = await _seed_workspace_admin(db_session, user, "Workspace", "workspace-rev") + + async with _build_client(db_session, user) as client: + created = await client.post( + f"/workspaces/{workspace.id}/sources", + json={ + "name": "OpenCLI Source", + "slug": "opencli-source", + "adapter_type": "opencli", + "adapter_config": {"endpoint": "https://one.example.com"}, + }, + ) + source_id = created.json()["data"]["id"] + assert created.json()["data"]["current_revision_number"] == 1 + + new_revision = await client.post( + f"/workspaces/{workspace.id}/sources/{source_id}/revisions", + json={"adapter_config": {"endpoint": "https://two.example.com"}}, + ) + assert new_revision.status_code == 201 + assert new_revision.json()["data"]["revision_number"] == 2 + + revisions = ( + await client.get(f"/workspaces/{workspace.id}/sources/{source_id}/revisions") + ).json()["data"] + assert [r["revision_number"] for r in revisions] == [1, 2] + assert revisions[0]["adapter_config"] == {"endpoint": "https://one.example.com"} + assert revisions[1]["adapter_config"] == {"endpoint": "https://two.example.com"} + + source_after = ( + await client.get(f"/workspaces/{workspace.id}/sources/{source_id}") + ).json()["data"] + assert source_after["current_revision_number"] == 2 + + +async def test_binding_pins_exact_revision_and_does_not_silently_drift(db_session): + user = User(subject="binding-owner") + db_session.add(user) + await db_session.flush() + workspace = await _seed_workspace_admin(db_session, user, "Workspace", "workspace-bind") + project = Project( + workspace_id=workspace.id, name="Project", slug="project", created_by_user_id=user.id + ) + db_session.add(project) + await db_session.commit() + + async with _build_client(db_session, user) as client: + source = ( + await client.post( + f"/workspaces/{workspace.id}/sources", + json={ + "name": "Source", + "slug": "source", + "adapter_type": "rss", + "adapter_config": {"feed_url": "https://v1.example.com"}, + }, + ) + ).json()["data"] + + binding = ( + await client.post( + f"/workspaces/{workspace.id}/projects/{project.id}/source-bindings", + json={ + "source_id": source["id"], + "name": "Binding", + "slug": "binding", + "source_revision_number": 1, + "scope_config": {"targets": ["*"]}, + }, + ) + ).json()["data"] + assert binding["current_revision_number"] == 1 + + # Source gets a new revision after the binding is created. + await client.post( + f"/workspaces/{workspace.id}/sources/{source['id']}/revisions", + json={"adapter_config": {"feed_url": "https://v2.example.com"}}, + ) + + # The binding must still be pinned to revision 1 — no silent drift. + binding_revisions = ( + await client.get( + f"/workspaces/{workspace.id}/projects/{project.id}/source-bindings/{binding['id']}/revisions" + ) + ).json()["data"] + assert len(binding_revisions) == 1 + assert binding_revisions[0]["revision_number"] == 1 + + source_revisions = ( + await client.get(f"/workspaces/{workspace.id}/sources/{source['id']}/revisions") + ).json()["data"] + rev1_id = next(r["id"] for r in source_revisions if r["revision_number"] == 1) + rev2_id = next(r["id"] for r in source_revisions if r["revision_number"] == 2) + assert binding_revisions[0]["pinned_source_revision_id"] == rev1_id + assert binding_revisions[0]["pinned_source_revision_id"] != rev2_id + + # Explicit re-pin to revision 2 creates a new immutable binding revision. + repinned = await client.post( + f"/workspaces/{workspace.id}/projects/{project.id}/source-bindings/{binding['id']}/revisions", + json={"source_revision_number": 2, "scope_config": {"targets": ["*"]}}, + ) + assert repinned.status_code == 201 + assert repinned.json()["data"]["revision_number"] == 2 + assert repinned.json()["data"]["pinned_source_revision_id"] == rev2_id + + +async def test_project_cannot_bind_source_from_another_workspace(db_session): + user = User(subject="cross-workspace-user") + db_session.add(user) + await db_session.flush() + workspace_a = await _seed_workspace_admin(db_session, user, "Workspace A", "cross-a") + workspace_b = await _seed_workspace_admin(db_session, user, "Workspace B", "cross-b") + project_b = Project( + workspace_id=workspace_b.id, name="Project B", slug="project-b", created_by_user_id=user.id + ) + db_session.add(project_b) + await db_session.commit() + + async with _build_client(db_session, user) as client: + source_a = ( + await client.post( + f"/workspaces/{workspace_a.id}/sources", + json={ + "name": "Source A", + "slug": "source-a", + "adapter_type": "rss", + "adapter_config": {"feed_url": "https://a.example.com"}, + }, + ) + ).json()["data"] + + rejected = await client.post( + f"/workspaces/{workspace_b.id}/projects/{project_b.id}/source-bindings", + json={ + "source_id": source_a["id"], + "name": "Cross Binding", + "slug": "cross-binding", + "source_revision_number": 1, + "scope_config": {}, + }, + ) + assert rejected.status_code == 404 diff --git a/tests/unit/test_migration_heads.py b/tests/unit/test_migration_heads.py index af60639..3f963ee 100644 --- a/tests/unit/test_migration_heads.py +++ b/tests/unit/test_migration_heads.py @@ -13,7 +13,7 @@ def test_alembic_has_one_head(): config = Config() config.set_main_option("script_location", "backend/migrations") - assert ScriptDirectory.from_config(config).get_heads() == ["1901f6da7138"] + assert ScriptDirectory.from_config(config).get_heads() == ["f3g4h5i6j7k8"] def test_upgrade_head_creates_identity_and_operations_tables(monkeypatch): From 94e0cf1b8016636931508d7f65c4946a1330e82c Mon Sep 17 00:00:00 2001 From: Curry Date: Sun, 26 Jul 2026 19:48:55 +0800 Subject: [PATCH 2/3] feat: govern chat writes through agent control --- backend/api/v1/chat.py | 292 +++++----- backend/control/agent_control.py | 655 +++++++++++++++++++++++ tests/integration/test_chat_api.py | 140 ++++- tests/unit/control/test_agent_control.py | 23 + 4 files changed, 957 insertions(+), 153 deletions(-) create mode 100644 backend/control/agent_control.py create mode 100644 tests/unit/control/test_agent_control.py diff --git a/backend/api/v1/chat.py b/backend/api/v1/chat.py index 471e3fd..6e49fd8 100644 --- a/backend/api/v1/chat.py +++ b/backend/api/v1/chat.py @@ -6,7 +6,7 @@ - 只读工具 (list_sources) 直接执行, 喂回结果让 agent 继续推理。 - 写工具 (toggle_source) **不立即落库**, 返回一个 proposal 让前端弹 diff 确认。 -确认后前端调 /chat/confirm, 这里才走现有 source_service 落库。写前确认是硬底线。 +确认后前端调 /chat/confirm, 这里才走统一 Agent Control 服务落库。写前确认是硬底线。 v1 薄闭环: 唯一写动作 = 启停 source。验证通后按同模式扩 trigger_task / update_schedule。 """ @@ -16,16 +16,16 @@ import re from typing import Any, Literal, Optional -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from backend.control.agent_control import ACTION_REGISTRY, agent_control_service from backend.database import get_db from backend.models.provider import ModelProvider from backend.schemas.common import ApiResponse -from backend.schemas.schedule import CronScheduleUpdate -from backend.schemas.source import DataSourceUpdate +from backend.security.identity import RequestIdentity, get_request_identity from backend.services import schedule_service, source_service, task_service from backend.skills.toolcall import _is_xml_tool_model, _parse_tool_use, _safe_json @@ -142,7 +142,22 @@ }, ] -WRITE_TOOLS = {"toggle_source", "trigger_task", "update_schedule", "update_provider"} +WRITE_TOOLS = ACTION_REGISTRY.action_names + + +async def _optional_request_identity(request: Request) -> RequestIdentity | None: + """Preserve unauthenticated read-chat compatibility; writes still fail closed.""" + + scheme, _, token = request.headers.get("authorization", "").partition(" ") + if scheme.lower() != "bearer" or not token: + return None + return await get_request_identity(request) + + +def _require_write_identity(identity: RequestIdentity | None) -> RequestIdentity: + if identity is None: + raise HTTPException(status_code=401, detail="Bearer token required for write proposals") + return identity # ── request / response 模型 ───────────────────────────────────────────────── @@ -163,6 +178,9 @@ class Proposal(BaseModel): args: dict[str, Any] summary: str diff: str + work_item_id: Optional[str] = None + workspace_id: Optional[str] = None + proposal_version: Optional[str] = None class ChatReply(BaseModel): @@ -270,83 +288,65 @@ async def _run_read_tool(db: AsyncSession, name: str, args: dict[str, Any]) -> A return {"error": f"unknown read tool: {name}"} -async def _build_proposal(db: AsyncSession, name: str, args: dict[str, Any]) -> Proposal: - if name == "toggle_source": - source_id = args.get("source_id", "") - enabled = bool(args.get("enabled")) - source = await source_service.get_source(db, source_id) - if not source: - raise HTTPException(status_code=404, detail=f"数据源 {source_id} 不存在") - verb = "启用" if enabled else "停用" - return Proposal( - tool=name, - args={"source_id": source_id, "enabled": enabled}, - summary=f"{verb}数据源「{source.name}」", - diff=f"{source.name}: enabled {source.enabled} → {enabled}", - ) - if name == "trigger_task": - source_id = args.get("source_id", "") - source = await source_service.get_source(db, source_id) - if not source: - raise HTTPException(status_code=404, detail=f"数据源 {source_id} 不存在") +def _workspace_id(context: dict[str, Any] | None) -> str | None: + if not context: + return None + value = context.get("workspace_id") + if not isinstance(value, str) or not value.strip(): + return None + return value + + +async def _build_proposal( + db: AsyncSession, + name: str, + args: dict[str, Any], + *, + identity: RequestIdentity | None = None, + workspace_id: str | None = None, +) -> Proposal: + """Preview an action and, for authenticated transports, persist its proposal.""" + + if identity is None: + # Kept for internal callers that only need the existing preview shape. + preview = await agent_control_service.preview(db, name, args) return Proposal( - tool=name, - args={"source_id": source_id}, - summary=f"立即采集「{source.name}」", - diff=f"触发一次手动采集: {source.name} ({'已启用' if source.enabled else '已停用'})", + tool=preview.action_name, + args=preview.args, + summary=preview.summary, + diff=preview.diff, ) - if name == "update_schedule": - schedule_id = args.get("schedule_id", "") - schedule = await schedule_service.get_schedule(db, schedule_id) - if not schedule: - raise HTTPException(status_code=404, detail=f"调度 {schedule_id} 不存在") - out_args: dict[str, Any] = {"schedule_id": schedule_id} - changes: list[str] = [] - if args.get("cron_expression") is not None: - new_cron = str(args["cron_expression"]) - if not schedule_service.validate_cron_expression(new_cron): - raise HTTPException(status_code=400, detail=f"非法 cron 表达式: {new_cron}") - out_args["cron_expression"] = new_cron - changes.append(f"cron {schedule.cron_expression} → {new_cron}") - if args.get("enabled") is not None: - out_args["enabled"] = bool(args["enabled"]) - changes.append(f"enabled {schedule.enabled} → {bool(args['enabled'])}") - if not changes: - raise HTTPException(status_code=400, detail="update_schedule 未指定要改的字段 (cron_expression 或 enabled)") - return Proposal( - tool=name, - args=out_args, - summary=f"修改调度「{schedule.name}」", - diff="; ".join(changes), - ) - if name == "update_provider": - provider_id = args.get("provider_id", "") - provider = await db.get(ModelProvider, provider_id) - if not provider: - raise HTTPException(status_code=404, detail=f"模型提供商 {provider_id} 不存在") - out_args: dict[str, Any] = {"provider_id": provider_id} - changes: list[str] = [] - if args.get("default_model") is not None: - new_model = str(args["default_model"]) - out_args["default_model"] = new_model - changes.append(f"default_model {provider.default_model} → {new_model}") - if args.get("enabled") is not None: - out_args["enabled"] = bool(args["enabled"]) - state = "启用" if out_args["enabled"] else "停用" - changes.append(f"{state} (enabled {provider.enabled} → {out_args['enabled']})") - if not changes: - raise HTTPException(status_code=400, detail="update_provider 未指定要改的字段 (default_model 或 enabled)") - return Proposal( - tool=name, - args=out_args, - summary=f"配置 AI 模型提供商「{provider.name}」", - diff="; ".join(changes), - ) - raise HTTPException(status_code=400, detail=f"unknown write tool: {name}") + + resolved_workspace_id = await agent_control_service.resolve_workspace_id( + db, + identity, + workspace_id, + ) + recorded = await agent_control_service.create_proposal( + db, + workspace_id=resolved_workspace_id, + identity=identity, + action_name=name, + args=args, + origin="chat", + ) + return Proposal( + tool=recorded.preview.action_name, + args=recorded.preview.args, + summary=recorded.preview.summary, + diff=recorded.preview.diff, + work_item_id=recorded.work_item_id, + workspace_id=recorded.workspace_id, + proposal_version=recorded.proposal_version, + ) @router.post("", response_model=ApiResponse[ChatReply]) -async def chat(body: ChatRequest, db: AsyncSession = Depends(get_db)) -> ApiResponse: +async def chat( + body: ChatRequest, + identity: RequestIdentity | None = Depends(_optional_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: provider = await _pick_provider(db, body.provider_id) client = await _build_client(provider) model = provider.default_model or "gpt-4o-mini" @@ -356,7 +356,7 @@ async def chat(body: ChatRequest, db: AsyncSession = Depends(get_db)) -> ApiResp system += f"\n\n当前用户操作上下文 (JSON): {json.dumps(body.context, ensure_ascii=False)}" if _is_xml_tool_model(model): - return await _chat_xml(client, model, system, body, db) + return await _chat_xml(client, model, system, body, db, identity) messages: list[dict[str, Any]] = [{"role": "system", "content": system}] messages += [{"role": m.role, "content": m.content} for m in body.messages] @@ -380,7 +380,13 @@ async def chat(body: ChatRequest, db: AsyncSession = Depends(get_db)) -> ApiResp for tc in tool_calls: if tc.function.name in WRITE_TOOLS: args = _safe_json(tc.function.arguments) - proposal = await _build_proposal(db, tc.function.name, args) + proposal = await _build_proposal( + db, + tc.function.name, + args, + identity=_require_write_identity(identity), + workspace_id=_workspace_id(body.context), + ) return ApiResponse.ok(ChatReply(type="proposal", proposal=proposal)) # 只读工具 → 执行, 喂回结果, 继续循环 @@ -408,78 +414,53 @@ async def chat(body: ChatRequest, db: AsyncSession = Depends(get_db)) -> ApiResp @router.post("/confirm", response_model=ApiResponse[dict]) -async def confirm(body: ConfirmRequest, db: AsyncSession = Depends(get_db)) -> ApiResponse: - """Execute a confirmed proposal. Dispatches by tool; each writes via existing services.""" +async def confirm( + body: ConfirmRequest, + identity: RequestIdentity = Depends(get_request_identity), + db: AsyncSession = Depends(get_db), +) -> ApiResponse: + """Execute a proposal only through the confirmed Agent Control path.""" proposal = body.proposal - args = proposal.args - - if proposal.tool == "toggle_source": - source = await source_service.get_source(db, args.get("source_id", "")) - if not source: - raise HTTPException(status_code=404, detail="数据源不存在") - await source_service.update_source(db, source, DataSourceUpdate(enabled=bool(args.get("enabled")))) - await db.commit() - logger.info("chat confirm | toggle_source %s -> %s", source.id, args.get("enabled")) - return ApiResponse.ok({"applied": True, "tool": proposal.tool, "summary": proposal.summary}) - - if proposal.tool == "trigger_task": - source = await source_service.get_source(db, args.get("source_id", "")) - if not source: - raise HTTPException(status_code=404, detail="数据源不存在") - if not source.enabled: - raise HTTPException(status_code=400, detail="数据源已停用, 无法采集") - task = await task_service.create_task( - db, source_id=source.id, trigger_type="manual", parameters={}, priority=0, agent_id=None - ) - await db.commit() - from backend.executor import get_executor + workspace_id = await agent_control_service.resolve_workspace_id( + db, + identity, + proposal.workspace_id, + ) + work_item_id = proposal.work_item_id + proposal_version = proposal.proposal_version - try: - dispatch = await get_executor().dispatch_collection(task.id, {}) - except Exception as exc: - # Task row is already committed; surface the dispatch failure instead - # of reporting applied=True with a silently dead task. - logger.exception("chat confirm | trigger_task dispatch failed source=%s task=%s", source.id, task.id) - raise HTTPException( - status_code=502, detail=f"任务已创建但派发失败 (task_id={task.id}), 请到工作项里重试" - ) from exc - logger.info("chat confirm | trigger_task source=%s task=%s", source.id, task.id) - return ApiResponse.ok( - { - "applied": True, - "tool": proposal.tool, - "task_id": task.id, - "summary": proposal.summary, - "dispatch": dispatch, - } + if (work_item_id is None) != (proposal_version is None): + raise HTTPException( + status_code=409, + detail="Agent Control proposal metadata is incomplete", ) - - if proposal.tool == "update_schedule": - schedule = await schedule_service.get_schedule(db, args.get("schedule_id", "")) - if not schedule: - raise HTTPException(status_code=404, detail="调度不存在") - fields = {k: args[k] for k in ("cron_expression", "enabled") if k in args} - await schedule_service.update_schedule(db, schedule, CronScheduleUpdate(**fields)) - await db.commit() - logger.info("chat confirm | update_schedule %s %s", schedule.id, fields) - return ApiResponse.ok({"applied": True, "tool": proposal.tool, "summary": proposal.summary}) - - if proposal.tool == "update_provider": - provider = await db.get(ModelProvider, args.get("provider_id", "")) - if not provider: - raise HTTPException(status_code=404, detail="模型提供商不存在") - if "default_model" in args: - provider.default_model = str(args["default_model"]) - if "enabled" in args: - provider.enabled = bool(args["enabled"]) - await db.commit() - logger.info( - "chat confirm | update_provider %s %s", - provider.id, {k: args[k] for k in ("default_model", "enabled") if k in args}, + if work_item_id is None: + # Compatibility for clients that still send the original Proposal + # shape. The confirmation endpoint itself is the explicit gate, so + # persist the governed proposal immediately before executing it. + recorded = await agent_control_service.create_proposal( + db, + workspace_id=workspace_id, + identity=identity, + action_name=proposal.tool, + args=proposal.args, + origin="chat.confirm.compat", ) - return ApiResponse.ok({"applied": True, "tool": proposal.tool, "summary": proposal.summary}) - - raise HTTPException(status_code=400, detail=f"unknown proposal tool: {proposal.tool}") + work_item_id = recorded.work_item_id + proposal_version = recorded.proposal_version + + assert work_item_id is not None + assert proposal_version is not None + result = await agent_control_service.execute_confirmed( + db, + workspace_id=workspace_id, + identity=identity, + work_item_id=work_item_id, + proposal_version=proposal_version, + confirmation_path="chat.confirm", + expected_action=proposal.tool, + ) + return ApiResponse.ok(result) # ── XML-style tool models (e.g. Qwable-v1: emits XML, not OpenAI tool_calls) ── @@ -506,7 +487,14 @@ async def confirm(body: ConfirmRequest, db: AsyncSession = Depends(get_db)) -> A ) -async def _chat_xml(client: Any, model: str, system: str, body: ChatRequest, db: AsyncSession) -> ApiResponse: +async def _chat_xml( + client: Any, + model: str, + system: str, + body: ChatRequest, + db: AsyncSession, + identity: RequestIdentity | None, +) -> ApiResponse: """Tool loop for XML-style models (parse from content, feed results back as text).""" messages: list[dict[str, Any]] = [{"role": "system", "content": system + XML_TOOL_TEXT}] messages += [{"role": m.role, "content": m.content} for m in body.messages] @@ -528,7 +516,13 @@ async def _chat_xml(client: Any, model: str, system: str, body: ChatRequest, db: # write tool hit → return proposal immediately for name, args in calls: if name in WRITE_TOOLS: - proposal = await _build_proposal(db, name, args) + proposal = await _build_proposal( + db, + name, + args, + identity=_require_write_identity(identity), + workspace_id=_workspace_id(body.context), + ) return ApiResponse.ok(ChatReply(type="proposal", proposal=proposal)) # read tools → execute, feed results back as text, loop diff --git a/backend/control/agent_control.py b/backend/control/agent_control.py new file mode 100644 index 0000000..c4be9ee --- /dev/null +++ b/backend/control/agent_control.py @@ -0,0 +1,655 @@ +"""Govern agent-originated mutations through one proposal and execution path. + +The registry in this module is the single source of truth for Agent Control +write actions. Transports (the first-party chat dock today, MCP/SDK adapters +later) may preview an action, but only :class:`AgentControlService` may persist +and execute it. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import uuid +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any + +from fastapi import HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from backend.models.identity import User, Workspace, WorkspaceMembership +from backend.models.operations_work_item import ( + OperationsWorkItem, + Priority, + Severity, + WorkItemStatus, + WorkItemType, +) +from backend.models.provider import ModelProvider +from backend.schemas.schedule import CronScheduleUpdate +from backend.schemas.source import DataSourceUpdate +from backend.security.identity import RequestIdentity +from backend.security.workspace_rbac import ( + WorkspaceAccess, + WorkspacePermission, + get_workspace_access, + require_permission, +) +from backend.services import schedule_service, source_service, task_service + +logger = logging.getLogger(__name__) + +EVIDENCE_SCHEMA_VERSION = "agent-control-evidence/v1" +POLICY_STATE_VERSION = "agent-control-policy/v1" + + +@dataclass(frozen=True) +class ActionPreview: + action_name: str + args: dict[str, Any] + summary: str + diff: str + target_kind: str + target_id: str + target_resource_version: str + + +@dataclass(frozen=True) +class RecordedActionProposal: + work_item_id: str + workspace_id: str + proposal_version: str + preview: ActionPreview + + +PrepareAction = Callable[[AsyncSession, dict[str, Any]], Awaitable[ActionPreview]] +ExecuteAction = Callable[[AsyncSession, dict[str, Any]], Awaitable[dict[str, Any]]] + + +@dataclass(frozen=True) +class RegisteredAction: + name: str + permission: WorkspacePermission + severity: Severity + prepare: PrepareAction + execute: ExecuteAction + + +class AgentControlActionRegistry: + """Reusable registry for every governed Agent Control mutation.""" + + def __init__(self) -> None: + self._actions: dict[str, RegisteredAction] = {} + + def register(self, action: RegisteredAction) -> None: + if action.name in self._actions: + raise ValueError(f"Agent Control action already registered: {action.name}") + self._actions[action.name] = action + + def get(self, action_name: str) -> RegisteredAction: + action = self._actions.get(action_name) + if action is None: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + f"unknown proposal tool: {action_name}", + ) + return action + + @property + def action_names(self) -> frozenset[str]: + return frozenset(self._actions) + + +class CommittedActionError(Exception): + """An action committed authoritative state but its follow-up failed.""" + + def __init__(self, *, status_code: int, detail: str, result: dict[str, Any]) -> None: + super().__init__(detail) + self.status_code = status_code + self.detail = detail + self.result = result + + +def _resource_version( + target_kind: str, + target_id: str, + updated_at: datetime, + state: dict[str, Any], +) -> str: + payload = { + "target_kind": target_kind, + "target_id": target_id, + "updated_at": updated_at.isoformat(), + "state": state, + } + digest = hashlib.sha256( + json.dumps(payload, ensure_ascii=True, sort_keys=True, default=str).encode() + ).hexdigest() + return f"{target_kind}-state/v1:{digest}" + + +def _permission_state_version( + workspace_id: str, + access: WorkspaceAccess, +) -> str: + payload = f"workspace-rbac/v1:{workspace_id}:{access.user_id}:{access.role.value}" + return f"workspace-permission/v1:{hashlib.sha256(payload.encode()).hexdigest()}" + + +async def _prepare_toggle_source( + db: AsyncSession, + args: dict[str, Any], +) -> ActionPreview: + source_id = str(args.get("source_id", "")) + enabled = bool(args.get("enabled")) + source = await source_service.get_source(db, source_id) + if source is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, f"数据源 {source_id} 不存在") + verb = "启用" if enabled else "停用" + return ActionPreview( + action_name="toggle_source", + args={"source_id": source_id, "enabled": enabled}, + summary=f"{verb}数据源「{source.name}」", + diff=f"{source.name}: enabled {source.enabled} → {enabled}", + target_kind="source", + target_id=source.id, + target_resource_version=_resource_version( + "source", + source.id, + source.updated_at, + {"enabled": source.enabled}, + ), + ) + + +async def _execute_toggle_source( + db: AsyncSession, + args: dict[str, Any], +) -> dict[str, Any]: + source = await source_service.get_source(db, str(args.get("source_id", ""))) + if source is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "数据源不存在") + await source_service.update_source( + db, + source, + DataSourceUpdate.model_validate({"enabled": bool(args.get("enabled"))}), + ) + return {} + + +async def _prepare_trigger_task( + db: AsyncSession, + args: dict[str, Any], +) -> ActionPreview: + source_id = str(args.get("source_id", "")) + source = await source_service.get_source(db, source_id) + if source is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, f"数据源 {source_id} 不存在") + return ActionPreview( + action_name="trigger_task", + args={"source_id": source_id}, + summary=f"立即采集「{source.name}」", + diff=f"触发一次手动采集: {source.name} ({'已启用' if source.enabled else '已停用'})", + target_kind="source", + target_id=source.id, + target_resource_version=_resource_version( + "source", + source.id, + source.updated_at, + {"enabled": source.enabled}, + ), + ) + + +async def _execute_trigger_task( + db: AsyncSession, + args: dict[str, Any], +) -> dict[str, Any]: + source = await source_service.get_source(db, str(args.get("source_id", ""))) + if source is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "数据源不存在") + if not source.enabled: + raise HTTPException(status.HTTP_400_BAD_REQUEST, "数据源已停用, 无法采集") + task = await task_service.create_task( + db, + source_id=source.id, + trigger_type="manual", + parameters={}, + priority=0, + agent_id=None, + ) + + # Preserve the existing dispatch contract: the task must be durable before + # it is handed to the executor. + await db.commit() + from backend.executor import get_executor + + try: + dispatch = await get_executor().dispatch_collection(task.id, {}) + except Exception as exc: + logger.exception( + "agent control | trigger_task dispatch failed source=%s task=%s", + source.id, + task.id, + ) + raise CommittedActionError( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=f"任务已创建但派发失败 (task_id={task.id}), 请到工作项里重试", + result={"task_id": task.id, "dispatch_error": type(exc).__name__}, + ) from exc + return {"task_id": task.id, "dispatch": dispatch} + + +async def _prepare_update_schedule( + db: AsyncSession, + args: dict[str, Any], +) -> ActionPreview: + schedule_id = str(args.get("schedule_id", "")) + schedule = await schedule_service.get_schedule(db, schedule_id) + if schedule is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, f"调度 {schedule_id} 不存在") + + normalized: dict[str, Any] = {"schedule_id": schedule_id} + changes: list[str] = [] + if args.get("cron_expression") is not None: + new_cron = str(args["cron_expression"]) + if not schedule_service.validate_cron_expression(new_cron): + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + f"非法 cron 表达式: {new_cron}", + ) + normalized["cron_expression"] = new_cron + changes.append(f"cron {schedule.cron_expression} → {new_cron}") + if args.get("enabled") is not None: + enabled = bool(args["enabled"]) + normalized["enabled"] = enabled + changes.append(f"enabled {schedule.enabled} → {enabled}") + if not changes: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + "update_schedule 未指定要改的字段 (cron_expression 或 enabled)", + ) + + return ActionPreview( + action_name="update_schedule", + args=normalized, + summary=f"修改调度「{schedule.name}」", + diff="; ".join(changes), + target_kind="schedule", + target_id=schedule.id, + target_resource_version=_resource_version( + "schedule", + schedule.id, + schedule.updated_at, + { + "cron_expression": schedule.cron_expression, + "enabled": schedule.enabled, + }, + ), + ) + + +async def _execute_update_schedule( + db: AsyncSession, + args: dict[str, Any], +) -> dict[str, Any]: + schedule = await schedule_service.get_schedule(db, str(args.get("schedule_id", ""))) + if schedule is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "调度不存在") + fields = {key: args[key] for key in ("cron_expression", "enabled") if key in args} + await schedule_service.update_schedule(db, schedule, CronScheduleUpdate(**fields)) + return {} + + +async def _prepare_update_provider( + db: AsyncSession, + args: dict[str, Any], +) -> ActionPreview: + provider_id = str(args.get("provider_id", "")) + provider = await db.get(ModelProvider, provider_id) + if provider is None: + raise HTTPException( + status.HTTP_404_NOT_FOUND, + f"模型提供商 {provider_id} 不存在", + ) + + normalized: dict[str, Any] = {"provider_id": provider_id} + changes: list[str] = [] + if args.get("default_model") is not None: + new_model = str(args["default_model"]) + normalized["default_model"] = new_model + changes.append(f"default_model {provider.default_model} → {new_model}") + if args.get("enabled") is not None: + enabled = bool(args["enabled"]) + normalized["enabled"] = enabled + state = "启用" if enabled else "停用" + changes.append(f"{state} (enabled {provider.enabled} → {enabled})") + if not changes: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + "update_provider 未指定要改的字段 (default_model 或 enabled)", + ) + + return ActionPreview( + action_name="update_provider", + args=normalized, + summary=f"配置 AI 模型提供商「{provider.name}」", + diff="; ".join(changes), + target_kind="model_provider", + target_id=provider.id, + target_resource_version=_resource_version( + "model_provider", + provider.id, + provider.updated_at, + { + "default_model": provider.default_model, + "enabled": provider.enabled, + }, + ), + ) + + +async def _execute_update_provider( + db: AsyncSession, + args: dict[str, Any], +) -> dict[str, Any]: + provider = await db.get(ModelProvider, str(args.get("provider_id", ""))) + if provider is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "模型提供商不存在") + if "default_model" in args: + provider.default_model = str(args["default_model"]) + if "enabled" in args: + provider.enabled = bool(args["enabled"]) + await db.flush() + return {} + + +def _build_registry() -> AgentControlActionRegistry: + registry = AgentControlActionRegistry() + registry.register( + RegisteredAction( + name="toggle_source", + permission=WorkspacePermission.MANAGE_CONFIGURATION, + severity=Severity.MEDIUM, + prepare=_prepare_toggle_source, + execute=_execute_toggle_source, + ) + ) + registry.register( + RegisteredAction( + name="trigger_task", + permission=WorkspacePermission.RUN_OPERATIONS_AGENTS, + severity=Severity.LOW, + prepare=_prepare_trigger_task, + execute=_execute_trigger_task, + ) + ) + registry.register( + RegisteredAction( + name="update_schedule", + permission=WorkspacePermission.MANAGE_CONFIGURATION, + severity=Severity.MEDIUM, + prepare=_prepare_update_schedule, + execute=_execute_update_schedule, + ) + ) + registry.register( + RegisteredAction( + name="update_provider", + permission=WorkspacePermission.MANAGE_CONFIGURATION, + severity=Severity.MEDIUM, + prepare=_prepare_update_provider, + execute=_execute_update_provider, + ) + ) + return registry + + +class AgentControlService: + def __init__(self, registry: AgentControlActionRegistry) -> None: + self.registry = registry + + async def resolve_workspace_id( + self, + db: AsyncSession, + identity: RequestIdentity, + requested_workspace_id: str | None, + ) -> str: + """Resolve legacy chat requests without weakening Workspace scope.""" + + if requested_workspace_id: + await get_workspace_access(db, requested_workspace_id, identity) + return requested_workspace_id + + rows = ( + await db.scalars( + select(WorkspaceMembership.workspace_id) + .join(User, User.id == WorkspaceMembership.user_id) + .join(Workspace, Workspace.id == WorkspaceMembership.workspace_id) + .where(User.subject == identity.subject) + .where(User.disabled.is_(False)) + .where(Workspace.active.is_(True)) + ) + ).all() + workspace_ids = list(dict.fromkeys(rows)) + if not workspace_ids: + raise HTTPException( + status.HTTP_403_FORBIDDEN, + "Workspace membership required", + ) + if len(workspace_ids) != 1: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + "workspace_id is required when the actor belongs to multiple workspaces", + ) + return workspace_ids[0] + + async def preview( + self, + db: AsyncSession, + action_name: str, + args: dict[str, Any], + ) -> ActionPreview: + return await self.registry.get(action_name).prepare(db, args) + + async def create_proposal( + self, + db: AsyncSession, + *, + workspace_id: str, + identity: RequestIdentity, + action_name: str, + args: dict[str, Any], + origin: str, + ) -> RecordedActionProposal: + action = self.registry.get(action_name) + access = await get_workspace_access(db, workspace_id, identity) + require_permission(access, action.permission) + preview = await action.prepare(db, args) + proposal_version = f"agent-control-proposal/v1:{uuid.uuid4()}" + now = datetime.now(UTC) + evidence = { + "schema_version": EVIDENCE_SCHEMA_VERSION, + "proposal_version": proposal_version, + "target_resource_version": preview.target_resource_version, + "policy_state_version": POLICY_STATE_VERSION, + "permission_state_version": _permission_state_version(workspace_id, access), + "diff": {"summary": preview.diff}, + "observations": [ + "Agent-originated mutation requires explicit confirmation before execution." + ], + "agent_control": { + "action": preview.action_name, + "args": preview.args, + "target": { + "kind": preview.target_kind, + "id": preview.target_id, + }, + "required_permission": action.permission.value, + "origin": origin, + }, + "confirmation": { + "required": True, + "state": "pending", + "created_at": now.isoformat(), + }, + "actor_identity": { + "subject": identity.subject, + "auth_method": identity.auth_method, + }, + } + work_item = OperationsWorkItem( + workspace_id=workspace_id, + type=WorkItemType.CHANGE_PROPOSAL, + status=WorkItemStatus.OPEN, + severity=action.severity, + priority=Priority.NORMAL, + author_actor_type="user", + author_actor_id=access.user_id, + evidence=evidence, + reason=preview.summary, + ) + db.add(work_item) + await db.flush() + return RecordedActionProposal( + work_item_id=work_item.id, + workspace_id=workspace_id, + proposal_version=proposal_version, + preview=preview, + ) + + async def execute_confirmed( + self, + db: AsyncSession, + *, + workspace_id: str, + identity: RequestIdentity, + work_item_id: str, + proposal_version: str, + confirmation_path: str, + expected_action: str | None = None, + ) -> dict[str, Any]: + access = await get_workspace_access(db, workspace_id, identity) + work_item = await db.scalar( + select(OperationsWorkItem) + .where(OperationsWorkItem.id == work_item_id) + .where(OperationsWorkItem.workspace_id == workspace_id) + .where(OperationsWorkItem.type == WorkItemType.CHANGE_PROPOSAL) + .with_for_update() + ) + if work_item is None: + raise HTTPException( + status.HTTP_404_NOT_FOUND, + "Agent Control proposal not found", + ) + if work_item.status != WorkItemStatus.OPEN: + raise HTTPException( + status.HTTP_409_CONFLICT, + "Agent Control proposal is not actionable", + ) + + evidence = dict(work_item.evidence or {}) + recorded_version = evidence.get("proposal_version") + if recorded_version != proposal_version: + raise HTTPException( + status.HTTP_409_CONFLICT, + "Agent Control proposal version changed", + ) + control = evidence.get("agent_control") + if not isinstance(control, dict): + raise HTTPException( + status.HTTP_409_CONFLICT, + "Agent Control evidence is missing", + ) + action_name = control.get("action") + args = control.get("args") + if not isinstance(action_name, str) or not isinstance(args, dict): + raise HTTPException( + status.HTTP_409_CONFLICT, + "Agent Control action evidence is invalid", + ) + if expected_action is not None and action_name != expected_action: + raise HTTPException( + status.HTTP_409_CONFLICT, + "Confirmed action does not match the recorded proposal", + ) + + action = self.registry.get(action_name) + require_permission(access, action.permission) + preview = await action.prepare(db, args) + if evidence.get("target_resource_version") != preview.target_resource_version: + raise HTTPException( + status.HTTP_409_CONFLICT, + "Agent Control target changed; create a new proposal", + ) + + now = datetime.now(UTC) + evidence["confirmation"] = { + "required": True, + "state": "confirmed", + "path": confirmation_path, + "confirmed_at": now.isoformat(), + "actor_user_id": access.user_id, + "actor_subject": identity.subject, + "actor_role": access.role.value, + } + evidence["approval_grant"] = { + "grant_type": "explicit_confirmation", + "proposal_version": proposal_version, + "granted_at": now.isoformat(), + "approver_user_ids": [access.user_id], + "confirmation_path": confirmation_path, + } + work_item.evidence = evidence + work_item.status = WorkItemStatus.IN_PROGRESS + await db.flush() + + try: + action_result = await action.execute(db, preview.args) + except CommittedActionError as exc: + failure_evidence = dict(evidence) + failure_evidence["execution"] = { + "status": "failed_after_commit", + "failed_at": datetime.now(UTC).isoformat(), + "result": exc.result, + } + work_item.evidence = failure_evidence + await db.commit() + raise HTTPException(exc.status_code, exc.detail) from exc + except Exception: + await db.rollback() + raise + + result = { + "applied": True, + "tool": action_name, + "summary": work_item.reason or preview.summary, + "work_item_id": work_item.id, + "proposal_version": proposal_version, + **action_result, + } + completed_evidence = dict(evidence) + completed_evidence["execution"] = { + "status": "applied", + "executed_at": datetime.now(UTC).isoformat(), + "result": result, + } + work_item.evidence = completed_evidence + work_item.status = WorkItemStatus.RESOLVED + await db.commit() + logger.info( + "agent control | applied action=%s proposal=%s workspace=%s actor=%s", + action_name, + work_item.id, + workspace_id, + identity.subject, + ) + return result + + +ACTION_REGISTRY = _build_registry() +agent_control_service = AgentControlService(ACTION_REGISTRY) diff --git a/tests/integration/test_chat_api.py b/tests/integration/test_chat_api.py index 510eef6..e7f1a02 100644 --- a/tests/integration/test_chat_api.py +++ b/tests/integration/test_chat_api.py @@ -6,9 +6,14 @@ import pytest from fastapi import HTTPException +from sqlalchemy import select from backend.api.v1.chat import _build_proposal, _run_read_tool +from backend.main import app +from backend.models.identity import User, Workspace, WorkspaceMembership, WorkspaceRole +from backend.models.operations_work_item import OperationsWorkItem, WorkItemStatus from backend.models.provider import ModelProvider +from backend.security.identity import RequestIdentity, get_request_identity async def _make_provider(db_session, **overrides) -> ModelProvider: @@ -26,6 +31,33 @@ async def _make_provider(db_session, **overrides) -> ModelProvider: return provider +async def _authorize_chat( + db_session, + *, + subject: str = "chat-admin", + role: WorkspaceRole = WorkspaceRole.ADMIN, +) -> tuple[RequestIdentity, User, Workspace]: + identity = RequestIdentity(subject=subject) + user = User(subject=subject) + workspace = Workspace(name=f"Chat {subject}", slug=f"chat-{subject}") + db_session.add_all([user, workspace]) + await db_session.flush() + db_session.add( + WorkspaceMembership( + workspace_id=workspace.id, + user_id=user.id, + role=role, + ) + ) + await db_session.commit() + + async def override_identity(): + return identity + + app.dependency_overrides[get_request_identity] = override_identity + return identity, user, workspace + + # ── read: list_providers ───────────────────────────────────────────────────── @pytest.mark.asyncio async def test_list_providers_read_tool_empty(db_session): @@ -44,7 +76,14 @@ async def test_list_providers_read_tool_returns_enabled_and_disabled(db_session) by_id = {p["id"]: p for p in result} assert by_id[enabled.id]["enabled"] is True assert by_id[disabled.id]["enabled"] is False - assert set(by_id[enabled.id]) == {"id", "name", "provider_type", "default_model", "base_url", "enabled"} + assert set(by_id[enabled.id]) == { + "id", + "name", + "provider_type", + "default_model", + "base_url", + "enabled", + } # ── write: update_provider proposal ────────────────────────────────────────── @@ -68,7 +107,11 @@ async def test_update_provider_proposal_default_model(db_session): async def test_update_provider_proposal_enabled_toggle(db_session): provider = await _make_provider(db_session, enabled=True) - proposal = await _build_proposal(db_session, "update_provider", {"provider_id": provider.id, "enabled": False}) + proposal = await _build_proposal( + db_session, + "update_provider", + {"provider_id": provider.id, "enabled": False}, + ) assert proposal.args == {"provider_id": provider.id, "enabled": False} assert "启用" in proposal.diff or "停用" in proposal.diff @@ -77,7 +120,11 @@ async def test_update_provider_proposal_enabled_toggle(db_session): @pytest.mark.asyncio async def test_update_provider_proposal_not_found(db_session): with pytest.raises(HTTPException) as exc_info: - await _build_proposal(db_session, "update_provider", {"provider_id": "nonexistent-id", "enabled": True}) + await _build_proposal( + db_session, + "update_provider", + {"provider_id": "nonexistent-id", "enabled": True}, + ) assert exc_info.value.status_code == 404 @@ -93,6 +140,7 @@ async def test_update_provider_proposal_no_fields(db_session): @pytest.mark.asyncio async def test_confirm_update_provider(client, db_session): provider = await _make_provider(db_session, default_model="gpt-4o-mini", enabled=True) + _, user, workspace = await _authorize_chat(db_session) response = await client.post( "/api/v1/chat/confirm", @@ -114,10 +162,19 @@ async def test_confirm_update_provider(client, db_session): await db_session.refresh(provider) assert provider.default_model == "qwen3:4b" assert provider.enabled is False + work_item = await db_session.get(OperationsWorkItem, body["work_item_id"]) + assert work_item is not None + assert work_item.workspace_id == workspace.id + assert work_item.author_actor_type == "user" + assert work_item.author_actor_id == user.id + assert work_item.status == WorkItemStatus.RESOLVED + assert work_item.evidence["proposal_version"] == body["proposal_version"] + assert work_item.evidence["approval_grant"]["proposal_version"] == body["proposal_version"] @pytest.mark.asyncio -async def test_confirm_update_provider_not_found(client): +async def test_confirm_update_provider_not_found(client, db_session): + await _authorize_chat(db_session) response = await client.post( "/api/v1/chat/confirm", json={ @@ -148,6 +205,7 @@ async def test_confirm_trigger_task_reports_dispatch_failure(client, db_session, db_session.add(source) await db_session.commit() await db_session.refresh(source) + await _authorize_chat(db_session) class _BoomExecutor: async def dispatch_collection(self, task_id: str, parameters: dict) -> dict: @@ -169,3 +227,77 @@ async def dispatch_collection(self, task_id: str, parameters: dict) -> dict: assert response.status_code == 502 assert "派发失败" in response.json()["detail"] + work_item = await db_session.scalar(select(OperationsWorkItem)) + assert work_item is not None + assert work_item.status == WorkItemStatus.IN_PROGRESS + assert work_item.evidence["execution"]["status"] == "failed_after_commit" + + +@pytest.mark.asyncio +async def test_viewer_confirmation_is_denied_without_mutation(client, db_session): + provider = await _make_provider(db_session, default_model="gpt-4o-mini", enabled=True) + await _authorize_chat( + db_session, + subject="chat-viewer", + role=WorkspaceRole.VIEWER, + ) + + response = await client.post( + "/api/v1/chat/confirm", + json={ + "proposal": { + "tool": "update_provider", + "args": {"provider_id": provider.id, "enabled": False}, + "summary": "配置 AI 模型提供商", + "diff": "enabled true -> false", + } + }, + ) + + assert response.status_code == 403 + await db_session.refresh(provider) + assert provider.enabled is True + work_items = (await db_session.scalars(select(OperationsWorkItem))).all() + assert work_items == [] + + +@pytest.mark.asyncio +async def test_cross_workspace_confirmation_cannot_apply_recorded_proposal( + client, + db_session, +): + provider = await _make_provider(db_session, default_model="gpt-4o-mini", enabled=True) + identity_a, user_a, workspace_a = await _authorize_chat( + db_session, + subject="workspace-a-admin", + ) + proposal = await _build_proposal( + db_session, + "update_provider", + {"provider_id": provider.id, "enabled": False}, + identity=identity_a, + workspace_id=workspace_a.id, + ) + await db_session.commit() + + _, _, workspace_b = await _authorize_chat( + db_session, + subject="workspace-b-admin", + ) + response = await client.post( + "/api/v1/chat/confirm", + json={"proposal": proposal.model_dump()}, + ) + + assert response.status_code == 403 + await db_session.refresh(provider) + assert provider.enabled is True + work_item = await db_session.get(OperationsWorkItem, proposal.work_item_id) + assert work_item is not None + assert work_item.workspace_id == workspace_a.id + assert work_item.workspace_id != workspace_b.id + assert work_item.author_actor_id == user_a.id + assert work_item.status == WorkItemStatus.OPEN + assert work_item.evidence["schema_version"] == "agent-control-evidence/v1" + assert work_item.evidence["proposal_version"] == proposal.proposal_version + assert work_item.evidence["confirmation"]["state"] == "pending" diff --git a/tests/unit/control/test_agent_control.py b/tests/unit/control/test_agent_control.py new file mode 100644 index 0000000..09d574e --- /dev/null +++ b/tests/unit/control/test_agent_control.py @@ -0,0 +1,23 @@ +from backend.control.agent_control import ACTION_REGISTRY +from backend.security.workspace_rbac import WorkspacePermission + + +def test_agent_control_registry_is_the_complete_chat_write_surface(): + assert ACTION_REGISTRY.action_names == { + "toggle_source", + "trigger_task", + "update_schedule", + "update_provider", + } + + +def test_registry_assigns_write_permissions_per_action(): + assert ( + ACTION_REGISTRY.get("trigger_task").permission + == WorkspacePermission.RUN_OPERATIONS_AGENTS + ) + for action_name in ("toggle_source", "update_schedule", "update_provider"): + assert ( + ACTION_REGISTRY.get(action_name).permission + == WorkspacePermission.MANAGE_CONFIGURATION + ) From 8960d092356ef038645b84d941837790841320bc Mon Sep 17 00:00:00 2001 From: 2233admin <2233admin@users.noreply.github.com> Date: Sun, 26 Jul 2026 20:20:55 +0800 Subject: [PATCH 3/3] test: integrate source and agent control evidence --- backend/api/v1/project_source_bindings.py | 19 ++- backend/api/v1/workspace_sources.py | 13 ++- ...4h5i6j7k8_add_source_and_source_binding.py | 2 + backend/models/source_binding.py | 2 +- backend/schemas/source_binding.py | 5 +- docs/backend-architecture-consolidated.md | 10 +- docs/backend-capability-exposure-matrix.yaml | 110 +++++++++++++++++- .../2026-07-26-source-agent-ere-report.md | 106 +++++++++++++++++ tests/integration/test_chat_api.py | 30 +++++ ...st_legacy_native_intelligence_migration.py | 2 +- .../test_legacy_plugin_migration.py | 2 +- tests/unit/api/test_source_binding.py | 34 +++++- 12 files changed, 318 insertions(+), 17 deletions(-) create mode 100644 docs/verification/2026-07-26-source-agent-ere-report.md diff --git a/backend/api/v1/project_source_bindings.py b/backend/api/v1/project_source_bindings.py index 61124b4..28c3f8a 100644 --- a/backend/api/v1/project_source_bindings.py +++ b/backend/api/v1/project_source_bindings.py @@ -3,7 +3,12 @@ from sqlalchemy.ext.asyncio import AsyncSession from backend.database import get_db -from backend.models.source_binding import Source, SourceBinding, SourceBindingRevision, SourceRevision +from backend.models.source_binding import ( + Source, + SourceBinding, + SourceBindingRevision, + SourceRevision, +) from backend.models.workflow import Project from backend.schemas.common import ApiResponse from backend.schemas.source_binding import ( @@ -14,7 +19,11 @@ SourceBindingUpdate, ) from backend.security.identity import RequestIdentity, get_request_identity -from backend.security.workspace_rbac import WorkspacePermission, get_workspace_access, require_permission +from backend.security.workspace_rbac import ( + WorkspacePermission, + get_workspace_access, + require_permission, +) router = APIRouter( prefix="/workspaces/{workspace_id}/projects/{project_id}/source-bindings", @@ -43,7 +52,11 @@ async def _get_source_in_workspace(db: AsyncSession, workspace_id: str, source_i return source -async def _get_source_revision(db: AsyncSession, source_id: str, revision_number: int) -> SourceRevision: +async def _get_source_revision( + db: AsyncSession, + source_id: str, + revision_number: int, +) -> SourceRevision: revision = await db.scalar( select(SourceRevision).where( SourceRevision.source_id == source_id, diff --git a/backend/api/v1/workspace_sources.py b/backend/api/v1/workspace_sources.py index fb6ad03..45087fa 100644 --- a/backend/api/v1/workspace_sources.py +++ b/backend/api/v1/workspace_sources.py @@ -13,12 +13,21 @@ SourceUpdate, ) from backend.security.identity import RequestIdentity, get_request_identity -from backend.security.workspace_rbac import WorkspacePermission, get_workspace_access, require_permission +from backend.security.workspace_rbac import ( + WorkspacePermission, + get_workspace_access, + require_permission, +) router = APIRouter(prefix="/workspaces/{workspace_id}/sources", tags=["sources"]) -async def _get_source(db: AsyncSession, workspace_id: str, source_id: str, for_update: bool = False) -> Source: +async def _get_source( + db: AsyncSession, + workspace_id: str, + source_id: str, + for_update: bool = False, +) -> Source: query = select(Source).where(Source.workspace_id == workspace_id, Source.id == source_id) if for_update: query = query.with_for_update() diff --git a/backend/migrations/versions/f3g4h5i6j7k8_add_source_and_source_binding.py b/backend/migrations/versions/f3g4h5i6j7k8_add_source_and_source_binding.py index a6e264d..1aa1832 100644 --- a/backend/migrations/versions/f3g4h5i6j7k8_add_source_and_source_binding.py +++ b/backend/migrations/versions/f3g4h5i6j7k8_add_source_and_source_binding.py @@ -117,3 +117,5 @@ def downgrade() -> None: op.drop_table("source_bindings") op.drop_table("source_revisions") op.drop_table("sources") + sa.Enum(name="source_binding_lifecycle_status").drop(op.get_bind(), checkfirst=True) + sa.Enum(name="source_lifecycle_status").drop(op.get_bind(), checkfirst=True) diff --git a/backend/models/source_binding.py b/backend/models/source_binding.py index 1926f15..32ea930 100644 --- a/backend/models/source_binding.py +++ b/backend/models/source_binding.py @@ -15,7 +15,7 @@ from enum import StrEnum -from sqlalchemy import Enum, ForeignKey, Integer, JSON, String, Text, UniqueConstraint +from sqlalchemy import JSON, Enum, ForeignKey, Integer, String, Text, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from backend.models.base import TimestampMixin diff --git a/backend/schemas/source_binding.py b/backend/schemas/source_binding.py index 2f2a6b8..ebbc938 100644 --- a/backend/schemas/source_binding.py +++ b/backend/schemas/source_binding.py @@ -2,6 +2,7 @@ from pydantic import BaseModel, ConfigDict, Field +from backend.models.source_binding import SourceLifecycleStatus from backend.schemas.common import UTCModel @@ -20,7 +21,7 @@ class SourceUpdate(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=255) description: str | None = Field(default=None, max_length=4000) - status: str | None = None + status: SourceLifecycleStatus | None = None class SourceRead(UTCModel): @@ -71,7 +72,7 @@ class SourceBindingUpdate(BaseModel): model_config = ConfigDict(extra="forbid") name: str | None = Field(default=None, min_length=1, max_length=255) - status: str | None = None + status: SourceLifecycleStatus | None = None class SourceBindingRead(UTCModel): diff --git a/docs/backend-architecture-consolidated.md b/docs/backend-architecture-consolidated.md index 00a24e7..31b593f 100644 --- a/docs/backend-architecture-consolidated.md +++ b/docs/backend-architecture-consolidated.md @@ -1,15 +1,15 @@ # opencli-admin 后端架构声明（分支归一后） -> 基线：main `b65708b`（2026-07-26）。三条产品线——安全加固线（`fix/sec-correctness-hardening`）、运营治理线（`notification-ack-cleanup-work`）、unified 产品线（`codex/unified-product-3002`）——已全部并入 main；迁移头归一为 `1901f6da7138`；八处跨线接缝已缝合（见 [#43](https://github.com/2233admin/opencli-admin/issues/43)）。 +> 基线：main `b65708b`（2026-07-26）。三条产品线——安全加固线（`fix/sec-correctness-hardening`）、运营治理线（`notification-ack-cleanup-work`）、unified 产品线（`codex/unified-product-3002`）——已全部并入 main；迁移头归一后由 Source/Binding V1 顺延为 `f3g4h5i6j7k8`；八处跨线接缝已缝合（见 [#43](https://github.com/2233admin/opencli-admin/issues/43)）。 > > 本文是给协作 Agent / 新会话的**架构声明**：动 `backend/` 前先读这份。`docs/ARCHITECTURE.md` 为合并前的旧全量文档, 与本文冲突处以本文为准。 | 指标 | 值 | |---|---| | v1 API 路由模块 | 34 | -| OpenAPI operations（治理台账对账后） | 194（台账：`docs/backend-capability-exposure-matrix.yaml`） | +| OpenAPI operations（治理台账对账后） | 205（台账：`docs/backend-capability-exposure-matrix.yaml`） | | 测试 | unit+compat+integration ≈ 2374 通过, 0 xfail | -| Alembic | 单头 `1901f6da7138` | +| Alembic | 单头 `f3g4h5i6j7k8` | ## 1. 分层总图 @@ -87,12 +87,12 @@ flowchart LR - **Workspace RBAC**：`workspaces.py`（成员/角色）+ `identity.py`（User / Team / ServiceIdentity）。**已知 TODO**：RBAC 版工作区列表被 Studio 全量版 `GET /workspaces` 遮蔽——dev 模式（无 OIDC）依赖全量版；接 OIDC 前必须切换。 - **Operations 控制面**：`operations_inbox`（工作项）、`operations_agents`（版本化 Agent 身份 + 权限画像）、`automations`、`consumer_grants`。 -- **暴露台账**：`docs/backend-capability-exposure-matrix.yaml` 锁 194 个 operations, `test_capability_exposure_matrix` 盯漂移。合并新增的 workspace-governance 条目标注 "governance review pending", 待复核。 +- **暴露台账**：`docs/backend-capability-exposure-matrix.yaml` 锁 205 个 operations, `test_capability_exposure_matrix` 盯漂移。合并新增的 workspace-governance 条目标注 "governance review pending", 待复核。 - **通知**：`NotificationSendResult` 契约（含 ack 字段）+ 三阶段派发（计划 / 发送 / 短会话回写）；webhook 出站走 SSRF guard, 网络异常收敛为 `WorkflowWebhookDeliveryError`。 ## 5. 存储与迁移 -- SQLite（dev, `aiosqlite`）/ PostgreSQL（compose）。迁移单头 `1901f6da7138`——运营线（workspace RBAC / operations / versioning）与产品线（plugin / feed / intelligence session）双链在此汇合。 +- SQLite（dev, `aiosqlite`）/ PostgreSQL（compose）。迁移单头 `f3g4h5i6j7k8`——`1901f6da7138` 汇合运营线与产品线后，Source/Binding V1 单链顺延。 - **Legacy 升级路径已验证**：旧 plugin-hub 链 rejoin 拓扑下, 版本化迁移对缺失的 `workflow_runs` 表容忍跳过（online inspector guard；offline SQL 渲染不受影响）。模式参照 spine 迁移的缺表 early-return 惯例。 - **去重是存储层保证**：`store_records` 按 `(source, content_hash)` 拦截。实测：A 股采集两轮 125→126, 仅真实新增入库。 diff --git a/docs/backend-capability-exposure-matrix.yaml b/docs/backend-capability-exposure-matrix.yaml index a1f35bf..524723a 100644 --- a/docs/backend-capability-exposure-matrix.yaml +++ b/docs/backend-capability-exposure-matrix.yaml @@ -1,6 +1,6 @@ version: 1 source: backend.main.app.openapi -openapi_operation_count: 193 +openapi_operation_count: 205 allowed_dispositions: - operator_ui - studio_binding @@ -1975,6 +1975,114 @@ operations: decision: Added by branch-consolidation merge; governance review pending. target_epic: Epic 8 capability_id: operator.workspace-governance +- method: GET + path: /api/v1/workspaces/{workspace_id}/sources + operation_id: list_sources_api_v1_workspaces__workspace_id__sources_get + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Bind workspace Source management to Studio without redefining legacy DataSource. + target_epic: Epic 8 + capability_id: studio.sources +- method: POST + path: /api/v1/workspaces/{workspace_id}/sources + operation_id: create_source_api_v1_workspaces__workspace_id__sources_post + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Bind workspace Source management to Studio without redefining legacy DataSource. + target_epic: Epic 8 + capability_id: studio.sources +- method: GET + path: /api/v1/workspaces/{workspace_id}/sources/{source_id} + operation_id: get_source_api_v1_workspaces__workspace_id__sources__source_id__get + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Bind workspace Source management to Studio without redefining legacy DataSource. + target_epic: Epic 8 + capability_id: studio.sources +- method: PATCH + path: /api/v1/workspaces/{workspace_id}/sources/{source_id} + operation_id: update_source_api_v1_workspaces__workspace_id__sources__source_id__patch + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Bind workspace Source management to Studio without redefining legacy DataSource. + target_epic: Epic 8 + capability_id: studio.sources +- method: GET + path: /api/v1/workspaces/{workspace_id}/sources/{source_id}/revisions + operation_id: list_source_revisions_api_v1_workspaces__workspace_id__sources__source_id__revisions_get + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Expose immutable Source revision history through the Studio Source surface. + target_epic: Epic 8 + capability_id: studio.sources +- method: POST + path: /api/v1/workspaces/{workspace_id}/sources/{source_id}/revisions + operation_id: create_source_revision_api_v1_workspaces__workspace_id__sources__source_id__revisions_post + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Create immutable Source revisions through the Studio Source surface. + target_epic: Epic 8 + capability_id: studio.sources +- method: GET + path: /api/v1/workspaces/{workspace_id}/projects/{project_id}/source-bindings + operation_id: list_source_bindings_api_v1_workspaces__workspace_id__projects__project_id__source_bindings_get + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Bind project SourceBinding management to Studio with explicit revision pins. + target_epic: Epic 8 + capability_id: studio.sources +- method: POST + path: /api/v1/workspaces/{workspace_id}/projects/{project_id}/source-bindings + operation_id: create_source_binding_api_v1_workspaces__workspace_id__projects__project_id__source_bindings_post + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Bind project SourceBinding management to Studio with explicit revision pins. + target_epic: Epic 8 + capability_id: studio.sources +- method: GET + path: /api/v1/workspaces/{workspace_id}/projects/{project_id}/source-bindings/{binding_id} + operation_id: get_source_binding_api_v1_workspaces__workspace_id__projects__project_id__source_bindings__binding_id__get + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Bind project SourceBinding management to Studio with explicit revision pins. + target_epic: Epic 8 + capability_id: studio.sources +- method: PATCH + path: /api/v1/workspaces/{workspace_id}/projects/{project_id}/source-bindings/{binding_id} + operation_id: update_source_binding_api_v1_workspaces__workspace_id__projects__project_id__source_bindings__binding_id__patch + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Bind project SourceBinding management to Studio with explicit revision pins. + target_epic: Epic 8 + capability_id: studio.sources +- method: GET + path: /api/v1/workspaces/{workspace_id}/projects/{project_id}/source-bindings/{binding_id}/revisions + operation_id: list_source_binding_revisions_api_v1_workspaces__workspace_id__projects__project_id__source_bindings__binding_id__revisions_get + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Expose immutable SourceBinding revision history through Studio. + target_epic: Epic 8 + capability_id: studio.sources +- method: POST + path: /api/v1/workspaces/{workspace_id}/projects/{project_id}/source-bindings/{binding_id}/revisions + operation_id: create_source_binding_revision_api_v1_workspaces__workspace_id__projects__project_id__source_bindings__binding_id__revisions_post + disposition: studio_binding + frontend_route: /studio + wrapper: null + decision: Re-pin a SourceBinding only by creating an immutable binding revision. + target_epic: Epic 8 + capability_id: studio.sources unreferenced_wrappers: - wrapper: getWorkspaceSettings operation_id: null diff --git a/docs/verification/2026-07-26-source-agent-ere-report.md b/docs/verification/2026-07-26-source-agent-ere-report.md new file mode 100644 index 0000000..54bf5f1 --- /dev/null +++ b/docs/verification/2026-07-26-source-agent-ere-report.md @@ -0,0 +1,106 @@ +# Source/Binding + Global Agent Engineering Readiness & Evidence (ERE) + +Date: 2026-07-26 + +Integration baseline: `origin/main` at `8da964cf533297407786d353fe08dc6e6118f80c` + +Integration branch: `codex/source-agent-integration` + +## Outcome + +The two backend seams identified after the architecture consolidation are now +implemented together: + +1. Workspace-owned `Source` and immutable `SourceRevision`, with Project-owned + `SourceBinding` and immutable `SourceBindingRevision` that pins an exact + Source revision. +2. One Agent Control registry and execution service for all four existing chat + write actions, with Workspace/RBAC rechecks, explicit confirmation, + versioned `OperationsWorkItem` evidence, target-version conflict detection, + and legacy Proposal compatibility. + +Legacy `DataSource` and `FeedProvider` behavior remains unchanged. No new +frontend Source editor was introduced; the 12 new operations are recorded under +the existing `studio.sources` capability projection. + +## ORCA / 5090 execution evidence + +| Lane | ORCA task | Agent | Branch | Worker commit | Worker evidence | +| --- | --- | --- | --- | --- | --- | +| Source/Binding V1 | `task_cc2686e426db` | Claude | `source-binding-v1` | `c4aa92e6b7d8321f8eab65a93d46c89c0cf62f93` | 8/8 focused API and migration-head tests | +| Global Agent Control V1 | `task_cd679bec801a` | Codex | `global-agent-control-v1` | `2ab91e1` | 40 focused tests plus targeted mypy and ruff | + +Both lanes ran in isolated clean worktrees on the saved ORCA environment +`5090` (`ws://100.80.105.128:6768`) from the same `origin/main` baseline. The +dirty 5090 `main` worktree and the dirty local root worktree were not modified. + +The first 5090 Codex terminal hit the host's known Windows +`CreateProcessAsUserW failed: 5` sandbox error. ORCA reset the dispatch and +restarted Codex in the same isolated worktree with its supported +no-sandbox execution mode; the recovered task then completed normally. + +## Integrated changes + +- Added four Source/Binding models, schemas, scoped routers, model/router + registration, and one Alembic migration. +- Added an Agent Control action registry/service and routed chat confirmations + through it instead of direct service mutations. +- Added or extended regression coverage for: + - Workspace ownership and cross-Workspace rejection. + - Immutable Source revisions. + - Exact SourceBinding revision pins and explicit re-pinning. + - Viewer denial and cross-Workspace confirmation denial. + - Proposal attribution/versioning and dispatch-failure evidence. + - Stale target-version rejection without mutation. + - Migration-head and legacy-database upgrade compatibility. +- Updated the capability exposure ledger from 193 to 205 OpenAPI operations and + verified the existing generated catalog remains current. +- Updated the consolidated architecture document to the new single Alembic head + `f3g4h5i6j7k8`. + +## Verification evidence + +| Check | Result | +| --- | --- | +| Related Source/Agent/capability tests | 50 passed | +| Added stale proposal conflict test | 1 passed | +| Legacy migration and single-head tests | 6 passed | +| Ruff on new Source/Agent files and tests | Passed | +| Mypy on six affected backend modules | Passed | +| Capability catalog generator `--check` | Passed | +| Fresh SQLite migration upgrade → downgrade → upgrade | Passed; current head `f3g4h5i6j7k8` | +| Full non-live/non-Postgres suite, excluding DNS-affected RSS files | 2489 passed, 3 skipped, 48 deselected | +| Existing 8030 frontend process | HTTP 200; running from `D:\projects\opencli-admin-wt-unified-3002\frontend` | + +The first complete non-live run reported 2491 passed and 12 failed. Two failures +were stale assertions for the previous migration head and were fixed. The other +10 failures came from this machine resolving `example.com` to the reserved +benchmark address `198.18.1.6`, which the production SSRF guard correctly +rejects. The same RSS success test fails identically on the unmodified main +worktree, proving this is a local test-environment dependency rather than an +integration regression. The SSRF protection was not weakened. + +The ECC pre-push hook also invokes the first `pytest` on `PATH`, currently a +Python 3.10 x-cmd installation, even though this repository requires Python +3.13. That hook fails while importing `enum.StrEnum`; the same tests pass under +the repository's locked uv environment. This is a development-system routing +bug, not an application failure, and should be fixed in the shared hook by +running the repository-declared test command. + +## Readiness decision + +Ready for review and PR. The integrated backend behavior, migration chain, +capability ledger, and affected test surfaces are verified. + +Residual work is deliberately bounded: + +- Legacy mutable targets (`DataSource`, schedule, provider) are still globally + modeled, so Agent Control V1 enforces Workspace isolation at proposal and + confirmation boundaries rather than through target-row ownership. +- MCP/SDK adapters can now reuse Agent Control but are not added in this slice. +- A Source/Binding frontend editor remains a later Studio task; this slice only + establishes the backend contract and capability projection. +- RSS channel tests should eventually stub DNS/SSRF resolution so workstation + DNS policy cannot decide their result. +- The shared ECC pre-push Python runner should use `uv run --extra dev pytest` + instead of an unqualified global `pytest`. diff --git a/tests/integration/test_chat_api.py b/tests/integration/test_chat_api.py index e7f1a02..cb90b12 100644 --- a/tests/integration/test_chat_api.py +++ b/tests/integration/test_chat_api.py @@ -301,3 +301,33 @@ async def test_cross_workspace_confirmation_cannot_apply_recorded_proposal( assert work_item.evidence["schema_version"] == "agent-control-evidence/v1" assert work_item.evidence["proposal_version"] == proposal.proposal_version assert work_item.evidence["confirmation"]["state"] == "pending" + + +@pytest.mark.asyncio +async def test_confirmation_rejects_stale_target_version(client, db_session): + provider = await _make_provider(db_session, default_model="gpt-4o-mini", enabled=True) + identity, _, workspace = await _authorize_chat(db_session, subject="stale-proposal-admin") + proposal = await _build_proposal( + db_session, + "update_provider", + {"provider_id": provider.id, "enabled": False}, + identity=identity, + workspace_id=workspace.id, + ) + await db_session.commit() + + provider.default_model = "changed-after-proposal" + await db_session.commit() + + response = await client.post( + "/api/v1/chat/confirm", + json={"proposal": proposal.model_dump()}, + ) + + assert response.status_code == 409 + await db_session.refresh(provider) + assert provider.enabled is True + assert provider.default_model == "changed-after-proposal" + work_item = await db_session.get(OperationsWorkItem, proposal.work_item_id) + assert work_item is not None + assert work_item.status == WorkItemStatus.OPEN diff --git a/tests/integration/test_legacy_native_intelligence_migration.py b/tests/integration/test_legacy_native_intelligence_migration.py index ff6eae9..8ca9732 100644 --- a/tests/integration/test_legacy_native_intelligence_migration.py +++ b/tests/integration/test_legacy_native_intelligence_migration.py @@ -65,7 +65,7 @@ def test_legacy_plugin_head_rejoins_native_intelligence_head(tmp_path: Path) -> ) } - assert revision == ("1901f6da7138",) + assert revision == ("f3g4h5i6j7k8",) assert marker == ("workspace-1", "native-intelligence-workspace") assert "intelligence_sessions" in tables assert "intelligence_artifacts" in tables diff --git a/tests/integration/test_legacy_plugin_migration.py b/tests/integration/test_legacy_plugin_migration.py index 28f6e13..ee74976 100644 --- a/tests/integration/test_legacy_plugin_migration.py +++ b/tests/integration/test_legacy_plugin_migration.py @@ -87,7 +87,7 @@ def test_legacy_plugin_database_rejoins_current_migration_head(tmp_path: Path) - finally: connection.close() - assert revision == ("1901f6da7138",) + assert revision == ("f3g4h5i6j7k8",) assert "version" in cursor_columns assert "identity_key" in record_columns assert "ix_collected_records_source_identity" in record_indexes diff --git a/tests/unit/api/test_source_binding.py b/tests/unit/api/test_source_binding.py index 32fc641..9969291 100644 --- a/tests/unit/api/test_source_binding.py +++ b/tests/unit/api/test_source_binding.py @@ -25,7 +25,12 @@ async def override_identity(): return AsyncClient(transport=ASGITransport(app=app), base_url="http://test") -async def _seed_workspace_admin(db_session, user, workspace_name: str, workspace_slug: str) -> Workspace: +async def _seed_workspace_admin( + db_session, + user, + workspace_name: str, + workspace_slug: str, +) -> Workspace: workspace = Workspace(name=workspace_name, slug=workspace_slug) db_session.add(workspace) await db_session.flush() @@ -210,3 +215,30 @@ async def test_project_cannot_bind_source_from_another_workspace(db_session): }, ) assert rejected.status_code == 404 + + +async def test_source_status_rejects_unknown_lifecycle_value(db_session): + user = User(subject="status-owner") + db_session.add(user) + await db_session.flush() + workspace = await _seed_workspace_admin(db_session, user, "Workspace", "workspace-status") + + async with _build_client(db_session, user) as client: + source = ( + await client.post( + f"/workspaces/{workspace.id}/sources", + json={ + "name": "Source", + "slug": "source", + "adapter_type": "rss", + "adapter_config": {}, + }, + ) + ).json()["data"] + + rejected = await client.patch( + f"/workspaces/{workspace.id}/sources/{source['id']}", + json={"status": "unknown"}, + ) + + assert rejected.status_code == 422