Fix rabbit

docsgpt/api/user/agents/routes.py

@@ -1242,7 +1242,7 @@ class UpdateAgent(Resource):
                         if not normalized:
                             if workflow_required:
                                 return _reject("Workflow is required", user, field)
-                            update_fields["workflow_id"] = None
+                            pg_workflow_id = None
                         else:
                             # The agent runs its workflow as the owner, so it
                             # must be one of the owner's workflows.
@@ -1251,19 +1251,20 @@ class UpdateAgent(Resource):
                             )
                             if wf_err:
                                 return wf_err
-                            # Only the owner may point the agent at a different
-                            # workflow: editing rights on this agent extend to
-                            # the graph it uses, so swapping in the workflow of
-                            # another of the owner's agents would hand that
-                            # graph to the editor.
-                            current_workflow = existing_agent.get("workflow_id")
-                            if is_team_editor and pg_workflow_id != (
-                                str(current_workflow) if current_workflow else None
-                            ):
-                                return _denied(
-                                    AccessDenied(403, "Only the owner can change this agent's workflow")
-                                )
-                            update_fields["workflow_id"] = pg_workflow_id
+                        # Only the owner may change which workflow the agent
+                        # uses, detaching included: editing rights on this
+                        # agent extend to the graph it uses, so swapping in the
+                        # workflow of another of the owner's agents would hand
+                        # that graph to the editor. Editing the graph itself
+                        # goes through the workflow routes.
+                        current_workflow = existing_agent.get("workflow_id")
+                        if is_team_editor and pg_workflow_id != (
+                            str(current_workflow) if current_workflow else None
+                        ):
+                            return _denied(
+                                AccessDenied(403, "Only the owner can change this agent's workflow")
+                            )
+                        update_fields["workflow_id"] = pg_workflow_id
                     elif field == "prompt_id":
                         value = data["prompt_id"]
                         if not value or value == "default":

tests/api/user/test_attach_owner_refs.py

@@ -133,6 +133,25 @@ class TestAgentWorkflowSwap:
         assert _status(_put(app, pg_conn, agent_a, OWNER, {"workflow": wf_b})) == 200
         assert str(_row(pg_conn, agent_a)["workflow_id"]) == wf_b
 
+    def test_editor_cannot_detach_workflow(self, app, pg_conn):
+        # Unpublishing in the same request makes the empty workflow allowed
+        # for a draft; detaching is still the owner's call.
+        agent_a, wf_a, _ = self._two_workflow_agents(pg_conn)
+        resp = _put(app, pg_conn, agent_a, EDITOR, {"status": "draft", "workflow": ""})
+        assert _status(resp) == 403
+        assert str(_row(pg_conn, agent_a)["workflow_id"]) == wf_a
+
+    def test_owner_may_detach_workflow(self, app, pg_conn):
+        agent_a, _, _ = self._two_workflow_agents(pg_conn)
+        resp = _put(app, pg_conn, agent_a, OWNER, {"status": "draft", "workflow": ""})
+        assert _status(resp) == 200, resp.get_json()
+        assert _row(pg_conn, agent_a)["workflow_id"] is None
+
+    def test_editor_empty_workflow_on_agent_without_one_is_a_no_op(self, app, pg_conn):
+        agent_id, _ = _agent(pg_conn)
+        resp = _put(app, pg_conn, agent_id, EDITOR, {"status": "draft", "workflow": ""})
+        assert _status(resp) == 200, resp.get_json()
+
 
 class TestWorkflowNodeAttach:
     def _setup(self, pg_conn):