feat: add scoped subject-turn submission

README.md

@@ -19,6 +19,56 @@ The Automation Service owns automation definitions, cron scheduling, webhooks, r
 
 ## Development
 
+### Run and conversation execution
+
+Set `AUTOMATION_AGENT_SERVER_URL` and `AUTOMATION_AGENT_SERVER_API_KEY`, then
+choose an `execution_scope` and optional saved `agent_profile_id` for each
+automation. Both values are snapshotted when a run is queued. Scope controls
+lifecycle: `run` executes the bundle without creating a conversation, while
+`conversation` creates a persistent conversation and requires a profile.
+Profile selection independently controls agent settings and credentials. A
+profile on run-scoped work limits the saved secrets available to its command.
+
+A running automation can submit work for an external subject to
+`POST /v1/runs/{run_id}/subject-turns` with a source, stable subject key,
+prompt, and idempotency key. The service creates or resumes that subject's
+deterministic conversation. The scanner never receives runtime credentials or
+attaches to the conversation itself, and one short run can fan out several
+independent conversations within the configured concurrency limit.
+
+Only conversation-scoped execution reads the server's authoritative
+`conversation_runtime`. The service uses the same profile-backed conversation
+API in local and Docker workspaces, so automation code remains independent of
+workspace kind. Both workspace kinds supply `AUTOMATION_CONVERSATION_ID`,
+`AGENT_SERVER_URL`,
+`SESSION_API_KEY`, and `WORKSPACE_BASE`, and use conversation-scoped upload,
+bash execution, and completion verification. Local workspaces live in per-run
+subdirectories of the configured workspace root. Docker workspaces use
+`/workspace` and receive only the selected inner session key, never the outer
+server key or shared callback key. Local mode retains its existing single-tenant
+server credential boundary; a local workspace is not a security sandbox.
+
+Conversation completion is detected by the watchdog through the scoped SDK
+runtime on each scan (`AUTOMATION_WATCHDOG_INTERVAL_SECONDS`, default 60 seconds).
+This is the primary completion path for conversation-scoped runs: workers
+deliberately do not receive the shared Automation callback credential.
+Run-scoped execution retains its callback path.
+
+`AUTOMATION_CONVERSATION_MAX_CONCURRENT_RUNS` defaults to 2. Docker servers must
+support runtime credential provisioning and release; bound container CPU, memory,
+and PIDs in the server configuration. Completed Docker runtimes are released while
+history remains; the persistent local server and its history are retained.
+
+The Agent Server resolves each selected profile at dispatch, including its
+model, tools, and `secret_refs`. Missing profiles or selected secrets fail
+instead of falling back to a more privileged scope. A separate `model`
+selection cannot be combined with an agent profile.
+
+Profile selection is advertised by the `agentProfiles` capability when an
+Agent Server is configured. Cloud dispatch without a configured Agent Server
+retains its existing behavior and rejects explicit profile selections. Local
+workspaces share the host security boundary; use Docker for process isolation.
+
 ### Prerequisites
 
 - Python 3.12+
