feat(orchestrator): add worker delegation and shared concurrency

.env.example

@@ -32,6 +32,15 @@ WORKER__HEARTBEAT_SECONDS=20
 WORKER__MAX_ATTEMPTS=3
 WORKER__SHUTDOWN_GRACE_SECONDS=60
 
+# Shared orchestrated-Run slots; all API processes must use the same database and values.
+ORCHESTRATOR_CONCURRENCY__GLOBAL_WORKERS=16
+ORCHESTRATOR_CONCURRENCY__REALM_WORKERS=16
+ORCHESTRATOR_CONCURRENCY__TARGET_WORKERS=1
+ORCHESTRATOR_CONCURRENCY__MODEL_WORKERS=4
+ORCHESTRATOR_CONCURRENCY__LEASE_SECONDS=120
+ORCHESTRATOR_CONCURRENCY__HEARTBEAT_SECONDS=20
+ORCHESTRATOR_CONCURRENCY__POLL_SECONDS=0.25
+
 # Provider Instance references may use env:NAME locally. Production must use file: or vault:.
 SECRETS__BACKEND=environment
 SECRETS__ALLOW_ENVIRONMENT_REFERENCES=true

README.md

@@ -4,8 +4,8 @@
 
 # Attacker
 
-主 Agent 委派多个独立自适应测试 Agent，并汇总其 Evidence：见
-[Subagent 测试模式](docs/subagents.md)。
+Orchestrator 模型可自主拆分测试 Case，委派多个独立 worker，并基于 Evidence
+生成总结：见 [Orchestrator 与 worker 测试](docs/subagents.md)。
 
 ### Evidence-driven security evaluation for AI Agents
 
@@ -214,6 +214,9 @@ Copy-Item .env.example .env
 | 配置项 | 默认值 | 用途 |
 |---|---|---|
 | `DATABASE__URL` | `sqlite+aiosqlite:///data/attacker.sqlite3` | 业务与审计事实存储 |
+| `ORCHESTRATOR_CONCURRENCY__GLOBAL_WORKERS` | `16` | 跨进程的主任务 worker 总并发额度 |
+| `ORCHESTRATOR_CONCURRENCY__TARGET_WORKERS` | `1` | 同一 Target 的跨进程 worker 并发额度 |
+| `ORCHESTRATOR_CONCURRENCY__MODEL_WORKERS` | `4` | 同一模型接口的跨进程并发额度 |
 | `SECURITY__API_KEY` | 空 | 设置后使用 `X-API-Key` 保护业务接口 |
 | `EQUIPMENT__ROOT` | `equipment` | 本地可写装备扩展目录；内置装备从安装包只读加载 |
 | `EQUIPMENT__ALLOW_UNTRUSTED` | `false` | 是否允许不受信任装备；默认关闭 |

alembic/versions/20260923_0012_concurrency_quotas.py

