代码提交

offline-backend-20260512/backend/CLAUDE.md

@@ -1132,6 +1132,26 @@ the initial intent workspace plus the summary/action-plan closing agents. Do
 not delete a built-in id because historical chain nodes persist those ids;
 custom role ids may be added and removed.
 
+### Position-roundtable task-scoped session history
+
+`position_roundtable_sessions.external_task_id` is optional. When populated from the frontend route `taskId`, new rows set `task_scoped = true`: a task can have multiple sessions and all users opening that task see the same sessions, nodes and persisted conversation snapshots. When it is NULL/empty, keep the original `user_id`-scoped personal workspace behavior. `20260808_01` adds `task_scoped` with default `false` (so all historical records remain personal) plus the task/update-time index; it must not backfill or make `external_task_id` non-null.
+
+### Roundtable artifact submission status
+
+`roundtable_artifact_submissions` keeps task-level submission state for a
+generated deliverable and is deliberately separate from task details and any
+external task-system delivery. The API is
+`GET /api/roundtable/tasks/{taskId}/artifacts?submitted=true` and
+`PUT /api/roundtable/tasks/{taskId}/artifacts/submission`. All callers use the
+normal Gateway Bearer authentication; records are shared by task while
+`created_by`/`submitted_by`/`cancelled_by` are audit-only. The server derives a
+SHA-256 idempotency key from `taskId + sourceKey + path + revision`: submitting
+the same revision is an upsert, while cancellation only flips `submitted` and
+sets `cancelled_at` (it never deletes history). Store only the sandbox file
+reference (`threadId`, path, source/node/session fields, name/MIME and revision)
+and timestamps; do not persist Markdown content or call a formal delivery
+interface. Previews continue to read the DeerFlow artifact by `threadId + path`.
+
 See `docs/` directory for detailed documentation:
 - [../README.md](../README.md) - Offline Docker wheelhouse preparation and startup workflow. The pinned versions in `../offline-runtime-requirements.txt` and `../start_backend.sh` must stay synchronized. `start_backend.sh` stays aligned with the deployment script and adds only the Redis wheel fallback line. The download target defaults to CPython 3.12 on Linux x86_64 and verifies dependency closure using wheelhouse only.
 - [CONFIGURATION.md](docs/CONFIGURATION.md) - Configuration options

offline-backend-20260512/backend/README.md

@@ -144,6 +144,8 @@ FastAPI application providing REST endpoints for frontend integration:
 
 The separately deployed TaskCOP Vue app uses `POST /api/v1/auth/login/username` directly (its `username`, `userId`, or `yUserId` URL parameter identifies the user), so this compatibility flow does not depend on the retired Consumer login service. It calls the Gateway address configured as `b1ConsumerUrl` directly; it does not use a Vite `/api` reverse proxy. An empty TaskCOP store remains empty; task data is introduced through normal creation or the JSON import API. The Gateway CORS middleware always merges the local TaskCOP Vite origins `http://localhost:8080` and `http://127.0.0.1:8080`, including when `GATEWAY_CORS_ORIGINS` is set for another frontend. `scripts/env.sh` also includes them in its startup default. For a deployment, set `GATEWAY_CORS_ORIGINS` to the exact additional frontend origins that are allowed to call the Gateway.
 
+Position-roundtable history supports two modes. A non-empty `external_task_id` from the frontend route's `taskId` creates a row with `task_scoped = true`, so all sessions, nodes and persisted conversation snapshots for that task are shared across users; `user_id` is then creator audit metadata only. When the field is absent or empty, the original per-user personal-session behavior remains unchanged. Alembic revision `20260808_01` adds `task_scoped` with the safe default `false` for historical rows plus the task/update-time index.
+
 AI writing supports a sample-imitation mode: the frontend can upload or paste a sample article, the Gateway extracts text for document files, and the `ai_writing` graph runs a `sample_analyzer` node before normal intent/research/outline/draft stages. The resulting style profile is injected into outline and draft prompts with explicit guardrails against copying source sentences, facts, names, or data.
 
 ### Custom Agents
@@ -507,3 +509,23 @@ is required. It is the initial intent-identification workspace and owns the
 summary/action-plan closing agents. Built-in rows retain stable ids so
 historical chain nodes remain readable; they can be edited or disabled but not
 deleted. Custom roles may be freely added or removed.
+
+## Roundtable artifact submission status
+
+The task-scoped roundtable workbench exposes a lightweight submission-status
+API for generated deliverables:
+
+- `GET /api/roundtable/tasks/{taskId}/artifacts?submitted=true` returns the
+  submitted artifact references for a task (or `items: []`).
+- `PUT /api/roundtable/tasks/{taskId}/artifacts/submission` records
+  `submitted: true` or cancels it with `submitted: false`.
+
+Both routes use the normal Gateway `Authorization: Bearer <token>` middleware.
+The record is shared by `taskId`; the authenticated user is retained only in
+audit columns. Its server-owned idempotency key is
+`taskId + sourceKey + path + revision`. Repeating the same submitted payload
+returns the existing record; cancellation updates its state and timestamp
+without deleting the audit row. DeerFlow stores only the artifact reference
+(`threadId` + sandbox `path`, plus the optional session/node/name/MIME fields),
+never the Markdown body and never a task-system delivery. Continue to preview
+the file through the existing thread-artifact endpoint.

offline-backend-20260512/backend/app/gateway/app.py

