feat: add direct oneliner follow-up actions

tests/test_main_agent_governance.py

@@ -51,6 +51,9 @@ class MainAgentGovernanceTests(unittest.TestCase):
         tables = [
             "job_events",
             "jobs",
+            "publish_reviews",
+            "live_recorder_bindings",
+            "live_recorder_sources",
             "agent_run_events",
             "agent_runs",
             "agent_policy_audit_logs",
@@ -61,6 +64,7 @@ class MainAgentGovernanceTests(unittest.TestCase):
             "agent_skills",
             "agent_memories",
             "platform_agent_profiles",
+            "oneliner_action_definitions",
             "tenant_usage_ledger",
             "tenant_quota_profiles",
             "admin_ops_audit_logs",
@@ -143,6 +147,74 @@ class MainAgentGovernanceTests(unittest.TestCase):
             "member_headers": {"Authorization": f"Bearer {member_token}"},
         }
 
+    def _insert_completed_job(
+        self,
+        *,
+        job_id: str = "job_completed",
+        title: str = "Completed Job",
+        result_json: str = '{"summary":"done"}',
+    ) -> str:
+        now = self.db_module.utc_now()
+        knowledge_base_id = "kb_member_default"
+        existing_kb = self.core.db.fetch_one("SELECT id FROM knowledge_bases WHERE id = ?", (knowledge_base_id,))
+        if not existing_kb:
+            self.core.db.execute(
+                """
+                INSERT INTO knowledge_bases (id, user_id, project_id, name, description, sync_status, created_at, updated_at)
+                VALUES (?, ?, ?, 'Default KB', '', 'ready', ?, ?)
+                """,
+                (knowledge_base_id, self.ctx["member_id"], self.ctx["project_id"], now, now),
+            )
+        self.core.db.execute(
+            """
+            INSERT INTO jobs (
+                id, user_id, project_id, parent_job_id, assistant_id, knowledge_base_id, content_source_id,
+                source_type, line_type, workflow_key, orchestrator, provider_name, provider_task_id,
+                source_url, title, language, status, transcript_text, style_summary, upload_status,
+                error, artifacts_json, result_json, analysis_model_profile_id, created_at, updated_at
+            ) VALUES (?, ?, ?, '', NULL, ?, NULL, ?, ?, ?, 'n8n', 'collector', '', '', ?, 'auto', 'completed', '', '', 'completed', '', '{}', ?, '', ?, ?)
+            """,
+            (
+                job_id,
+                self.ctx["member_id"],
+                self.ctx["project_id"],
+                knowledge_base_id,
+                "text",
+                "analysis",
+                "analysis_pipeline",
+                title,
+                result_json,
+                now,
+                now,
+            ),
+        )
+        return job_id
+
+    def _insert_assistant(
+        self,
+        *,
+        assistant_id: str = "asst_member_default",
+        name: str = "Default Assistant",
+    ) -> str:
+        now = self.db_module.utc_now()
+        self.core.db.execute(
+            """
+            INSERT INTO assistants (
+                id, user_id, project_id, name, description, system_prompt, generation_goal, config_json,
+                model_profile_id, created_at, updated_at
+            ) VALUES (?, ?, ?, ?, '', '', '', '{}', '', ?, ?)
+            """,
+            (
+                assistant_id,
+                self.ctx["member_id"],
+                self.ctx["project_id"],
+                name,
+                now,
+                now,
+            ),
+        )
+        return assistant_id
+
     def _seed_approved_member_without_project(self) -> dict[str, Any]:
         now = self.db_module.utc_now()
         admin_id = "acct_admin"
@@ -703,6 +775,61 @@ class MainAgentGovernanceTests(unittest.TestCase):
         self.assertIn("categories", usage_payload)
         self.assertIn("storage_bytes", usage_payload)
 
+    def test_direct_oneliner_actions_return_structured_followup_targets(self) -> None:
+        self._insert_completed_job(job_id="job_review_source", title="Review Source Job")
+        self._insert_assistant()
+
+        review_response = self.client.post(
+            "/v2/oneliner/actions/execute",
+            headers=self.ctx["member_headers"],
+            json={
+                "action_key": "review-draft",
+                "project_id": self.ctx["project_id"],
+                "platform": "douyin",
+                "payload": {},
+            },
+        )
+        self.assertEqual(review_response.status_code, 200, review_response.text)
+        review_payload = review_response.json()
+        self.assertEqual(review_payload["recommended_action"]["action"], "open-review-edit")
+        self.assertEqual(review_payload["recommended_action"]["screen"], "review")
+        self.assertEqual(review_payload["recommended_action"]["job_id"], "job_review_source")
+        self.assertTrue(review_payload["recommended_action"]["review_id"])
+
+        self_check_response = self.client.post(
+            "/v2/oneliner/actions/execute",
+            headers=self.ctx["member_headers"],
+            json={
+                "action_key": "platform-self-check",
+                "project_id": self.ctx["project_id"],
+                "platform": "douyin",
+                "payload": {},
+            },
+        )
+        self.assertEqual(self_check_response.status_code, 200, self_check_response.text)
+        self_check_payload = self_check_response.json()
+        self.assertEqual(self_check_payload["recommended_action"]["action"], "open-platform-agent-detail")
+        self.assertEqual(self_check_payload["recommended_action"]["screen"], "playbook")
+        self.assertEqual(self_check_payload["recommended_action"]["platform"], "douyin")
+
+        copy_response = self.client.post(
+            "/v2/oneliner/actions/execute",
+            headers=self.ctx["member_headers"],
+            json={
+                "action_key": "generate-copy",
+                "project_id": self.ctx["project_id"],
+                "platform": "douyin",
+                "payload": {
+                    "brief": "给我一版成交向短视频文案",
+                },
+            },
+        )
+        self.assertEqual(copy_response.status_code, 200, copy_response.text)
+        copy_payload = copy_response.json()
+        self.assertEqual(copy_payload["recommended_action"]["action"], "open-generate-copy")
+        self.assertEqual(copy_payload["recommended_action"]["screen"], "playbook")
+        self.assertEqual(copy_payload["recommended_action"]["platform"], "douyin")
+
     def test_platform_agent_routes_are_live(self) -> None:
         save_profile = self.client.put(
             "/v2/platform-agents/douyin/profile",