@@ -0,0 +1,37 @@
+"""Add shared, expiring concurrency slots for orchestrated Runs."""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "20260923_0012"
+down_revision = "20260912_0011"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "concurrency_quotas",
+        sa.Column("resource_key", sa.String(80), primary_key=True),
+        sa.Column("capacity", sa.Integer(), nullable=False),
+    )
+    op.create_table(
+        "concurrency_leases",
+        sa.Column(
+            "resource_key",
+            sa.String(80),
+            sa.ForeignKey("concurrency_quotas.resource_key", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        sa.Column("slot", sa.Integer(), primary_key=True),
+        sa.Column("owner_token", sa.String(64), nullable=False),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_concurrency_leases_owner", "concurrency_leases", ["owner_token"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_concurrency_leases_owner", table_name="concurrency_leases")
+    op.drop_table("concurrency_leases")
+    op.drop_table("concurrency_quotas")

app/models.py

@@ -39,6 +39,29 @@ class SubagentRunRecord(Base):
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utc_now)
 
 
+class ConcurrencyQuotaRecord(Base):
+    """Deployment-wide slot count; a changed configuration must be reconciled explicitly."""
+
+    __tablename__ = "concurrency_quotas"
+
+    resource_key: Mapped[str] = mapped_column(String(80), primary_key=True)
+    capacity: Mapped[int] = mapped_column(Integer, nullable=False)
+
+
+class ConcurrencyLeaseRecord(Base):
+    """One owner-bound, expiring claim on a shared concurrency slot."""
+
+    __tablename__ = "concurrency_leases"
+    __table_args__ = (Index("ix_concurrency_leases_owner", "owner_token"),)
+
+    resource_key: Mapped[str] = mapped_column(
+        ForeignKey("concurrency_quotas.resource_key", ondelete="CASCADE"), primary_key=True
+    )
+    slot: Mapped[int] = mapped_column(Integer, primary_key=True)
+    owner_token: Mapped[str] = mapped_column(String(64), nullable=False)
+    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+
 # Run 输入快照与顶层状态：回答“这次评测在什么条件下执行”。
 class TargetRecord(Base):
     __tablename__ = "targets"

app/repositories/concurrency_repository.py

@@ -0,0 +1,86 @@
+"""Atomic shared concurrency slots for SQLite and PostgreSQL deployments."""
+
+from datetime import UTC, datetime, timedelta
+from typing import Any, cast
+
+from sqlalchemy import delete, func, select, update
+from sqlalchemy.dialects.postgresql import insert as postgres_insert
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
+from sqlalchemy.engine import CursorResult
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.models import ConcurrencyLeaseRecord, ConcurrencyQuotaRecord
+
+
+class ConcurrencyRepository:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self.session_factory = session_factory
+
+    @staticmethod
+    def _insert(session: AsyncSession, model: type[Any]) -> Any:
+        dialect = session.bind.dialect.name if session.bind is not None else ""
+        if dialect == "postgresql":
+            return postgres_insert(model)
+        if dialect == "sqlite":
+            return sqlite_insert(model)
+        raise ValueError("shared concurrency requires PostgreSQL or SQLite")
+
+    @staticmethod
+    async def _database_now(session: AsyncSession) -> datetime:
+        value = await session.scalar(select(func.current_timestamp()))
+        if not isinstance(value, datetime):
+            raise TypeError("database did not return its current time")
+        return value if value.tzinfo is not None else value.replace(tzinfo=UTC)
+
+    async def try_claim(
+        self, resource_key: str, capacity: int, owner_token: str, lease_seconds: int
+    ) -> bool:
+        async with self.session_factory() as session, session.begin():
+            quota_insert = self._insert(session, ConcurrencyQuotaRecord).values(
+                resource_key=resource_key, capacity=capacity
+            )
+            await session.execute(
+                quota_insert.on_conflict_do_nothing(index_elements=["resource_key"])
+            )
+            quota = await session.get(ConcurrencyQuotaRecord, resource_key)
+            if quota is None or quota.capacity != capacity:
+                raise ValueError("shared concurrency capacity differs between processes")
+            expires_at = await self._database_now(session) + timedelta(seconds=lease_seconds)
+            for slot in range(capacity):
+                claim = self._insert(session, ConcurrencyLeaseRecord).values(
+                    resource_key=resource_key,
+                    slot=slot,
+                    owner_token=owner_token,
+                    expires_at=expires_at,
+                )
+                claimed = await session.scalar(
+                    claim.on_conflict_do_update(
+                        index_elements=["resource_key", "slot"],
+                        set_={"owner_token": owner_token, "expires_at": expires_at},
+                        where=ConcurrencyLeaseRecord.expires_at <= func.current_timestamp(),
+                    ).returning(ConcurrencyLeaseRecord.slot)
+                )
+                if claimed is not None:
+                    return True
+            return False
+
+    async def renew(self, owner_token: str, expected_slots: int, lease_seconds: int) -> bool:
+        async with self.session_factory() as session, session.begin():
+            expires_at = await self._database_now(session) + timedelta(seconds=lease_seconds)
+            result = await session.execute(
+                update(ConcurrencyLeaseRecord)
+                .where(
+                    ConcurrencyLeaseRecord.owner_token == owner_token,
+                    ConcurrencyLeaseRecord.expires_at > func.current_timestamp(),
+                )
+                .values(expires_at=expires_at)
+            )
+            return cast(CursorResult[Any], result).rowcount == expected_slots
+
+    async def release(self, owner_token: str) -> None:
+        async with self.session_factory() as session, session.begin():
+            await session.execute(
+                delete(ConcurrencyLeaseRecord).where(
+                    ConcurrencyLeaseRecord.owner_token == owner_token
+                )
+            )

app/repositories/subagent_repository.py

@@ -3,7 +3,7 @@
 from typing import Any
 from uuid import uuid4
 
-from sqlalchemy import select, update
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from app.models import SubagentRunRecord
@@ -19,13 +19,20 @@ async def create(self, manifest: dict[str, Any]) -> str:
             session.add(SubagentRunRecord(id=coordinator_id, manifest_json=manifest))
         return coordinator_id
 
-    async def finish(self, coordinator_id: str, errors: dict[str, str]) -> None:
+    async def finish(
+        self,
+        coordinator_id: str,
+        errors: dict[str, str],
+        model_summary: dict[str, Any] | None = None,
+    ) -> None:
         async with self.session_factory() as session, session.begin():
-            await session.execute(
-                update(SubagentRunRecord)
-                .where(SubagentRunRecord.id == coordinator_id)
-                .values(dispatch_finished=True, errors_json=errors)
-            )
+            row = await session.get(SubagentRunRecord, coordinator_id)
+            if row is None:
+                raise LookupError(f"subagent coordinator {coordinator_id} not found")
+            if model_summary is not None:
+                row.manifest_json = {**row.manifest_json, "model_summary": model_summary}
+            row.dispatch_finished = True
+            row.errors_json = errors
 
     async def get(self, coordinator_id: str) -> dict[str, Any]:
         async with self.session_factory() as session:
@@ -52,6 +59,8 @@ async def list_recent(self, limit: int) -> list[dict[str, Any]]:
                     "coordinator_id": row.id,
                     "created_at": row.created_at.isoformat(),
                     "dispatch_finished": row.dispatch_finished,
+                    "mode": row.manifest_json["mode"],
+                    "has_model_summary": "model_summary" in row.manifest_json,
                     "subagents": row.manifest_json["subagents"],
                 }
                 for row in rows

