Fix task status re-entry and verify persisted transitions

src/backend/services/task_management_service.py

@@ -2166,15 +2166,21 @@ def update_task_status(
 
     # Update status text relation
     try:
-        upsert_text_for_concept(
-            subject_concept_id=task_concept_id,
+        _upsert_optional_task_text(
+            task_concept_id=task_concept_id,
             predicate=PREDICATE_HAS_TASK_STATUS,
-            text=status,
+            value=status,
             lang="en",
         )
     except Exception as e:
         raise TaskManagementError(f"Failed to update task status: {e}") from e
 
+    # A transition must replace the current value, including when returning to
+    # a previously used status. Do not emit completion effects for a stale read.
+    persisted_task = get_task(task_concept_id)
+    if persisted_task.get("status") != status:
+        raise TaskManagementError("Task status update did not match canonical read-back")
+
     # Update timestamp
     now = _now()
     ConceptsRepository.update_one(
@@ -4642,7 +4648,11 @@ def update_task_fields(
         changed_fields.append("current_work_product_concept_id")
 
     if "status" in fields:
-        update_task_status(task_concept_id, str(fields.get("status") or ""))
+        update_task_status(
+            task_concept_id,
+            str(fields.get("status") or ""),
+            actor_concept_id=actor_concept_id,
+        )
         changed_fields.append("status")
 
     assignee_field_present = "assignee_concept_id" in fields or "assignee_id" in fields

tests/backend/test_task_management_service.py

@@ -474,7 +474,7 @@ class TestUpdateTaskStatus:
     @patch("src.backend.services.task_management_service.get_task")
     @patch("src.backend.services.task_management_service.ConceptsRepository")
     @patch("src.backend.services.task_management_service.get_texts_for_concept")
-    @patch("src.backend.services.task_management_service.upsert_text_for_concept")
+    @patch("src.backend.services.task_management_service._upsert_optional_task_text")
     @patch(
         "src.backend.services.task_management_service.maybe_launch_task_status_workflow"
     )
@@ -529,7 +529,7 @@ def test_update_status_valid(
     @patch("src.backend.services.task_management_service.get_task")
     @patch("src.backend.services.task_management_service.ConceptsRepository")
     @patch("src.backend.services.task_management_service.get_texts_for_concept")
-    @patch("src.backend.services.task_management_service.upsert_text_for_concept")
+    @patch("src.backend.services.task_management_service._upsert_optional_task_text")
     @patch("src.backend.services.task_management_service.ensure_effort_unit_ontology")
     @patch(
         "src.backend.services.task_management_service.maybe_launch_effort_unit_completed_workflow"
@@ -598,7 +598,7 @@ def test_update_status_completed_persists_successor_linkage(
 
     @patch("src.backend.services.task_management_service.ConceptsRepository")
     @patch("src.backend.services.task_management_service.get_texts_for_concept")
-    @patch("src.backend.services.task_management_service.upsert_text_for_concept")
+    @patch("src.backend.services.task_management_service._upsert_optional_task_text")
     @patch(
         "src.backend.services.task_management_service.maybe_launch_task_status_workflow"
     )

tests/backend/test_task_status_reentry.py

@@ -0,0 +1,59 @@
+"""A current status must survive revisiting an earlier state."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.backend.services import task_management_service as tasks
+
+
+def test_reentering_status_replaces_old_value_and_emits_actual_transition(monkeypatch):
+    doc = {"concept_id": "#V#task_reentry", "relationships": {}, "metadata": {}}
+    values = ["pending", "in_progress", "blocked"]
+
+    def read(_):
+        return {"task_concept_id": doc["concept_id"], "status": values[-1]}
+
+    def additive(**kwargs):
+        if kwargs["text"] not in values:
+            values.append(kwargs["text"])
+
+    def singleton(**kwargs):
+        assert kwargs["garbage_collect"] is False
+        values[:] = [kwargs["text"]]
+
+    monkeypatch.setattr(tasks, "_get_task_doc", lambda _: (doc["concept_id"], doc))
+    monkeypatch.setattr(tasks, "_build_task_response", lambda _: read(None))
+    monkeypatch.setattr(tasks, "get_task", read)
+    monkeypatch.setattr(tasks, "upsert_text_for_concept", additive)
+    monkeypatch.setattr(tasks, "upsert_singleton_text_relation", singleton)
+    monkeypatch.setattr(tasks, "_clear_task_text_relations", MagicMock())
+    monkeypatch.setattr(tasks, "ConceptsRepository", MagicMock())
+    history = MagicMock()
+    launches = MagicMock(return_value={})
+    monkeypatch.setattr(tasks, "_append_task_history_event", history)
+    monkeypatch.setattr(tasks, "maybe_launch_task_status_workflow", launches)
+
+    for status in ("in_progress", "blocked", "in_progress"):
+        assert tasks.update_task_status(doc["concept_id"], status)["status"] == status
+        assert values == [status]
+    assert history.call_count == launches.call_count == 3
+    tasks.update_task_status(doc["concept_id"], "in_progress")
+    assert history.call_count == launches.call_count == 3
+
+
+def test_failed_status_readback_does_not_emit_completion(monkeypatch):
+    doc = {"concept_id": "#V#task_readback", "relationships": {}}
+    stale = {"status": "blocked"}
+    monkeypatch.setattr(tasks, "_get_task_doc", lambda _: (doc["concept_id"], doc))
+    monkeypatch.setattr(tasks, "_build_task_response", lambda _: stale)
+    monkeypatch.setattr(tasks, "get_task", lambda _: stale)
+    monkeypatch.setattr(tasks, "_upsert_optional_task_text", MagicMock())
+    repository = MagicMock()
+    completion = MagicMock()
+    monkeypatch.setattr(tasks, "ConceptsRepository", repository)
+    monkeypatch.setattr(tasks, "maybe_launch_effort_unit_completed_workflow", completion)
+    with pytest.raises(tasks.TaskManagementError, match="canonical read-back"):
+        tasks.update_task_status(doc["concept_id"], "completed")
+    repository.update_one.assert_not_called()
+    completion.assert_not_called()