Enforce task session binding for authoring apply

services/api-server/src/api_server/main.py

@@ -3260,6 +3260,68 @@ def _authoring_guard_checks(
     ]
 
 
+def _authoring_task_session_check(
+    project_id: str,
+    task_id: str | None,
+    session_id: str | None,
+) -> AuthoringGuardCheck:
+    if not task_id:
+        return AuthoringGuardCheck(
+            name="task-session",
+            status="BLOCKED",
+            message="Task id is required for workspace apply",
+        )
+    task = _collaboration.tasks.get(task_id)
+    if task is None:
+        return AuthoringGuardCheck(
+            name="task-session",
+            status="BLOCKED",
+            message=f"Task {task_id} was not found",
+        )
+    if task.project_id != project_id:
+        return AuthoringGuardCheck(
+            name="task-session",
+            status="BLOCKED",
+            message=f"Task {task_id} belongs to project {task.project_id}",
+        )
+    if task.status.value in {"DONE", "CANCELED"}:
+        return AuthoringGuardCheck(
+            name="task-session",
+            status="BLOCKED",
+            message=f"Task {task_id} is {task.status.value}",
+        )
+    if not session_id:
+        return AuthoringGuardCheck(
+            name="task-session",
+            status="BLOCKED",
+            message="Session id is required for workspace apply",
+        )
+    session = _collaboration.sessions.get(session_id)
+    if session is None:
+        return AuthoringGuardCheck(
+            name="task-session",
+            status="BLOCKED",
+            message=f"Session {session_id} was not found",
+        )
+    if session.task_id != task_id:
+        return AuthoringGuardCheck(
+            name="task-session",
+            status="BLOCKED",
+            message=f"Session {session_id} belongs to task {session.task_id}",
+        )
+    if session.finished_at is not None:
+        return AuthoringGuardCheck(
+            name="task-session",
+            status="BLOCKED",
+            message=f"Session {session_id} is already finished",
+        )
+    return AuthoringGuardCheck(
+        name="task-session",
+        status="OK",
+        message=f"Task {task_id} and session {session_id} are active for project {project_id}",
+    )
+
+
 def _authoring_target_node(snapshot: SirSnapshot, request: AuthoringSemanticDiffPreviewRequest):
     if request.target_lineage_id:
         found = next((node for node in snapshot.nodes if node.lineage_id == request.target_lineage_id), None)
@@ -3393,6 +3455,7 @@ def _authoring_semantic_diff_preview(
             user_id=request.user_id,
         ),
     )