app/runtime.py

@@ -17,6 +17,7 @@
 from app.infrastructure.database import Database
 from app.infrastructure.secrets import build_secret_broker
 from app.repositories.adaptive_repository import AdaptiveRepository
+from app.repositories.concurrency_repository import ConcurrencyRepository
 from app.repositories.equipment_repository import EquipmentRepository
 from app.repositories.event_store import EventStore
 from app.repositories.job_repository import JobRepository
@@ -27,6 +28,7 @@
     AdaptiveRunService,
     DeterministicGrayBoxRunService,
 )
+from app.services.concurrency_limiter import SharedConcurrencyLimiter
 from app.services.equipment_service import EquipmentService
 from app.services.harness_service import HarnessService
 from app.services.job_service import JobApplicationService, JobDispatcher
@@ -142,7 +144,12 @@ async def create_runtime(
         )
         report_service = ReportService(run_repository, equipment_repository)
         subagent_service = SubagentService(
-            SubagentRepository(database.session_factory), adaptive_run_service, report_service
+            SubagentRepository(database.session_factory),
+            adaptive_run_service,
+            report_service,
+            SharedConcurrencyLimiter(
+                ConcurrencyRepository(database.session_factory), config.orchestrator_concurrency
+            ),
         )
         job_application_service = JobApplicationService(
             job_repository,

app/schemas/prompt_schema.py

@@ -11,6 +11,8 @@ class PromptTask(str, Enum):
     planner = "planner"
     planner_tools = "planner_tools"
     model_judge = "model_judge"
+    orchestrator_plan = "orchestrator_plan"
+    orchestrator_summary = "orchestrator_summary"
 
 
 # 定义受控 Prompt 的输入上限。

app/schemas/subagent_schema.py

@@ -1,4 +1,4 @@
-"""主 Agent 的显式委派契约；共享 Policy 是整组预算的上限。"""
+"""Orchestrator/worker 委派契约；共享 Policy 是整组预算的上限。"""
 
 from typing import Self
 
@@ -13,6 +13,7 @@ class SubagentAssignment(BaseModel):
     agent_id: str = Field(pattern=r"^[a-zA-Z0-9][a-zA-Z0-9._-]{0,63}$")
     case_ids: list[str] = Field(min_length=1)
     planner: PlannerConfig | None = None
+    objective: str | None = Field(default=None, max_length=300)
 
     @model_validator(mode="after")
     def unique_cases(self) -> Self:
@@ -25,18 +26,29 @@ class SubagentRunRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
     run: GrayBoxRunRequest
-    subagents: list[SubagentAssignment] = Field(min_length=2, max_length=16)
+    subagents: list[SubagentAssignment] | None = Field(default=None, min_length=2, max_length=16)
+    orchestrator: PlannerConfig | None = None
+    worker_count: int = Field(default=2, ge=2, le=16)
     concurrency: int = Field(default=1, ge=1, le=16)
     parallel_target_safe: bool = False
 
     @model_validator(mode="after")
     def validate_delegation(self) -> Self:
-        count = len(self.subagents)
-        if len({agent.agent_id for agent in self.subagents}) != count:
+        if (self.subagents is None) == (self.orchestrator is None):
+            raise ValueError("provide either subagents or orchestrator")
+        if self.orchestrator is not None and self.orchestrator.backend != "openai_compatible":
+            raise ValueError("orchestrator requires an openai_compatible model")
+        count = len(self.subagents) if self.subagents is not None else self.worker_count
+        if (
+            self.subagents is not None
+            and len({agent.agent_id for agent in self.subagents}) != count
+        ):
             raise ValueError("agent_id must be unique")
         if self.concurrency > 1 and not self.parallel_target_safe:
             raise ValueError("parallel execution requires parallel_target_safe=true")
         for field in ("max_steps", "max_target_calls"):
             if getattr(self.run.policy, field) < count:
                 raise ValueError(f"{field} must allocate at least one per subagent")
+        if self.orchestrator is not None and self.run.policy.max_provider_calls < count + 2:
+            raise ValueError("max_provider_calls must cover planning, workers and summary")
         return self