@@ -65,8 +65,9 @@ from app.gateway.routers import (
     recommend,
     flow_extract,
     recommended_questions,
-    roundtable_chains,
-    roundtable_diagnostics,
+    roundtable_chains,
+    roundtable_artifact_submissions,
+    roundtable_diagnostics,
     roundtable_draft_shares,
     roundtable_drafts,
     roundtable_jobs,
@@ -1465,8 +1466,11 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # Roundtable-planning drafts + recommendation history at /api/roundtable-drafts
     app.include_router(roundtable_drafts.router)
 
-    # Task-scoped roundtable drafts (taskId 深链聊天记录，独立存、不分权) at /api/roundtable-task-drafts
-    app.include_router(roundtable_task_drafts.router)
+    # Task-scoped roundtable drafts (taskId 深链聊天记录，独立存、不分权) at /api/roundtable-task-drafts
+    app.include_router(roundtable_task_drafts.router)
+
+    # Task-scoped artifact submission status; stores references only, never artifact content.
+    app.include_router(roundtable_artifact_submissions.router)
 
     # Step3 流程图结构化抽取 + 校验 (SSE) at /api/flow-extract
     app.include_router(flow_extract.router)

offline-backend-20260512/backend/app/gateway/deps.py

@@ -305,6 +305,11 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
         app.state.task_report_store = make_task_report_store(sf)
 
+        from deerflow.persistence.roundtable_artifact_submissions import make_roundtable_artifact_submission_store
+
+        # Task-scoped artifact submission status: shared by task; actor is audit-only.
+        app.state.roundtable_artifact_submission_store = make_roundtable_artifact_submission_store(sf)
+
         from deerflow.persistence.parallel_agent_tasks import make_parallel_agent_task_store
 
         app.state.parallel_agent_task_store = make_parallel_agent_task_store(sf)

offline-backend-20260512/backend/app/gateway/routers/position_roundtable.py

@@ -15,8 +15,8 @@ from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
-from fastapi import APIRouter, HTTPException, Request
-from pydantic import BaseModel, Field
+from fastapi import APIRouter, HTTPException, Query, Request
+from pydantic import BaseModel, Field, field_validator
 
 from app.gateway.path_utils import aresolve_thread_virtual_path
 from deerflow.runtime.user_context import get_effective_user_id
@@ -79,7 +79,10 @@ class NodeResponse(BaseModel):
 
 class SessionResponse(BaseModel):
     id: str
+    # Older personal rows can already contain a mock task id; task_scoped is
+    # the authoritative opt-in marker for cross-user task sharing.
     external_task_id: str | None = None
+    task_scoped: bool = False
     task_snapshot: dict[str, Any] = Field(default_factory=dict)
     chain_id: str | None = None
     chain_snapshot: dict[str, Any] | None = None
@@ -105,6 +108,14 @@ class SessionCreateRequest(BaseModel):
     external_task_id: str | None = Field(default=None, max_length=128)
     task_snapshot: dict[str, Any] = Field(default_factory=dict)
 
+    @field_validator("external_task_id")
+    @classmethod
+    def _normalize_external_task_id(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        normalized = value.strip()
+        return normalized or None
+
 
 class TaskSnapshotUpdateRequest(BaseModel):
     task_snapshot: dict[str, Any] = Field(default_factory=dict)
@@ -180,6 +191,25 @@ def _current_user_id(request: Request) -> str:
     return str(user.id) if user is not None else get_effective_user_id()
 
 
+def _task_snapshot_for_scope(
+    external_task_id: str | None,
+    task_snapshot: dict[str, Any],
+    *,
+    task_scoped: bool = True,
+) -> dict[str, Any]:
+    """Lock the snapshot id only for an explicit shared task scope."""
+    if not task_scoped or not external_task_id:
+        return dict(task_snapshot)
+    return {**task_snapshot, "id": external_task_id, "externalTaskId": external_task_id}
+
+
+def _normalize_external_task_id(value: str | None) -> str | None:
+    if value is None:
+        return None
+    normalized = value.strip()
+    return normalized or None
+
+
 def _get_store(request: Request):
     store = getattr(request.app.state, "position_roundtable_store", None)
     if store is None:
@@ -1082,9 +1112,13 @@ async def seed_demo_session(
 
 
 @router.get("/sessions", response_model=list[SessionResponse])
-async def list_sessions(request: Request, external_task_id: str | None = None) -> list[SessionResponse]:
+async def list_sessions(
+    request: Request,
+    external_task_id: str | None = Query(default=None, max_length=128),
+) -> list[SessionResponse]:
+    task_id = _normalize_external_task_id(external_task_id)
     rows = await _get_store(request).list_sessions(
-        _current_user_id(request), external_task_id=external_task_id
+        _current_user_id(request), external_task_id=task_id
     )
     return [_project_session(row) for row in rows]
 
@@ -1094,7 +1128,12 @@ async def create_session(request: Request, body: SessionCreateRequest) -> Sessio
     row = await _get_store(request).create_session(
         _current_user_id(request),
         external_task_id=body.external_task_id,
-        task_snapshot=body.task_snapshot,
+        task_snapshot=_task_snapshot_for_scope(
+            body.external_task_id,
+            body.task_snapshot,
+            task_scoped=bool(body.external_task_id),
+        ),
+        task_scoped=bool(body.external_task_id),
     )
     return _project_session(row)
 
@@ -1108,8 +1147,16 @@ async def get_session(request: Request, session_id: str) -> SessionDetailRespons
 async def update_task_snapshot(
     request: Request, session_id: str, body: TaskSnapshotUpdateRequest
 ) -> SessionResponse:
-    store, user_id, _ = await _require_writable_session(request, session_id)
-    row = await store.update_session(session_id, user_id, task_snapshot=body.task_snapshot)
+    store, user_id, current = await _require_writable_session(request, session_id)
+    row = await store.update_session(
+        session_id,
+        user_id,
+        task_snapshot=_task_snapshot_for_scope(
+            current.get("external_task_id"),
+            body.task_snapshot,
+            task_scoped=bool(current.get("task_scoped")),
+        ),
+    )
     if row is None:
         raise HTTPException(status_code=404, detail="Position roundtable session not found")
     return _project_session(row)

offline-backend-20260512/backend/app/gateway/routers/roundtable_artifact_submissions.py

@@ -0,0 +1,196 @@
+"""Task-scoped submission status for roundtable artifacts.
+
+This API records only an artifact reference (thread + sandbox path) and its
+submission state.  It intentionally does not fetch task detail, copy Markdown,
+or deliver a file to an external task system.
+
+Routes (authenticated by the Gateway middleware):
+    GET /api/roundtable/tasks/{task_id}/artifacts?submitted=true
+    PUT /api/roundtable/tasks/{task_id}/artifacts/submission
+"""
+
+from __future__ import annotations
+
+from pathlib import PurePosixPath
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, Query, Request
+from pydantic import AliasChoices, BaseModel, ConfigDict, Field, StrictInt, StrictStr, field_validator
+
+from app.gateway.deps import get_current_user
+from deerflow.persistence.roundtable_artifact_submissions import RoundtableArtifactSubmissionStore
+
+router = APIRouter(prefix="/api/roundtable/tasks", tags=["roundtable-artifact-submissions"])
+
+
+class ArtifactReferenceInput(BaseModel):
+    """Reference to a generated file; the file content remains in its thread sandbox."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    session_id: StrictStr | None = Field(default=None, max_length=128, validation_alias=AliasChoices("sessionId", "session_id"))
+    thread_id: StrictStr | None = Field(default=None, max_length=128, validation_alias=AliasChoices("threadId", "thread_id"))
+    node_key: StrictStr | None = Field(default=None, max_length=256, validation_alias=AliasChoices("nodeKey", "node_key"))
+    source_key: StrictStr = Field(..., min_length=1, max_length=512, validation_alias=AliasChoices("sourceKey", "source_key"))
+    path: StrictStr = Field(..., min_length=1, max_length=2048)
+    name: StrictStr | None = Field(default=None, max_length=512)
+    mime_type: StrictStr | None = Field(default=None, max_length=255, validation_alias=AliasChoices("mimeType", "mime_type"))
+    revision: StrictInt = Field(..., ge=0)
+
+    @field_validator("session_id", "thread_id", "node_key", "source_key", "path", "name", "mime_type", mode="before")
+    @classmethod
+    def _trim_text(cls, value: Any) -> Any:
+        if isinstance(value, str):
+            value = value.strip()
+            return value or None
+        return value
+
+
+class ArtifactSubmissionRequest(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    task_id: StrictStr | None = Field(default=None, max_length=128, validation_alias=AliasChoices("taskId", "task_id"))
+    submitted: bool
+    submission_id: StrictStr | None = Field(default=None, max_length=64, validation_alias=AliasChoices("submissionId", "submission_id"))
+    task_delivery_id: StrictStr | None = Field(default=None, max_length=255, validation_alias=AliasChoices("taskDeliveryId", "task_delivery_id"))
+    artifact: ArtifactReferenceInput
+
+    @field_validator("task_id", "submission_id", "task_delivery_id", mode="before")
+    @classmethod
+    def _trim_optional_text(cls, value: Any) -> Any:
+        if isinstance(value, str):
+            value = value.strip()
+            return value or None
+        return value
+
+
+class ArtifactReferenceResponse(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    session_id: str | None = Field(default=None, serialization_alias="sessionId")
+    thread_id: str | None = Field(default=None, serialization_alias="threadId")
+    node_key: str | None = Field(default=None, serialization_alias="nodeKey")
+    source_key: str = Field(serialization_alias="sourceKey")
+    path: str
+    name: str | None = None
+    mime_type: str | None = Field(default=None, serialization_alias="mimeType")
+    revision: int
+
+
+class ArtifactSubmissionResponse(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    id: str
+    task_id: str = Field(serialization_alias="taskId")
+    submitted: bool
+    submitted_at: str | None = Field(default=None, serialization_alias="submittedAt")
+    cancelled_at: str | None = Field(default=None, serialization_alias="cancelledAt")
+    task_delivery_id: str | None = Field(default=None, serialization_alias="taskDeliveryId")
+    artifact: ArtifactReferenceResponse
+
+
+class ArtifactSubmissionListResponse(BaseModel):
+    items: list[ArtifactSubmissionResponse]
+
+
+def _get_store(request: Request) -> RoundtableArtifactSubmissionStore:
+    store = getattr(request.app.state, "roundtable_artifact_submission_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="Roundtable artifact submission store not available")
+    return store
+
+
+def _task_id_or_422(value: str) -> str:
+    task_id = value.strip()
+    if not task_id:
+        raise HTTPException(status_code=422, detail="task id is required")
+    if len(task_id) > 128:
+        raise HTTPException(status_code=422, detail="task id must be at most 128 characters")
+    return task_id
+
+
+def _artifact_payload(artifact: ArtifactReferenceInput) -> dict[str, Any]:
+    source_key = artifact.source_key
+    path = artifact.path
+    if not source_key or not path:
+        raise HTTPException(status_code=422, detail="artifact.sourceKey and artifact.path are required")
+    name = artifact.name or PurePosixPath(path).name or path
+    return {
+        "session_id": artifact.session_id,
+        "thread_id": artifact.thread_id,
+        "node_key": artifact.node_key,
+        "source_key": source_key,
+        "path": path,
+        "name": name,
+        "mime_type": artifact.mime_type,
+        "revision": artifact.revision,
+    }
+
+
+def _response(row: dict[str, Any]) -> ArtifactSubmissionResponse:
+    return ArtifactSubmissionResponse(
+        id=str(row["id"]),
+        task_id=str(row["task_id"]),
+        submitted=bool(row["submitted"]),
+        submitted_at=row.get("submitted_at"),
+        cancelled_at=row.get("cancelled_at"),
+        task_delivery_id=row.get("task_delivery_id"),
+        artifact=ArtifactReferenceResponse(
+            session_id=row.get("session_id"),
+            thread_id=row.get("thread_id"),
+            node_key=row.get("node_key"),
+            source_key=str(row["source_key"]),
+            path=str(row["path"]),
+            name=row.get("name"),
+            mime_type=row.get("mime_type"),
+            revision=int(row["revision"]),
+        ),
+    )
+
+
+@router.get("/{task_id}/artifacts", response_model=ArtifactSubmissionListResponse)
+async def list_artifact_submissions(
+    request: Request,
+    task_id: str,
+    submitted: bool | None = Query(default=True),
+) -> ArtifactSubmissionListResponse:
+    """List task artifacts by submission state (defaults to submitted only)."""
+    rows = await _get_store(request).list_by_task(_task_id_or_422(task_id), submitted=submitted)
+    return ArtifactSubmissionListResponse(items=[_response(row) for row in rows])
+
+
+@router.put("/{task_id}/artifacts/submission", response_model=ArtifactSubmissionResponse)
+async def set_artifact_submission(
+    request: Request,
+    task_id: str,
+    body: ArtifactSubmissionRequest,
+) -> ArtifactSubmissionResponse:
+    """Create/update a submission or retain an audit row while cancelling it."""
+    path_task_id = _task_id_or_422(task_id)
+    if body.task_id is not None and body.task_id != path_task_id:
+        raise HTTPException(status_code=422, detail="taskId must match the path task id")
+
+    artifact = _artifact_payload(body.artifact)
+    actor_id = await get_current_user(request)
+    store = _get_store(request)
+    if body.submitted:
+        # The frontend previews submitted artifacts by threadId + path.  Without
+        # a thread reference a saved state could never be previewed correctly.
+        if not artifact["thread_id"]:
+            raise HTTPException(status_code=422, detail="artifact.threadId is required when submitted is true")
+        row = await store.submit(
+            path_task_id,
+            artifact,
+            task_delivery_id=body.task_delivery_id,
+            actor_id=actor_id,
+        )
+    else:
+        row = await store.cancel(
+            path_task_id,
+            artifact,
+            submission_id=body.submission_id,
+            actor_id=actor_id,
+        )
+        if row is None:
+            raise HTTPException(status_code=404, detail="Artifact submission not found")
+    return _response(row)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260808_01_position_roundtable_shared_task_scope.py

@@ -0,0 +1,69 @@
+"""Index optional task-scoped position-roundtable history.
+
+Revision ID: 20260808_01
+Revises: 20260729_01
+Create Date: 2026-08-08
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+revision = "20260808_01"
+down_revision = "20260729_01"
+branch_labels = None
+depends_on = None
+
+_TABLE = "position_roundtable_sessions"
+_INDEX = "ix_position_roundtable_sessions_task_updated"
+
+
+def _has_table(table_name: str) -> bool:
+    return table_name in inspect(op.get_bind()).get_table_names()
+
+
+def _has_column(table_name: str, column_name: str) -> bool:
+    if not _has_table(table_name):
+        return False
+    return any(
+        column["name"] == column_name
+        for column in inspect(op.get_bind()).get_columns(table_name)
+    )
+
+
+def _has_index(table_name: str, index_name: str) -> bool:
+    if not _has_table(table_name):
+        return False
+    return index_name in {
+        index["name"] for index in inspect(op.get_bind()).get_indexes(table_name)
+    }
+
+
+def upgrade() -> None:
+    if not _has_table(_TABLE):
+        return
+    # Historical records must stay in the original personal mode. New records
+    # set this flag only when a route taskId explicitly requests sharing.
+    if not _has_column(_TABLE, "task_scoped"):
+        op.add_column(
+            _TABLE,
+            sa.Column(
+                "task_scoped",
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.false(),
+            ),
+        )
+    if not _has_index(_TABLE, _INDEX):
+        op.create_index(_INDEX, _TABLE, ["external_task_id", "updated_at"])
+
+
+def downgrade() -> None:
+    if not _has_table(_TABLE):
+        return
+    if _has_index(_TABLE, _INDEX):
+        op.drop_index(_INDEX, table_name=_TABLE)
+    if _has_column(_TABLE, "task_scoped"):
+        op.drop_column(_TABLE, "task_scoped")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py

@@ -56,6 +56,7 @@ from deerflow.persistence.positions.model import PositionRow, PositionTagBinding
 from deerflow.persistence.recommended_questions.model import RecommendedQuestionRow
 from deerflow.persistence.research_templates.model import ResearchTemplateRow
 from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
+from deerflow.persistence.roundtable_artifact_submissions.model import RoundtableArtifactSubmissionRow
 from deerflow.persistence.roundtable_diagnostics.model import RoundtableDiagnosticRow
 from deerflow.persistence.roundtable_draft_shares.model import RoundtableDraftShareRow
 from deerflow.persistence.roundtable_drafts.model import (
@@ -118,6 +119,7 @@ __all__ = [
     "RecommendedQuestionRow",
     "ResearchTemplateRow",
     "RoundtableChainRow",
+    "RoundtableArtifactSubmissionRow",
     "RoundtableDraftRow",
     "RoundtableDraftShareRow",
     "RoundtableDiagnosticRow",

offline-backend-20260512/backend/packages/harness/deerflow/persistence/position_roundtable/model.py

@@ -12,7 +12,7 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from sqlalchemy import ForeignKey, Integer, String, UniqueConstraint
+from sqlalchemy import Boolean, ForeignKey, Index, Integer, String, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column
 
 from deerflow.persistence.base import Base
@@ -21,12 +21,23 @@ from deerflow.persistence.types import BeijingDateTime, PortableLongText
 
 class PositionRoundtableSessionRow(Base):
     __tablename__ = "position_roundtable_sessions"
+    __table_args__ = (
+        # History is read as "all conversations for one task, newest first".
+        # `user_id` remains audit data only; it must not partition a shared
+        # task workspace.
+        Index(
+            "ix_position_roundtable_sessions_task_updated",
+            "external_task_id",
+            "updated_at",
+        ),
+    )
 
     id: Mapped[str] = mapped_column(String(64), primary_key=True)
-    # Phase one uses normal account isolation. A future membership / role ACL
-    # can be added without changing the session or thread identifiers.
+    # Creator audit field. `task_scoped`, not merely external_task_id, opts a
+    # session into sharing. This keeps historical sidebar/mock rows private.
     user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
     external_task_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    task_scoped: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="0")
     task_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     chain_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     chain_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/position_roundtable/sql.py

@@ -46,11 +46,17 @@ class PositionRoundtableRepository:
     def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
         self._sf = session_factory
 
+    @staticmethod
+    def _can_access(row: PositionRoundtableSessionRow, user_id: str) -> bool:
+        """Task-bound sessions are shared; unbound legacy sessions stay private."""
+        return bool(row.task_scoped) or row.user_id == user_id
+
     @staticmethod
     def _session_dict(row: PositionRoundtableSessionRow) -> dict[str, Any]:
         return {
             "id": row.id,
             "external_task_id": row.external_task_id,
+            "task_scoped": bool(row.task_scoped),
             "task_snapshot": _loads(row.task_snapshot) or {},
             "chain_id": row.chain_id,
             "chain_snapshot": _loads(row.chain_snapshot),
@@ -98,12 +104,14 @@ class PositionRoundtableRepository:
         *,
         external_task_id: str | None,
         task_snapshot: dict[str, Any] | None,
+        task_scoped: bool | None = None,
     ) -> dict[str, Any]:
         now = datetime.now(UTC)
         row = PositionRoundtableSessionRow(
             id=uuid4().hex,
             user_id=user_id,
             external_task_id=external_task_id or None,
+            task_scoped=bool(external_task_id) if task_scoped is None else task_scoped,
             task_snapshot=_dumps(task_snapshot or {}),
             status="intent_pending",
             created_at=now,
@@ -165,7 +173,8 @@ class PositionRoundtableRepository:
         row = PositionRoundtableSessionRow(
             id=session_id,
             user_id=user_id,
-            external_task_id=session_payload.get("external_task_id") or None,
+            external_task_id=str(session_payload.get("external_task_id") or "").strip() or None,
+            task_scoped=False,
             task_snapshot=_dumps(session_payload.get("task_snapshot") or {}),
             chain_id=session_payload.get("chain_id") or None,
             chain_snapshot=_dumps(session_payload.get("chain_snapshot")),
@@ -246,9 +255,17 @@ class PositionRoundtableRepository:
     async def list_sessions(
         self, user_id: str, *, external_task_id: str | None = None, limit: int = 50
     ) -> list[dict[str, Any]]:
-        stmt = select(PositionRoundtableSessionRow).where(PositionRoundtableSessionRow.user_id == user_id)
+        """List shared task history or the caller's legacy personal history."""
         if external_task_id:
-            stmt = stmt.where(PositionRoundtableSessionRow.external_task_id == external_task_id)
+            stmt = select(PositionRoundtableSessionRow).where(
+                PositionRoundtableSessionRow.external_task_id == external_task_id,
+                PositionRoundtableSessionRow.task_scoped.is_(True),
+            )
+        else:
+            stmt = select(PositionRoundtableSessionRow).where(
+                PositionRoundtableSessionRow.user_id == user_id,
+                PositionRoundtableSessionRow.task_scoped.is_(False),
+            )
         stmt = stmt.order_by(PositionRoundtableSessionRow.updated_at.desc()).limit(limit)
         async with self._sf() as session:
             rows = (await session.execute(stmt)).scalars().all()
@@ -257,7 +274,7 @@ class PositionRoundtableRepository:
     async def get_session(self, session_id: str, user_id: str) -> dict[str, Any] | None:
         async with self._sf() as session:
             row = await session.get(PositionRoundtableSessionRow, session_id)
-            if row is None or row.user_id != user_id:
+            if row is None or not self._can_access(row, user_id):
                 return None
             return self._session_dict(row)
 
@@ -280,7 +297,7 @@ class PositionRoundtableRepository:
     ) -> dict[str, Any] | None:
         async with self._sf() as session:
             row = await session.get(PositionRoundtableSessionRow, session_id)
-            if row is None or row.user_id != user_id:
+            if row is None or not self._can_access(row, user_id):
                 return None
             if task_snapshot is not _UNSET:
                 row.task_snapshot = _dumps(task_snapshot)
@@ -326,7 +343,7 @@ class PositionRoundtableRepository:
         now = datetime.now(UTC)
         async with self._sf() as session:
             row = await session.get(PositionRoundtableSessionRow, session_id)
-            if row is None or row.user_id != user_id:
+            if row is None or not self._can_access(row, user_id):
                 return None
             node_rows = (
                 await session.execute(
@@ -371,7 +388,7 @@ class PositionRoundtableRepository:
     async def list_nodes(self, session_id: str, user_id: str) -> list[dict[str, Any]] | None:
         async with self._sf() as session:
             parent = await session.get(PositionRoundtableSessionRow, session_id)
-            if parent is None or parent.user_id != user_id:
+            if parent is None or not self._can_access(parent, user_id):
                 return None
             rows = (
                 await session.execute(
@@ -384,16 +401,13 @@ class PositionRoundtableRepository:
 
     async def get_node(self, session_id: str, node_key: str, user_id: str) -> dict[str, Any] | None:
         async with self._sf() as session:
-            stmt = (
-                select(PositionRoundtableNodeRow)
-                .join(PositionRoundtableSessionRow, PositionRoundtableSessionRow.id == PositionRoundtableNodeRow.session_id)
-                .where(
-                    PositionRoundtableNodeRow.session_id == session_id,
-                    PositionRoundtableNodeRow.node_key == node_key,
-                    PositionRoundtableSessionRow.user_id == user_id,
-                )
-                .limit(1)
-            )
+            parent = await session.get(PositionRoundtableSessionRow, session_id)
+            if parent is None or not self._can_access(parent, user_id):
+                return None
+            stmt = select(PositionRoundtableNodeRow).where(
+                PositionRoundtableNodeRow.session_id == session_id,
+                PositionRoundtableNodeRow.node_key == node_key,
+            ).limit(1)
             row = (await session.execute(stmt)).scalars().first()
             return self._node_dict(row) if row is not None else None
 
@@ -415,16 +429,13 @@ class PositionRoundtableRepository:
         increment_rejection: bool = False,
     ) -> dict[str, Any] | None:
         async with self._sf() as session:
-            stmt = (
-                select(PositionRoundtableNodeRow)
-                .join(PositionRoundtableSessionRow, PositionRoundtableSessionRow.id == PositionRoundtableNodeRow.session_id)
-                .where(
-                    PositionRoundtableNodeRow.session_id == session_id,
-                    PositionRoundtableNodeRow.node_key == node_key,
-                    PositionRoundtableSessionRow.user_id == user_id,
-                )
-                .limit(1)
-            )
+            parent = await session.get(PositionRoundtableSessionRow, session_id)
+            if parent is None or not self._can_access(parent, user_id):
+                return None
+            stmt = select(PositionRoundtableNodeRow).where(
+                PositionRoundtableNodeRow.session_id == session_id,
+                PositionRoundtableNodeRow.node_key == node_key,
+            ).limit(1)
             row = (await session.execute(stmt)).scalars().first()
             if row is None:
                 return None
@@ -472,7 +483,7 @@ class PositionRoundtableRepository:
         now = datetime.now(UTC)
         async with self._sf() as session:
             parent = await session.get(PositionRoundtableSessionRow, session_id)
-            if parent is None or parent.user_id != user_id:
+            if parent is None or not self._can_access(parent, user_id):
                 return None
             rows = (
                 await session.execute(
@@ -538,19 +549,18 @@ class PositionRoundtableRepository:
         return await self.update_session(session_id, user_id, status="archived")
 
     async def delete_session(self, session_id: str, user_id: str) -> bool:
-        """Delete one user-owned history record and all of its node snapshots."""
+        """Delete a task-shared row or the caller's personal legacy row."""
         async with self._sf() as session:
             parent = (
                 await session.execute(
                     select(PositionRoundtableSessionRow)
                     .where(
                         PositionRoundtableSessionRow.id == session_id,
-                        PositionRoundtableSessionRow.user_id == user_id,
                     )
                     .limit(1)
                 )
             ).scalar_one_or_none()
-            if parent is None:
+            if parent is None or not self._can_access(parent, user_id):
                 return False
 
             # Do not rely on database-specific FK cascade settings. SQLite
@@ -564,7 +574,6 @@ class PositionRoundtableRepository:
             await session.execute(
                 delete(PositionRoundtableSessionRow).where(
                     PositionRoundtableSessionRow.id == session_id,
-                    PositionRoundtableSessionRow.user_id == user_id,
                 )
             )
             await session.commit()

offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_artifact_submissions/__init__.py

@@ -0,0 +1,22 @@
+"""Task-scoped roundtable artifact submission status persistence."""
+
+from __future__ import annotations
+
+from deerflow.persistence.roundtable_artifact_submissions.base import RoundtableArtifactSubmissionStore
+from deerflow.persistence.roundtable_artifact_submissions.memory import MemoryRoundtableArtifactSubmissionStore
+from deerflow.persistence.roundtable_artifact_submissions.model import RoundtableArtifactSubmissionRow
+from deerflow.persistence.roundtable_artifact_submissions.sql import RoundtableArtifactSubmissionRepository
+
+__all__ = [
+    "MemoryRoundtableArtifactSubmissionStore",
+    "RoundtableArtifactSubmissionRepository",
+    "RoundtableArtifactSubmissionRow",
+    "RoundtableArtifactSubmissionStore",
+    "make_roundtable_artifact_submission_store",
+]
+
+
+def make_roundtable_artifact_submission_store(session_factory):
+    if session_factory is None:
+        return MemoryRoundtableArtifactSubmissionStore()
+    return RoundtableArtifactSubmissionRepository(session_factory)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_artifact_submissions/base.py

@@ -0,0 +1,44 @@
+"""Persistence contract for task-scoped roundtable artifact submissions.
+
+The store deliberately keeps only a stable reference to an artifact in a
+DeerFlow thread sandbox.  It does not copy the artifact body and never calls a
+downstream task-delivery service.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class RoundtableArtifactSubmissionStore(ABC):
+    """Shared submission-status store, partitioned only by task id."""
+
+    @abstractmethod
+    async def list_by_task(self, task_id: str, *, submitted: bool | None = True) -> list[dict[str, Any]]:
+        """Return artifact submission records for one task, newest first."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def submit(
+        self,
+        task_id: str,
+        artifact: dict[str, Any],
+        *,
+        task_delivery_id: str | None = None,
+        actor_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Create or update the idempotent submission record for an artifact."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def cancel(
+        self,
+        task_id: str,
+        artifact: dict[str, Any],
+        *,
+        submission_id: str | None = None,
+        actor_id: str | None = None,
+    ) -> dict[str, Any] | None:
+        """Mark a submission as cancelled without deleting its audit record."""
+        raise NotImplementedError

offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_artifact_submissions/memory.py

@@ -0,0 +1,127 @@
+"""In-memory roundtable artifact-submission store."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from datetime import UTC, datetime
+from hashlib import sha256
+from typing import Any
+from uuid import uuid4
+
+from deerflow.persistence.roundtable_artifact_submissions.base import RoundtableArtifactSubmissionStore
+from deerflow.persistence.types import BEIJING_TZ
+
+
+def submission_key(task_id: str, artifact: dict[str, Any]) -> str:
+    """Return the canonical server-owned idempotency key for an artifact."""
+    source_key = str(artifact["source_key"])
+    path = str(artifact["path"])
+    revision = int(artifact["revision"])
+    raw = "\x1f".join((task_id, source_key, path, str(revision)))
+    return sha256(raw.encode("utf-8")).hexdigest()
+
+
+def _now() -> str:
+    return datetime.now(UTC).astimezone(BEIJING_TZ).isoformat()
+
+
+def _artifact_values(artifact: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "session_id": artifact.get("session_id"),
+        "thread_id": artifact.get("thread_id"),
+        "node_key": artifact.get("node_key"),
+        "source_key": artifact["source_key"],
+        "path": artifact["path"],
+        "name": artifact.get("name"),
+        "mime_type": artifact.get("mime_type"),
+        "revision": int(artifact["revision"]),
+    }
+
+
+class MemoryRoundtableArtifactSubmissionStore(RoundtableArtifactSubmissionStore):
+    def __init__(self) -> None:
+        self._rows: dict[tuple[str, str], dict[str, Any]] = {}
+
+    async def list_by_task(self, task_id: str, *, submitted: bool | None = True) -> list[dict[str, Any]]:
+        rows = [row for row in self._rows.values() if row["task_id"] == task_id and (submitted is None or row["submitted"] is submitted)]
+        rows.sort(key=lambda row: row["updated_at"], reverse=True)
+        return deepcopy(rows)
+
+    async def submit(
+        self,
+        task_id: str,
+        artifact: dict[str, Any],
+        *,
+        task_delivery_id: str | None = None,
+        actor_id: str | None = None,
+    ) -> dict[str, Any]:
+        key = submission_key(task_id, artifact)
+        row = self._rows.get((task_id, key))
+        values = _artifact_values(artifact)
+        if row is None:
+            now = _now()
+            row = {
+                "id": uuid4().hex,
+                "task_id": task_id,
+                "idempotency_key": key,
+                **values,
+                "task_delivery_id": task_delivery_id,
+                "submitted": True,
+                "submitted_at": now,
+                "cancelled_at": None,
+                "created_by": actor_id,
+                "updated_by": actor_id,
+                "submitted_by": actor_id,
+                "cancelled_by": None,
+                "created_at": now,
+                "updated_at": now,
+            }
+            self._rows[(task_id, key)] = row
+            return deepcopy(row)
+
+        changed = (
+            not row["submitted"]
+            or row.get("task_delivery_id") != task_delivery_id
+            or any(row.get(name) != value for name, value in values.items())
+        )
+        if changed:
+            now = _now()
+            row.update(values)
+            row.update(
+                {
+                    "task_delivery_id": task_delivery_id,
+                    "submitted": True,
+                    "submitted_at": now,
+                    "cancelled_at": None,
+                    "updated_by": actor_id,
+                    "submitted_by": actor_id,
+                    "cancelled_by": None,
+                    "updated_at": now,
+                }
+            )
+        return deepcopy(row)
+
+    async def cancel(
+        self,
+        task_id: str,
+        artifact: dict[str, Any],
+        *,
+        submission_id: str | None = None,
+        actor_id: str | None = None,
+    ) -> dict[str, Any] | None:
+        key = submission_key(task_id, artifact)
+        row = self._rows.get((task_id, key))
+        if row is None or (submission_id is not None and row["id"] != submission_id):
+            return None
+        if row["submitted"]:
+            now = _now()
+            row.update(
+                {
+                    "submitted": False,
+                    "cancelled_at": now,
+                    "updated_by": actor_id,
+                    "cancelled_by": actor_id,
+                    "updated_at": now,
+                }
+            )
+        return deepcopy(row)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_artifact_submissions/model.py

@@ -0,0 +1,56 @@
+"""ORM model for task-scoped roundtable artifact submission status."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import Boolean, Index, Integer, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime
+
+
+class RoundtableArtifactSubmissionRow(Base):
+    """One auditable submission state for one task artifact revision.
+
+    The natural idempotency key is a SHA-256 digest of
+    ``task_id + source_key + artifact_path + revision``.  Keeping the digest
+    short avoids a MySQL unique-index limit while the component columns remain
+    available for diagnostics and response shaping.
+    """
+
+    __tablename__ = "roundtable_artifact_submissions"
+    __table_args__ = (
+        UniqueConstraint("task_id", "idempotency_key", name="uq_roundtable_artifact_submission_task_key"),
+        Index("ix_roundtable_artifact_submission_task_state_updated", "task_id", "submitted", "updated_at"),
+    )
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    task_id: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
+    idempotency_key: Mapped[str] = mapped_column(String(64), nullable=False)
+    source_key: Mapped[str] = mapped_column(String(512), nullable=False)
+    artifact_path: Mapped[str] = mapped_column(String(2048), nullable=False)
+    revision: Mapped[int] = mapped_column(Integer, nullable=False)
+
+    session_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    thread_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    node_key: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    artifact_name: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    mime_type: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    task_delivery_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
+
+    submitted: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    submitted_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
+    cancelled_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
+    created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    updated_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    submitted_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    cancelled_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
+    updated_at: Mapped[datetime] = mapped_column(
+        BeijingDateTime(),
+        nullable=False,
+        default=lambda: datetime.now(UTC),
+        onupdate=lambda: datetime.now(UTC),
+    )

offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_artifact_submissions/sql.py

@@ -0,0 +1,188 @@
+"""SQLAlchemy-backed task-scoped roundtable artifact-submission store."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+from uuid import uuid4
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.roundtable_artifact_submissions.base import RoundtableArtifactSubmissionStore
+from deerflow.persistence.roundtable_artifact_submissions.memory import submission_key
+from deerflow.persistence.roundtable_artifact_submissions.model import RoundtableArtifactSubmissionRow
+
+
+def _timestamp(value: datetime | None) -> str | None:
+    return value.isoformat() if isinstance(value, datetime) else value
+
+
+def _to_dict(row: RoundtableArtifactSubmissionRow) -> dict[str, Any]:
+    return {
+        "id": row.id,
+        "task_id": row.task_id,
+        "idempotency_key": row.idempotency_key,
+        "session_id": row.session_id,
+        "thread_id": row.thread_id,
+        "node_key": row.node_key,
+        "source_key": row.source_key,
+        "path": row.artifact_path,
+        "name": row.artifact_name,
+        "mime_type": row.mime_type,
+        "revision": row.revision,
+        "task_delivery_id": row.task_delivery_id,
+        "submitted": row.submitted,
+        "submitted_at": _timestamp(row.submitted_at),
+        "cancelled_at": _timestamp(row.cancelled_at),
+        "created_by": row.created_by,
+        "updated_by": row.updated_by,
+        "submitted_by": row.submitted_by,
+        "cancelled_by": row.cancelled_by,
+        "created_at": _timestamp(row.created_at),
+        "updated_at": _timestamp(row.updated_at),
+    }
+
+
+def _apply_artifact(row: RoundtableArtifactSubmissionRow, artifact: dict[str, Any]) -> None:
+    row.session_id = artifact.get("session_id")
+    row.thread_id = artifact.get("thread_id")
+    row.node_key = artifact.get("node_key")
+    row.source_key = artifact["source_key"]
+    row.artifact_path = artifact["path"]
+    row.artifact_name = artifact.get("name")
+    row.mime_type = artifact.get("mime_type")
+    row.revision = int(artifact["revision"])
+
+
+def _requires_submission_update(
+    row: RoundtableArtifactSubmissionRow,
+    artifact: dict[str, Any],
+    task_delivery_id: str | None,
+) -> bool:
+    return (
+        not row.submitted
+        or row.task_delivery_id != task_delivery_id
+        or row.session_id != artifact.get("session_id")
+        or row.thread_id != artifact.get("thread_id")
+        or row.node_key != artifact.get("node_key")
+        or row.artifact_name != artifact.get("name")
+        or row.mime_type != artifact.get("mime_type")
+    )
+
+
+class RoundtableArtifactSubmissionRepository(RoundtableArtifactSubmissionStore):
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    async def list_by_task(self, task_id: str, *, submitted: bool | None = True) -> list[dict[str, Any]]:
+        stmt = select(RoundtableArtifactSubmissionRow).where(RoundtableArtifactSubmissionRow.task_id == task_id)
+        if submitted is not None:
+            stmt = stmt.where(RoundtableArtifactSubmissionRow.submitted.is_(submitted))
+        stmt = stmt.order_by(RoundtableArtifactSubmissionRow.updated_at.desc())
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return [_to_dict(row) for row in result.scalars()]
+
+    async def submit(
+        self,
+        task_id: str,
+        artifact: dict[str, Any],
+        *,
+        task_delivery_id: str | None = None,
+        actor_id: str | None = None,
+    ) -> dict[str, Any]:
+        key = submission_key(task_id, artifact)
+        stmt = select(RoundtableArtifactSubmissionRow).where(
+            RoundtableArtifactSubmissionRow.task_id == task_id,
+            RoundtableArtifactSubmissionRow.idempotency_key == key,
+        )
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            row = result.scalars().first()
+            now = datetime.now(UTC)
+            created = row is None
+            if row is None:
+                row = RoundtableArtifactSubmissionRow(
+                    id=uuid4().hex,
+                    task_id=task_id,
+                    idempotency_key=key,
+                    task_delivery_id=task_delivery_id,
+                    submitted=True,
+                    submitted_at=now,
+                    created_by=actor_id,
+                    updated_by=actor_id,
+                    submitted_by=actor_id,
+                    created_at=now,
+                    updated_at=now,
+                )
+                _apply_artifact(row, artifact)
+                session.add(row)
+            elif _requires_submission_update(row, artifact, task_delivery_id):
+                _apply_artifact(row, artifact)
+                row.task_delivery_id = task_delivery_id
+                row.submitted = True
+                row.submitted_at = now
+                row.cancelled_at = None
+                row.updated_by = actor_id
+                row.submitted_by = actor_id
+                row.cancelled_by = None
+                row.updated_at = now
+            try:
+                await session.commit()
+            except IntegrityError:
+                # Two callers can both observe a missing row.  The unique key
+                # turns that race into a normal idempotent retry instead of a
+                # 500.  If their non-key metadata differs, apply this request
+                # after loading the winner so PUT still has upsert semantics.
+                await session.rollback()
+                if not created:
+                    raise
+                result = await session.execute(stmt)
+                row = result.scalars().first()
+                if row is None:
+                    raise
+                if _requires_submission_update(row, artifact, task_delivery_id):
+                    now = datetime.now(UTC)
+                    _apply_artifact(row, artifact)
+                    row.task_delivery_id = task_delivery_id
+                    row.submitted = True
+                    row.submitted_at = now
+                    row.cancelled_at = None
+                    row.updated_by = actor_id
+                    row.submitted_by = actor_id
+                    row.cancelled_by = None
+                    row.updated_at = now
+                    await session.commit()
+            await session.refresh(row)
+            return _to_dict(row)
+
+    async def cancel(
+        self,
+        task_id: str,
+        artifact: dict[str, Any],
+        *,
+        submission_id: str | None = None,
+        actor_id: str | None = None,
+    ) -> dict[str, Any] | None:
+        key = submission_key(task_id, artifact)
+        stmt = select(RoundtableArtifactSubmissionRow).where(
+            RoundtableArtifactSubmissionRow.task_id == task_id,
+            RoundtableArtifactSubmissionRow.idempotency_key == key,
+        )
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            row = result.scalars().first()
+            if row is None or (submission_id is not None and row.id != submission_id):
+                return None
+            if row.submitted:
+                now = datetime.now(UTC)
+                row.submitted = False
+                row.cancelled_at = now
+                row.updated_by = actor_id
+                row.cancelled_by = actor_id
+                row.updated_at = now
+                await session.commit()
+                await session.refresh(row)
+            return _to_dict(row)

offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql

@@ -117,6 +117,11 @@ CALL _add_column_if_missing('knowledge_notes', 'folder', 'VARCHAR(512) NULL DEFA
 --          ORM: RoundtableJobRow.task_id = String(128), nullable, indexed。
 CALL _add_column_if_missing('roundtable_jobs', 'task_id', 'VARCHAR(128) NULL DEFAULT NULL');
 
+-- ---- position_roundtable_sessions: opt-in task sharing, migration 20260808_01 ----
+-- Existing records remain personal (`task_scoped = 0`). New records become
+-- shared only when the caller explicitly sends a non-empty route taskId.
+CALL _add_column_if_missing('position_roundtable_sessions', 'task_scoped', 'TINYINT(1) NOT NULL DEFAULT 0');
+
 -- ---- 按钮管理「外部链接」追加登录参数,对应 migration 20260625_01 ----------------
 -- login_params_json: url 外链追加的登录参数列表 JSON([{key,source,custom_value},...]),
 --                    由前端运行时从 userInfo 解析后拼到地址尾部;NULL/空 = 不追加(旧行为)。
@@ -199,5 +204,6 @@ SELECT table_name, column_name
         OR (table_name = 'agents' AND column_name IN ('square_id'))
         OR (table_name = 'roundtable_drafts' AND column_name = 'task_id')
         OR (table_name = 'roundtable_jobs' AND column_name = 'task_id')
+        OR (table_name = 'position_roundtable_sessions' AND column_name = 'task_scoped')
         OR (table_name = 'task_buttons' AND column_name IN ('append_auth', 'auth_token_param', 'auth_name_param')))
  ORDER BY table_name, column_name;

offline-backend-20260512/backend/scripts/sql/leaderboard_indexes_mysql.sql

@@ -68,6 +68,10 @@ CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_run_creat
 CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_created_user', '`created_at`, `user_id`');
 CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_user_created', '`user_id`, `created_at`');
 
+-- ---- position_roundtable_sessions --------------------------------------------
+-- Explicit taskId history is loaded by taskId and newest update time.
+CALL _create_index_if_missing('position_roundtable_sessions', 'ix_position_roundtable_sessions_task_updated', '`external_task_id`, `updated_at`');
+
 -- ---- 清理临时存储过程 --------------------------------------------------------
 DROP PROCEDURE IF EXISTS _create_index_if_missing;
 

offline-backend-20260512/backend/tests/test_position_roundtable_sessions.py

@@ -208,14 +208,91 @@ async def test_conversation_snapshots_round_trip_and_history_delete_removes_node
     assert saved_node is not None
     assert saved_node["conversation_snapshot"] == node_conversation
 
-    # Ownership checks apply to delete as well.
-    assert not await store.delete_session(created["id"], "user-b")
-    assert await store.get_session(created["id"], "user-a") is not None
-    assert await store.delete_session(created["id"], "user-a")
+    # Position roundtable history is shared by taskId, so a collaborator can
+    # read and remove the same task conversation without being its creator.
+    assert await store.get_session(created["id"], "user-b") is not None
+    assert await store.delete_session(created["id"], "user-b")
     assert await store.get_session(created["id"], "user-a") is None
     assert await store.list_nodes(created["id"], "user-a") is None
 
 
+@pytest.mark.asyncio
+async def test_task_scoped_history_is_shared_but_never_mixes_tasks(store):
+    shared = await store.create_session(
+        "user-a",
+        external_task_id="external-task-100",
+        task_snapshot={"id": "external-task-100", "title": "共享任务"},
+    )
+    await store.create_session(
+        "user-b",
+        external_task_id="external-task-200",
+        task_snapshot={"id": "external-task-200", "title": "另一任务"},
+    )
+
+    visible_to_collaborator = await store.list_sessions(
+        "user-b", external_task_id="external-task-100"
+    )
+    assert [row["id"] for row in visible_to_collaborator] == [shared["id"]]
+
+    updated = await store.update_session(
+        shared["id"],
+        "user-b",
+        task_snapshot={"id": "external-task-100", "title": "协作者已更新"},
+    )
+    assert updated is not None
+    assert updated["task_snapshot"]["title"] == "协作者已更新"
+    assert (await store.get_session(shared["id"], "user-a"))["task_snapshot"]["title"] == "协作者已更新"
+
+
+@pytest.mark.asyncio
+async def test_unscoped_history_keeps_original_user_isolation(store):
+    personal = await store.create_session(
+        "user-a",
+        external_task_id=None,
+        task_snapshot={"id": "legacy-card-id", "title": "个人会话"},
+    )
+
+    assert await store.get_session(personal["id"], "user-b") is None
+    assert await store.list_sessions("user-b") == []
+    assert await store.update_session(personal["id"], "user-b", status="active") is None
+    assert not await store.delete_session(personal["id"], "user-b")
+
+    # Old client records could already carry a mock task id. The new database
+    # marker keeps them personal after the migration; only a new explicit
+    # route taskId creates a shared row.
+    legacy_with_mock_task_id = await store.create_session(
+        "user-a",
+        external_task_id="legacy-mock-task-id",
+        task_scoped=False,
+        task_snapshot={"id": "legacy-card-id", "title": "鏃х増涓汉浼氳瘽"},
+    )
+    assert await store.get_session(legacy_with_mock_task_id["id"], "user-b") is None
+    assert await store.list_sessions("user-b", external_task_id="legacy-mock-task-id") == []
+
+
+def test_session_create_allows_missing_task_id_and_scoped_snapshots_keep_scope():
+    body = position_roundtable_router.SessionCreateRequest(
+        external_task_id="  task-route-001  ",
+        task_snapshot={"id": "wrong-id", "title": "任务"},
+    )
+    assert body.external_task_id == "task-route-001"
+    assert position_roundtable_router._task_snapshot_for_scope(
+        body.external_task_id,
+        body.task_snapshot,
+    ) == {
+        "id": "task-route-001",
+        "externalTaskId": "task-route-001",
+        "title": "任务",
+    }
+    assert position_roundtable_router.SessionCreateRequest(task_snapshot={}).external_task_id is None
+    assert position_roundtable_router.SessionCreateRequest(external_task_id="   ").external_task_id is None
+    assert position_roundtable_router._task_snapshot_for_scope(
+        "legacy-mock-task-id",
+        {"id": "legacy-card-id", "title": "旧记录"},
+        task_scoped=False,
+    ) == {"id": "legacy-card-id", "title": "旧记录"}
+
+
 @pytest.mark.asyncio
 async def test_demo_seed_is_complete_idempotent_and_user_scoped(store):
     payload = {
@@ -490,6 +567,7 @@ async def test_builtin_summary_completion_persists_fallback_markdown_when_agent_
     }
     session = {
         "id": "session-a",
+        "external_task_id": "task-fallback",
         "status": "active",
         "task_snapshot": {},
         "chain_id": "chain-a",
@@ -561,7 +639,7 @@ async def test_builtin_summary_completion_persists_fallback_markdown_when_agent_
 
 
 @pytest.mark.asyncio
-async def test_node_updates_are_user_scoped_and_keep_thread_out_of_messages(store):
+async def test_node_updates_are_user_scoped_without_task_id(store):
     created = await store.create_session("user-a", external_task_id=None, task_snapshot={})
     _, nodes = await store.activate_session(
         created["id"],
@@ -602,7 +680,9 @@ async def test_node_updates_are_user_scoped_and_keep_thread_out_of_messages(stor
 
 @pytest.mark.asyncio
 async def test_reject_node_marks_current_and_invalidates_later_stages(store):
-    created = await store.create_session("user-a", external_task_id=None, task_snapshot={})
+    created = await store.create_session(
+        "user-a", external_task_id="task-reject", task_snapshot={}
+    )
     _, nodes = await store.activate_session(
         created["id"],
         "user-a",
@@ -682,7 +762,7 @@ async def test_reject_node_marks_current_and_invalidates_later_stages(store):
 async def test_builtin_summary_and_action_plan_snapshots_are_invalidated_by_reject(store):
     created = await store.create_session(
         "user-a",
-        external_task_id=None,
+        external_task_id="task-builtin-invalidation",
         task_snapshot={
             "title": "测试任务",
             "description": "形成协同方案",