+    checks.append(_authoring_task_session_check(project_id, request.task_id, request.session_id))
     version_preview = _authoring_version_preview(target, request.proposed_text, request.task_id, request.session_id)
     return AuthoringSemanticDiffPreviewResponse(
         project_id=project_id,
@@ -5219,6 +5282,7 @@ def _authoring_metadata_object_preview(
         AuthoringGuardCheck(name="preview", status="REQUIRED", message="Metadata draft must be reviewed before apply"),
         AuthoringGuardCheck(name="workspace-history", status="READY", message="Draft can be saved to SFERA workspace history"),
         AuthoringGuardCheck(name="production-1c", status="BLOCKED", message="Production 1C metadata write is disabled"),
+        _authoring_task_session_check(project_id, request.task_id, request.session_id),
     ]
     return AuthoringMetadataObjectPreviewResponse(
         project_id=project_id,
@@ -5706,6 +5770,12 @@ async def authoring_apply_metadata_object(
         raise HTTPException(status_code=400, detail="No metadata draft to apply")
     if preview.version_preview.next_version_id != request.expected_next_version_id:
         raise HTTPException(status_code=409, detail="Expected version id does not match current metadata preview")
+    blocking_checks = [
+        check for check in preview.checks
+        if check.status == "BLOCKED" and check.name not in {"production-1c"}
+    ]
+    if blocking_checks:
+        raise HTTPException(status_code=409, detail={"blocked_checks": [check.model_dump(mode="json") for check in blocking_checks]})
     version, change_id, path = _persist_authoring_metadata_object(project_id, preview, request)
     return AuthoringApplyMetadataObjectResponse(
         project_id=project_id,
@@ -5748,6 +5818,9 @@ async def authoring_apply_rollback(
         raise HTTPException(status_code=409, detail="Expected rollback version id does not match current preview")
     if not preview.apply_available:
         raise HTTPException(status_code=409, detail="Rollback apply is not available")
+    task_session_check = _authoring_task_session_check(project_id, request.task_id, request.session_id)
+    if task_session_check.status == "BLOCKED":
+        raise HTTPException(status_code=409, detail={"blocked_checks": [task_session_check.model_dump(mode="json")]})
     version, path = _persist_authoring_rollback(project_id, change_payload, preview, request)
     return AuthoringApplyRollbackResponse(
         project_id=project_id,

services/api-server/tests/test_api.py

@@ -10,6 +10,21 @@ from api_server.main import app
 from one_c_normalizer import ConfigurationRoot, MetadataGroup, MetadataObject, Module, NormalizedProject
 
 
+def create_authoring_session(client: TestClient, project_id: str, task_id: str, session_id: str, user_id: str = "dev.ivan") -> None:
+    user = client.post("/collaboration/users", json={"user_id": user_id, "display_name": user_id})
+    assert user.status_code == 200
+    task = client.post(
+        "/collaboration/tasks",
+        json={"task_id": task_id, "project_id": project_id, "title": f"Authoring {task_id}", "assignee_user_id": user_id},
+    )
+    assert task.status_code == 200
+    session = client.post(
+        "/collaboration/sessions",
+        json={"session": {"session_id": session_id, "task_id": task_id, "user_id": user_id}},
+    )
+    assert session.status_code == 200
+
+
 def test_cors_allows_lan_panel_origin():
     client = TestClient(app)
     response = client.options(
@@ -1931,6 +1946,8 @@ def test_authoring_context_and_completion_preview(tmp_path: Path):
 
     indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id})
     assert indexed.status_code == 200
+    create_authoring_session(client, project_id, "task.authoring", "session.authoring")
+    create_authoring_session(client, project_id, "task.rollback", "session.rollback")
 
     context = client.post(
         f"/projects/{project_id}/authoring/context",
@@ -1993,6 +2010,7 @@ def test_authoring_context_and_completion_preview(tmp_path: Path):
     assert diff_payload["target"]["name"] == "Проведение"
     assert diff_payload["version_preview"]["task_id"] == "task.authoring"
     assert diff_payload["version_preview"]["apply_available"] is False
+    assert any(row["name"] == "task-session" and row["status"] == "OK" for row in diff_payload["checks"])
     assert any(row["name"] == "apply" and row["status"] == "BLOCKED" for row in diff_payload["checks"])
 
     apply_response = client.post(
@@ -2093,6 +2111,45 @@ def test_authoring_context_and_completion_preview(tmp_path: Path):
     assert production_apply.status_code == 403
 
 
+def test_authoring_apply_requires_active_task_session(tmp_path: Path):
+    project_id = f"authoring-guard-{uuid4()}"
+    module = tmp_path / "guard_module.bsl"
+    source_text = "Процедура Проверить()\nКонецПроцедуры\n"
+    module.write_text(source_text, encoding="utf-8")
+    client = TestClient(app)
+
+    indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id})
+    assert indexed.status_code == 200
+
+    preview = client.post(
+        f"/projects/{project_id}/authoring/semantic-diff-preview",
+        json={
+            "routine_name": "Проверить",
+            "source_path": str(module),
+            "original_text": source_text,
+            "proposed_text": source_text.replace("КонецПроцедуры", "    Возврат;\nКонецПроцедуры"),
+        },
+    )
+    assert preview.status_code == 200
+    preview_payload = preview.json()
+    assert any(check["name"] == "task-session" and check["status"] == "BLOCKED" for check in preview_payload["checks"])
+
+    apply_response = client.post(
+        f"/projects/{project_id}/authoring/apply-change-set",
+        json={
+            "routine_name": "Проверить",
+            "source_path": str(module),
+            "original_text": source_text,
+            "proposed_text": source_text.replace("КонецПроцедуры", "    Возврат;\nКонецПроцедуры"),
+            "expected_next_version_id": preview_payload["version_preview"]["next_version_id"],
+            "approved_by": "dev.ivan",
+        },
+    )
+    assert apply_response.status_code == 409
+    blocked = apply_response.json()["detail"]["blocked_checks"]
+    assert blocked[0]["name"] == "task-session"
+
+
 def test_authoring_metadata_object_preview_and_apply(tmp_path: Path):
     project_id = f"metadata-authoring-api-{uuid4()}"
     (tmp_path / "metadata.xml").write_text(
@@ -2107,6 +2164,8 @@ def test_authoring_metadata_object_preview_and_apply(tmp_path: Path):
     client = TestClient(app)
     indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id})
     assert indexed.status_code == 200
+    create_authoring_session(client, project_id, "task.metadata", "session.metadata")
+    create_authoring_session(client, project_id, "task.metadata.rollback", "session.metadata.rollback")
 
     draft = {
         "object_kind": "DOCUMENT",
@@ -2136,6 +2195,7 @@ def test_authoring_metadata_object_preview_and_apply(tmp_path: Path):
     assert preview_payload["target"]["qualified_name"] == "Документ.ЗаявкаНаЗакупку"
     assert preview_payload["changed"] is True
     assert preview_payload["version_preview"]["apply_available"] is True
+    assert any(check["name"] == "task-session" and check["status"] == "OK" for check in preview_payload["checks"])
     assert any("Реквизит.Контрагент" in row["text"] for row in preview_payload["semantic_diff"])
     assert any("ТабличнаяЧасть.Товары" in row["text"] for row in preview_payload["semantic_diff"])
     assert any("Команда.Заполнить" in row["text"] for row in preview_payload["semantic_diff"])