@@ -101,3 +151,30 @@ containers/          # Docker configuration
 ## Deployment
 
 This service is deployed via the [deploy repository](https://github.com/All-Hands-AI/deploy). Docker images are automatically built and pushed to `ghcr.io/openhands/automation` on every push to main and on tags.
+
+Each automation chooses its saved Agent Server profile through the create or
+patch API. Profile IDs are included in git sync and run history:
+
+```json
+{
+  "agent_profile_id": "11111111-1111-4111-8111-111111111111",
+  "execution_scope": "run"
+}
+```
+
+The automation definition contains no token values or host-side permission
+map. Manage credential availability through the profile's `secret_refs` and
+the Agent Server's secret store.
+
+### SDK Integration Dependency
+
+The conversation backend and Agent Server execution helpers reuse `RemoteConversation` and `RemoteWorkspace`
+from [software-agent-sdk #5010](https://github.com/OpenHands/software-agent-sdk/pull/5010).
+Profile-scoped script commands use
+[software-agent-sdk #5046](https://github.com/OpenHands/software-agent-sdk/pull/5046).
+This draft pins the SDK implementation by immutable Git commit so its tests and
+source installation are reproducible. Replace that integration pin with the SDK
+release before merging. The live factory additionally integrates the server
+runtime stack; those server changes are separate from this SDK dependency.
+Workflow bundles
+receive the same environment contract in local and Docker workspaces.

migrations/versions/025_add_execution_scope.py

@@ -0,0 +1,40 @@
+"""Record whether a run or conversation owns each execution.
+
+Revision ID: 025
+Revises: 024
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = "025"
+down_revision = "024"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "automations",
+        sa.Column(
+            "execution_scope",
+            sa.String(20),
+            nullable=False,
+            server_default="run",
+        ),
+    )
+    op.add_column(
+        "automation_runs",
+        sa.Column(
+            "execution_scope",
+            sa.String(20),
+            nullable=False,
+            server_default="run",
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("automation_runs", "execution_scope")
+    op.drop_column("automations", "execution_scope")

migrations/versions/026_add_agent_profile.py

@@ -0,0 +1,24 @@
+"""Persist automation profile selection and snapshot it on queued runs.
+
+Revision ID: 026
+Revises: 025
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = "026"
+down_revision = "025"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    for table in ("automations", "automation_runs"):
+        op.add_column(table, sa.Column("agent_profile_id", sa.Uuid(), nullable=True))
+
+
+def downgrade() -> None:
+    for table in ("automation_runs", "automations"):
+        op.drop_column(table, "agent_profile_id")

migrations/versions/027_add_conversation_turn_runs.py

@@ -0,0 +1,74 @@
+"""Add service-owned conversation-turn runs.
+
+Revision ID: 027
+Revises: 026
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = "027"
+down_revision = "026"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.drop_index("ix_automation_runs_subject", table_name="automation_runs")
+    op.add_column(
+        "automation_runs", sa.Column("subject_source", sa.String(100), nullable=True)
+    )
+    op.add_column(
+        "automation_runs", sa.Column("conversation_turn", sa.Text(), nullable=True)
+    )
+    op.add_column(
+        "automation_runs",
+        sa.Column("conversation_wake_agent", sa.Boolean(), nullable=True),
+    )
+    op.execute(
+        """
+        UPDATE automation_runs
+        SET subject_source = (
+            SELECT json_extract(automations.trigger, '$.source')
+            FROM automations
+            WHERE automations.id = automation_runs.automation_id
+        )
+        WHERE subject_key IS NOT NULL
+        """
+        if op.get_context().dialect.name == "sqlite"
+        else """
+        UPDATE automation_runs AS runs
+        SET subject_source = automations.trigger ->> 'source'
+        FROM automations
+        WHERE automations.id = runs.automation_id
+          AND runs.subject_key IS NOT NULL
+        """
+    )
+    op.create_index(
+        "ix_automation_runs_subject",
+        "automation_runs",
+        ["automation_id", "subject_source", "subject_key", "created_at"],
+        unique=False,
+        postgresql_where=sa.text(
+            "subject_key IS NOT NULL AND subject_released_at IS NULL"
+        ),
+        sqlite_where=sa.text("subject_key IS NOT NULL AND subject_released_at IS NULL"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_automation_runs_subject", table_name="automation_runs")
+    op.drop_column("automation_runs", "conversation_turn")
+    op.drop_column("automation_runs", "conversation_wake_agent")
+    op.drop_column("automation_runs", "subject_source")
+    op.create_index(
+        "ix_automation_runs_subject",
+        "automation_runs",
+        ["automation_id", "subject_key", "created_at"],
+        unique=False,
+        postgresql_where=sa.text(
+            "subject_key IS NOT NULL AND subject_released_at IS NULL"
+        ),
+        sqlite_where=sa.text("subject_key IS NOT NULL AND subject_released_at IS NULL"),
+    )

migrations/versions/028_add_subject_turn_requests.py

@@ -0,0 +1,54 @@
+"""Add idempotent subject-turn requests.
+
+Revision ID: 028
+Revises: 027
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = "028"
+down_revision = "027"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "automation_subject_turns",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("automation_id", sa.Uuid(), nullable=False),
+        sa.Column("requester_run_id", sa.Uuid(), nullable=False),
+        sa.Column("subject_run_id", sa.Uuid(), nullable=False),
+        sa.Column("source", sa.String(100), nullable=False),
+        sa.Column("subject_key", sa.String(500), nullable=False),
+        sa.Column("idempotency_key", sa.String(500), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(
+            ["automation_id"], ["automations.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(
+            ["requester_run_id"], ["automation_runs.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(
+            ["subject_run_id"], ["automation_runs.id"], ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "automation_id",
+            "source",
+            "subject_key",
+            "idempotency_key",
+            name="uq_automation_subject_turn_idempotency",
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("automation_subject_turns")

openhands/automation/app.py

@@ -32,6 +32,7 @@
 from openhands.automation.router import router
 from openhands.automation.scheduler import scheduler_loop
 from openhands.automation.streams import stream_supervisor_loop
+from openhands.automation.subject_router import router as subject_router
 from openhands.automation.telemetry_router import router as telemetry_router
 from openhands.automation.uploads import router as uploads_router
 from openhands.automation.utils.version import get_sdk_version, get_server_version_info
@@ -295,6 +296,7 @@ def _create_app() -> FastAPI:
 app.include_router(webhook_router, prefix=_base_path)
 app.include_router(telemetry_router, prefix=_base_path)
 app.include_router(git_sync_router, prefix=_base_path)
+app.include_router(subject_router, prefix=_base_path)
 
 app.include_router(kv_router, prefix=_base_path)
 app.include_router(router, prefix=_base_path)