from __future__ import annotations import importlib import os import sys import tempfile import unittest from unittest.mock import AsyncMock, patch from pathlib import Path from typing import Any from fastapi.testclient import TestClient ROOT = Path(__file__).resolve().parents[1] APP_ROOT = ROOT / "collector-service" if str(APP_ROOT) not in sys.path: sys.path.insert(0, str(APP_ROOT)) class MainAgentGovernanceTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.tempdir = tempfile.TemporaryDirectory() temp_root = Path(cls.tempdir.name) os.environ["DATA_DIR"] = str(temp_root / "data") os.environ["DATABASE_PATH"] = str(temp_root / "data" / "storyforge.db") os.environ["DOWNLOADS_DIR"] = str(temp_root / "downloads") os.environ["JOBS_DIR"] = str(temp_root / "jobs") os.environ["MODELS_DIR"] = str(temp_root / "models") os.environ["ORCHESTRATOR_SHARED_SECRET"] = "test-secret" os.environ["WEB_AUTOLOGIN_ENABLED"] = "0" os.environ.setdefault("BOOTSTRAP_SUPERADMIN_USERNAME", "") os.environ.setdefault("BOOTSTRAP_SUPERADMIN_PASSWORD", "") cls.db_module = importlib.reload(importlib.import_module("app.database")) cls.oneliner = importlib.reload(importlib.import_module("app.oneliner_features")) cls.core = importlib.reload(importlib.import_module("app.core_main")) cls.app_main = importlib.reload(importlib.import_module("app.main")) cls.core.db.init_schema() cls.client = TestClient(cls.app_main.app) @classmethod def tearDownClass(cls) -> None: cls.client.close() cls.tempdir.cleanup() def setUp(self) -> None: self._clear_tables() self.ctx = self._seed_accounts() def _clear_tables(self) -> None: tables = [ "job_events", "jobs", "publish_reviews", "live_recorder_bindings", "live_recorder_sources", "agent_run_events", "agent_runs", "agent_policy_audit_logs", "agent_policy_effectivity", "agent_policy_versions", "agent_policy_scopes", "agent_skill_versions", "agent_skills", "agent_memories", "platform_agent_profiles", "oneliner_action_definitions", "tenant_usage_ledger", "tenant_quota_profiles", "admin_ops_audit_logs", "admin_ops_fix_runs", "admin_ops_incidents", "oneliner_messages", "oneliner_sessions", "oneliner_profiles", "auth_tokens", "assistants", "projects", "accounts", "model_profiles", ] with self.core.db.session() as conn: conn.execute("PRAGMA foreign_keys=OFF") for table in tables: try: conn.execute(f"DELETE FROM {table}") except Exception: continue conn.execute("PRAGMA foreign_keys=ON") def _seed_accounts(self) -> dict[str, Any]: now = self.db_module.utc_now() admin_id = "acct_admin" member_id = "acct_member" project_id = "proj_member" model_id = "model_default" admin_token = "token_admin" member_token = "token_member" self.core.db.execute( """ INSERT INTO accounts ( id, username, password_hash, password_salt, display_name, role, approval_status, approved_by, approved_at, preferred_analysis_model_id, created_at, updated_at ) VALUES (?, ?, 'hash', 'salt', ?, ?, 'approved', ?, ?, ?, ?, ?) """, (admin_id, "admin", "Admin", "super_admin", admin_id, now, model_id, now, now), ) self.core.db.execute( """ INSERT INTO accounts ( id, username, password_hash, password_salt, display_name, role, approval_status, approved_by, approved_at, preferred_analysis_model_id, created_at, updated_at ) VALUES (?, ?, 'hash', 'salt', ?, ?, 'approved', ?, ?, ?, ?, ?) """, (member_id, "member", "Member", "operator", admin_id, now, model_id, now, now), ) self.core.db.execute( """ INSERT INTO projects (id, user_id, name, description, created_at, updated_at) VALUES (?, ?, ?, '', ?, ?) """, (project_id, member_id, "Member Project", now, now), ) self.core.db.execute( """ INSERT INTO model_profiles ( id, owner_account_id, name, provider, base_url, api_key, model_name, is_system, is_default, created_at, updated_at ) VALUES (?, NULL, 'Default Model', 'openai_compat', 'http://127.0.0.1:8317/v1', '', 'GLM-5', 1, 1, ?, ?) """, (model_id, now, now), ) self.core.db.execute( "INSERT INTO auth_tokens (token, account_id, created_at) VALUES (?, ?, ?)", (admin_token, admin_id, now), ) self.core.db.execute( "INSERT INTO auth_tokens (token, account_id, created_at) VALUES (?, ?, ?)", (member_token, member_id, now), ) return { "admin_id": admin_id, "member_id": member_id, "project_id": project_id, "admin_headers": {"Authorization": f"Bearer {admin_token}"}, "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 = self._ensure_default_knowledge_base(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 _ensure_default_knowledge_base(self, now: str | None = None) -> str: knowledge_base_id = "kb_member_default" existing_kb = self.core.db.fetch_one("SELECT id FROM knowledge_bases WHERE id = ?", (knowledge_base_id,)) if existing_kb: return knowledge_base_id ts = now or self.db_module.utc_now() 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"], ts, ts), ) return knowledge_base_id def _insert_douyin_account( self, *, account_id: str, profile_url: str, nickname: str, ) -> str: now = self.db_module.utc_now() self.core.db.execute( """ INSERT INTO douyin_accounts ( id, user_id, profile_url, canonical_profile_url, sec_uid, douyin_uid, douyin_id, nickname, signature, avatar_url, tags_json, profile_stats_json, raw_profile_json, source_mode, sync_status, last_public_sync_at, last_creator_sync_at, last_analysis_at, created_at, updated_at ) VALUES (?, ?, ?, ?, '', '', '', ?, '', '', '[]', '{}', '{}', 'public', 'ready', NULL, NULL, NULL, ?, ?) """, (account_id, self.ctx["member_id"], profile_url, profile_url, nickname, now, now), ) return account_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 _insert_content_source_account( self, *, source_id: str = "source_member_platform", platform: str = "kuaishou", title: str = "平台账号", source_url: str = "https://example.com/account", ) -> str: self.core.create_content_source( account_id=self.ctx["member_id"], project_id=self.ctx["project_id"], source_kind="creator_account", platform=platform, handle=f"{platform}_handle", source_url=source_url, title=title, metadata={"nickname": title, "platform": platform}, ) created = self.core.db.fetch_one( """ SELECT * FROM content_sources WHERE user_id = ? AND project_id = ? AND platform = ? AND source_kind = 'creator_account' ORDER BY created_at DESC LIMIT 1 """, (self.ctx["member_id"], self.ctx["project_id"], platform), ) return created["id"] def _seed_approved_member_without_project(self) -> dict[str, Any]: now = self.db_module.utc_now() admin_id = "acct_admin" member_id = "acct_member_noproject" model_id = "model_default" admin_token = "token_admin" member_token = "token_member_noproject" self.core.db.execute( """ INSERT INTO accounts ( id, username, password_hash, password_salt, display_name, role, approval_status, approved_by, approved_at, preferred_analysis_model_id, created_at, updated_at ) VALUES (?, ?, 'hash', 'salt', ?, ?, 'approved', ?, ?, ?, ?, ?) """, (admin_id, "admin", "Admin", "super_admin", admin_id, now, model_id, now, now), ) self.core.db.execute( """ INSERT INTO accounts ( id, username, password_hash, password_salt, display_name, role, approval_status, approved_by, approved_at, preferred_analysis_model_id, created_at, updated_at ) VALUES (?, ?, 'hash', 'salt', ?, ?, 'approved', ?, ?, ?, ?, ?) """, (member_id, "member_noproject", "Member No Project", "operator", admin_id, now, model_id, now, now), ) self.core.db.execute( """ INSERT INTO model_profiles ( id, owner_account_id, name, provider, base_url, api_key, model_name, is_system, is_default, created_at, updated_at ) VALUES (?, NULL, 'Default Model', 'openai_compat', 'http://127.0.0.1:8317/v1', '', 'GLM-5', 1, 1, ?, ?) """, (model_id, now, now), ) self.core.db.execute( "INSERT INTO auth_tokens (token, account_id, created_at) VALUES (?, ?, ?)", (admin_token, admin_id, now), ) self.core.db.execute( "INSERT INTO auth_tokens (token, account_id, created_at) VALUES (?, ?, ?)", (member_token, member_id, now), ) return { "admin_id": admin_id, "member_id": member_id, "admin_headers": {"Authorization": f"Bearer {admin_token}"}, "member_headers": {"Authorization": f"Bearer {member_token}"}, } def test_agent_run_creation_snapshots_governance_and_needs_confirmation(self) -> None: response = self.client.post( "/v2/oneliner/runs", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "source_screen": "dashboard", "source_action_key": "homepage-primary-action", "title": "跟进重点账号", "summary": "先由主 Agent 评估优先级", "intent_key": "track_account", "platform": "douyin", "platform_scope": "single_platform", "plan_request": { "goal": "跟进重点账号", "steps": ["读取当前项目上下文", "检查重点账号变化", "决定下一步"], }, }, ) self.assertEqual(response.status_code, 200, response.text) payload = response.json() self.assertEqual(payload["run_status"], "needs_confirmation") self.assertEqual(payload["source_screen"], "dashboard") self.assertEqual(payload["platform"], "douyin") self.assertEqual(payload["platform_scope"], "single_platform") self.assertEqual(payload["session_id"][:5], "oline") self.assertEqual(payload["plan"]["goal"], "跟进重点账号") self.assertEqual(payload["recommended_preview_action"]["action"], "goto-discovery") self.assertEqual(payload["recommended_preview_action"]["screen"], "discovery") self.assertEqual(payload["governance"]["project_id"], self.ctx["project_id"]) self.assertIn("layers", payload["governance"]) self.assertIn("oneliner_profile", payload["governance"]) self.assertIn("oneliner_profile_version", payload["governance"]) self.assertGreaterEqual(payload["governance"]["oneliner_profile_version"]["version_no"], 1) self.assertIn("platform_agent_profile", payload["governance"]) self.assertEqual(payload["governance"]["platform_agent_profile"]["platform"], "douyin") self.assertEqual(payload["events"][0]["event_type"], "run.created") def test_agent_run_confirm_transitions_to_queue_or_running_and_logs_events(self) -> None: create = self.client.post( "/v2/oneliner/runs", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "source_screen": "strategy", "source_action_key": "handoff-to-main-agent", "title": "调整当前平台策略", "summary": "让主 Agent 先给执行计划", "intent_key": "custom", "platform": "douyin", "platform_scope": "single_platform", "plan_request": { "goal": "调整当前平台策略", "steps": ["读取当前平台策略", "生成调整建议"], }, }, ) self.assertEqual(create.status_code, 200, create.text) run_id = create.json()["id"] confirm = self.client.post( f"/v2/oneliner/runs/{run_id}/confirm", headers=self.ctx["member_headers"], json={"reason": "user confirmed"}, ) self.assertEqual(confirm.status_code, 200, confirm.text) payload = confirm.json() self.assertIn(payload["run_status"], {"queued", "running"}) self.assertEqual(payload["recommended_preview_action"]["action"], "goto-strategy") self.assertEqual(payload["recommended_preview_action"]["screen"], "strategy") event_types = [item["event_type"] for item in payload["events"]] self.assertIn("run.created", event_types) self.assertIn("run.confirmed", event_types) self.assertTrue("run.queued" in event_types or "run.started" in event_types) def test_running_agent_run_detail_advances_to_progress_and_done(self) -> None: create = self.client.post( "/v2/oneliner/runs", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "source_screen": "dashboard", "source_action_key": "homepage-primary-action", "title": "安排今日动作", "summary": "让主 Agent 给出执行收口", "intent_key": "custom", "platform": "douyin", "platform_scope": "single_platform", "plan_request": { "goal": "安排今日动作", "steps": ["读取当前项目上下文", "给出执行建议", "输出下一步"], }, }, ) self.assertEqual(create.status_code, 200, create.text) run_id = create.json()["id"] confirm = self.client.post( f"/v2/oneliner/runs/{run_id}/confirm", headers=self.ctx["member_headers"], json={"reason": "user confirmed"}, ) self.assertEqual(confirm.status_code, 200, confirm.text) detail = self.client.get( f"/v2/oneliner/runs/{run_id}", headers=self.ctx["member_headers"], ) self.assertEqual(detail.status_code, 200, detail.text) payload = detail.json() self.assertEqual(payload["run_status"], "done") self.assertTrue(payload["finished_at"]) self.assertEqual(payload["result"]["result_kind"], "main_agent_plan") self.assertEqual(payload["result"]["recommended_action"]["action"], "goto-discovery") self.assertEqual(payload["result"]["recommended_action"]["screen"], "discovery") self.assertEqual(payload["result"]["result_sections"]["workstream_key"], "discovery") self.assertGreaterEqual(len(payload["result"]["result_sections"]["cards"]), 2) self.assertEqual(payload["result"]["result_sections"]["cards"][0]["title"], "当前焦点") self.assertGreaterEqual(payload["result"]["execution_card"]["oneliner_profile_version"]["version_no"], 1) self.assertEqual(payload["result"]["execution_card"]["platform_agent_profile"]["platform"], "douyin") event_types = [item["event_type"] for item in payload["events"]] self.assertIn("run.progress", event_types) self.assertIn("run.done", event_types) def test_agent_run_result_recommended_action_preserves_requested_object_context(self) -> None: run_response = self.client.post( "/v2/oneliner/runs", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "source_screen": "discovery", "source_action_key": "selected-account-handoff", "title": "继续分析当前对象", "summary": "让主 Agent 沿着当前账号继续推进。", "intent_key": "analyze_account", "platform": "douyin", "platform_scope": "single_platform", "plan_request": { "goal": "继续分析当前对象", "steps": ["读取当前账号上下文", "结合当前策略生成下一步建议"], }, "payload": { "target_account_id": "acct_focus_target", }, }, ) self.assertEqual(run_response.status_code, 200, run_response.text) run_id = run_response.json()["id"] confirm = self.client.post( f"/v2/oneliner/runs/{run_id}/confirm", headers=self.ctx["member_headers"], json={"reason": "继续围绕当前对象推进"}, ) self.assertEqual(confirm.status_code, 200, confirm.text) detail = self.client.get( f"/v2/oneliner/runs/{run_id}", headers=self.ctx["member_headers"], ) self.assertEqual(detail.status_code, 200, detail.text) payload = detail.json() self.assertEqual(payload["run_status"], "done") self.assertEqual(payload["result"]["recommended_action"]["action"], "select-account") self.assertEqual(payload["result"]["recommended_action"]["screen"], "discovery") self.assertEqual(payload["result"]["recommended_action"]["account_id"], "acct_focus_target") def test_cancelled_run_can_be_retried_as_a_new_pending_run(self) -> None: create = self.client.post( "/v2/oneliner/runs", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "source_screen": "production", "source_action_key": "handoff-to-main-agent", "title": "恢复失败任务", "summary": "让主 Agent 重新接单", "intent_key": "custom", "platform": "douyin", "platform_scope": "single_platform", "plan_request": { "goal": "恢复失败任务", "steps": ["检查失败原因", "重新安排执行", "确认落点"], }, }, ) self.assertEqual(create.status_code, 200, create.text) original_run = create.json() run_id = original_run["id"] cancel = self.client.post( f"/v2/oneliner/runs/{run_id}/cancel", headers=self.ctx["member_headers"], json={"reason": "user cancelled"}, ) self.assertEqual(cancel.status_code, 200, cancel.text) self.assertEqual(cancel.json()["run_status"], "cancelled") retry = self.client.post( f"/v2/oneliner/runs/{run_id}/retry", headers=self.ctx["member_headers"], json={"reason": "retry from runtime"}, ) self.assertEqual(retry.status_code, 200, retry.text) payload = retry.json() self.assertNotEqual(payload["id"], run_id) self.assertEqual(payload["run_status"], "needs_confirmation") self.assertEqual(payload["title"], original_run["title"]) self.assertEqual(payload["project_id"], self.ctx["project_id"]) self.assertEqual(payload["plan"]["goal"], "恢复失败任务") self.assertEqual(payload["recommended_preview_action"]["screen"], "production") event_types = [item["event_type"] for item in payload["events"]] self.assertIn("run.created", event_types) self.assertIn("run.retried", event_types) def test_effective_policy_merges_system_user_global_and_platform_layers(self) -> None: system_response = self.client.put( "/v2/admin/oneliner/governance/system/main-agent", headers=self.ctx["admin_headers"], json={ "title": "System main agent", "summary": "Default baseline", "policy": { "tone": {"style": "default"}, "homepage": {"focus": "ops"}, "actions": {"max_cards": 3}, }, "reason": "seed system baseline", }, ) self.assertEqual(system_response.status_code, 200, system_response.text) global_response = self.client.put( "/v2/oneliner/governance/user/global", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "title": "Member global strategy", "summary": "Personal operating style", "policy": { "tone": {"style": "analytical"}, "memory": {"default_window": "30d"}, "actions": {"max_cards": 2}, }, "reason": "personalize global defaults", }, ) self.assertEqual(global_response.status_code, 200, global_response.text) platform_response = self.client.put( "/v2/oneliner/governance/user/platforms/douyin", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "title": "Douyin strategy", "summary": "Tighter benchmark workflow", "policy": { "actions": {"max_cards": 1}, "douyin": {"benchmark_mode": "strict"}, }, "reason": "tighten douyin execution", }, ) self.assertEqual(platform_response.status_code, 200, platform_response.text) effective_response = self.client.get( "/v2/oneliner/governance/effective", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"], "platform": "douyin"}, ) self.assertEqual(effective_response.status_code, 200, effective_response.text) payload = effective_response.json() self.assertEqual( [item["scope_kind"] for item in payload["layers"]], ["system_main", "user_global", "user_platform"], ) self.assertEqual(payload["effective_policy"]["tone"]["style"], "analytical") self.assertEqual(payload["effective_policy"]["homepage"]["focus"], "ops") self.assertEqual(payload["effective_policy"]["actions"]["max_cards"], 1) self.assertEqual(payload["effective_policy"]["douyin"]["benchmark_mode"], "strict") def test_admin_override_takes_precedence_in_effective_policy(self) -> None: self.client.put( "/v2/admin/oneliner/governance/system/main-agent", headers=self.ctx["admin_headers"], json={ "title": "System main agent", "policy": {"actions": {"max_cards": 3}}, "reason": "seed baseline", }, ) self.client.put( "/v2/oneliner/governance/user/platforms/douyin", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "title": "Douyin strategy", "policy": {"actions": {"max_cards": 1}}, "reason": "tighten douyin execution", }, ) override_response = self.client.post( "/v2/admin/oneliner/governance/overrides", headers=self.ctx["admin_headers"], json={ "target_user_id": self.ctx["member_id"], "target_project_id": self.ctx["project_id"], "platform": "douyin", "title": "Safety override", "summary": "Require review after recent drift", "policy": { "actions": {"max_cards": 5}, "guardrails": {"require_admin_review": True}, }, "reason": "contain unexpected drift", }, ) self.assertEqual(override_response.status_code, 200, override_response.text) effective_response = self.client.get( "/v2/oneliner/governance/effective", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"], "platform": "douyin"}, ) self.assertEqual(effective_response.status_code, 200, effective_response.text) payload = effective_response.json() self.assertEqual(payload["layers"][-1]["scope_kind"], "admin_override") self.assertEqual(payload["effective_policy"]["actions"]["max_cards"], 5) self.assertTrue(payload["effective_policy"]["guardrails"]["require_admin_review"]) def test_admin_override_without_target_project_applies_to_member_projects(self) -> None: override_response = self.client.post( "/v2/admin/oneliner/governance/overrides", headers=self.ctx["admin_headers"], json={ "target_user_id": self.ctx["member_id"], "title": "Global safety override", "summary": "Apply guardrails across every project", "policy": { "guardrails": {"require_admin_review": True}, "actions": {"max_cards": 4}, }, "reason": "global containment", }, ) self.assertEqual(override_response.status_code, 200, override_response.text) effective_response = self.client.get( "/v2/oneliner/governance/effective", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"], "platform": "douyin"}, ) self.assertEqual(effective_response.status_code, 200, effective_response.text) payload = effective_response.json() self.assertEqual(payload["layers"][-1]["scope_kind"], "admin_override") self.assertEqual(payload["effective_policy"]["actions"]["max_cards"], 4) self.assertTrue(payload["effective_policy"]["guardrails"]["require_admin_review"]) self.assertEqual(payload["active_admin_override_notice"]["title"], "Global safety override") def test_effective_policy_skips_future_scheduled_versions_until_window_opens(self) -> None: first_response = self.client.put( "/v2/admin/oneliner/governance/system/main-agent", headers=self.ctx["admin_headers"], json={ "title": "Current system baseline", "summary": "Active now", "policy": {"tone": {"style": "default"}}, "reason": "baseline", }, ) self.assertEqual(first_response.status_code, 200, first_response.text) second_response = self.client.put( "/v2/admin/oneliner/governance/system/main-agent", headers=self.ctx["admin_headers"], json={ "title": "Future strategy", "summary": "Should not be active yet", "policy": {"tone": {"style": "future"}}, "effect_mode": "scheduled", "starts_at": "2099-01-01T00:00:00Z", "reason": "future rollout", }, ) self.assertEqual(second_response.status_code, 200, second_response.text) effective_response = self.client.get( "/v2/oneliner/governance/effective", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"], "platform": "douyin"}, ) self.assertEqual(effective_response.status_code, 200, effective_response.text) payload = effective_response.json() self.assertEqual(payload["effective_policy"]["tone"]["style"], "default") self.assertEqual(payload["layers"][0]["current_version"]["title"], "Current system baseline") def test_scope_read_endpoints_keep_current_version_on_active_release_not_future_schedule(self) -> None: user_first = self.client.put( "/v2/oneliner/governance/user/global", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "title": "User global baseline", "policy": {"tone": {"style": "baseline"}}, "reason": "seed baseline", }, ) self.assertEqual(user_first.status_code, 200, user_first.text) user_future = self.client.put( "/v2/oneliner/governance/user/global", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "title": "User global future", "policy": {"tone": {"style": "future"}}, "effect_mode": "scheduled", "starts_at": "2099-01-01T00:00:00Z", "reason": "future rollout", }, ) self.assertEqual(user_future.status_code, 200, user_future.text) user_read = self.client.get( "/v2/oneliner/governance/user/global", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(user_read.status_code, 200, user_read.text) self.assertEqual(user_read.json()["current_version"]["title"], "User global baseline") system_first = self.client.put( "/v2/admin/oneliner/governance/system/main-agent", headers=self.ctx["admin_headers"], json={ "title": "System baseline", "policy": {"homepage": {"focus": "stable"}}, "reason": "stable baseline", }, ) self.assertEqual(system_first.status_code, 200, system_first.text) system_future = self.client.put( "/v2/admin/oneliner/governance/system/main-agent", headers=self.ctx["admin_headers"], json={ "title": "System future", "policy": {"homepage": {"focus": "future"}}, "effect_mode": "scheduled", "starts_at": "2099-01-01T00:00:00Z", "reason": "future rollout", }, ) self.assertEqual(system_future.status_code, 200, system_future.text) system_read = self.client.get( "/v2/admin/oneliner/governance/system/main-agent", headers=self.ctx["admin_headers"], ) self.assertEqual(system_read.status_code, 200, system_read.text) self.assertEqual(system_read.json()["current_version"]["title"], "System baseline") def test_governance_read_endpoints_do_not_create_default_project_when_project_is_missing(self) -> None: self._clear_tables() ctx = self._seed_approved_member_without_project() before_count = self.core.db.fetch_one("SELECT COUNT(*) AS count FROM projects WHERE user_id = ?", (ctx["member_id"],)) self.assertEqual(int((before_count or {}).get("count") or 0), 0) effective_response = self.client.get( "/v2/oneliner/governance/effective", headers=ctx["member_headers"], params={"platform": "douyin"}, ) self.assertEqual(effective_response.status_code, 200, effective_response.text) effective_payload = effective_response.json() self.assertEqual(effective_payload["project_id"], "") self.assertEqual(effective_payload["layers"], []) global_response = self.client.get( "/v2/oneliner/governance/user/global", headers=ctx["member_headers"], ) self.assertEqual(global_response.status_code, 200, global_response.text) self.assertEqual(global_response.json()["scope"]["subject_project_id"], "") self.assertIsNone(global_response.json()["current_version"]) after_count = self.core.db.fetch_one("SELECT COUNT(*) AS count FROM projects WHERE user_id = ?", (ctx["member_id"],)) self.assertEqual(int((after_count or {}).get("count") or 0), 0) def test_action_registry_and_tenant_quota_routes_are_live(self) -> None: registry_response = self.client.get( "/v2/oneliner/action-registry", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(registry_response.status_code, 200, registry_response.text) registry_payload = registry_response.json() self.assertGreater(registry_payload["count"], 0) default_action = next(item for item in registry_payload["items"] if item["action_key"] == "generate-copy") self.assertEqual(default_action["source"], "default") action_keys = {item["action_key"] for item in registry_payload["items"]} self.assertIn("analyze-account", action_keys) self.assertIn("track-account", action_keys) self.assertIn("refresh-tracking", action_keys) self.assertIn("mark-tracking-read", action_keys) self.assertIn("create-assistant", action_keys) self.assertIn("search-similar-accounts", action_keys) self.assertIn("save-benchmark-link", action_keys) save_registry = self.client.put( "/v2/oneliner/action-registry/generate-copy", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, json={ "label": "生成成交文案", "description": "直接拉起一版更偏成交的文案生成。", "category": "content", "status": "disabled", "config": {"tone": "sales"}, }, ) self.assertEqual(save_registry.status_code, 200, save_registry.text) saved_registry = save_registry.json() self.assertEqual(saved_registry["label"], "生成成交文案") self.assertEqual(saved_registry["status"], "disabled") self.assertEqual(saved_registry["config"]["tone"], "sales") self.assertEqual(saved_registry["source"], "override") growth_preset = { "monthly_budget_cents": 49900, "storage_limit_bytes": 21474836480, "analysis_quota": 160, "copy_quota": 320, "ai_video_quota": 12, "real_cut_quota": 8, "recorder_quota": 20, } quota_response = self.client.put( "/v2/tenant/quota", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, json={ "enabled": True, "monthly_budget_cents": 1, "storage_limit_bytes": 2, "analysis_quota": 3, "copy_quota": 4, "ai_video_quota": 5, "real_cut_quota": 6, "recorder_quota": 7, "config": {"package_label": "growth", "warn_threshold": 0.8, "custom_note": "keep"}, }, ) self.assertEqual(quota_response.status_code, 200, quota_response.text) quota_payload = quota_response.json() self.assertEqual(quota_payload["package_label"], "growth") self.assertEqual(quota_payload["monthly_budget_cents"], growth_preset["monthly_budget_cents"]) self.assertEqual(quota_payload["storage_limit_bytes"], growth_preset["storage_limit_bytes"]) self.assertEqual(quota_payload["analysis_quota"], growth_preset["analysis_quota"]) self.assertEqual(quota_payload["copy_quota"], growth_preset["copy_quota"]) self.assertEqual(quota_payload["ai_video_quota"], growth_preset["ai_video_quota"]) self.assertEqual(quota_payload["real_cut_quota"], growth_preset["real_cut_quota"]) self.assertEqual(quota_payload["recorder_quota"], growth_preset["recorder_quota"]) self.assertEqual(quota_payload["config"]["package_label"], "growth") self.assertEqual(quota_payload["config"]["warn_threshold"], 0.8) self.assertEqual(quota_payload["config"]["package_title"], "增长套餐") self.assertTrue(quota_payload["config"]["package_is_preset"]) self.assertEqual(quota_payload["config"]["custom_note"], "keep") quota_get_response = self.client.get( "/v2/tenant/quota", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(quota_get_response.status_code, 200, quota_get_response.text) quota_get_payload = quota_get_response.json() self.assertEqual(quota_get_payload["package_label"], "growth") self.assertEqual(quota_get_payload["monthly_budget_cents"], growth_preset["monthly_budget_cents"]) self.assertEqual(quota_get_payload["config"]["package_label"], "growth") self.assertEqual(quota_get_payload["config"]["warn_threshold"], 0.8) self.assertEqual(quota_get_payload["config"]["package_title"], "增长套餐") custom_response = self.client.put( "/v2/tenant/quota", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, json={ "enabled": False, "monthly_budget_cents": 9100, "storage_limit_bytes": 111, "analysis_quota": 22, "copy_quota": 33, "ai_video_quota": 44, "real_cut_quota": 55, "recorder_quota": 66, "config": {"package_label": "custom", "warn_threshold": 0.55, "custom_note": "manual"}, }, ) self.assertEqual(custom_response.status_code, 200, custom_response.text) custom_payload = custom_response.json() self.assertEqual(custom_payload["package_label"], "custom") self.assertEqual(custom_payload["monthly_budget_cents"], 9100) self.assertEqual(custom_payload["storage_limit_bytes"], 111) self.assertEqual(custom_payload["analysis_quota"], 22) self.assertEqual(custom_payload["copy_quota"], 33) self.assertEqual(custom_payload["ai_video_quota"], 44) self.assertEqual(custom_payload["real_cut_quota"], 55) self.assertEqual(custom_payload["recorder_quota"], 66) self.assertFalse(custom_payload["enabled"]) self.assertEqual(custom_payload["config"]["package_label"], "custom") self.assertEqual(custom_payload["config"]["warn_threshold"], 0.55) self.assertEqual(custom_payload["config"]["package_title"], "自定义套餐") self.assertFalse(custom_payload["config"]["package_is_preset"]) self.assertEqual(custom_payload["config"]["custom_note"], "manual") usage_response = self.client.get( "/v2/tenant/usage", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(usage_response.status_code, 200, usage_response.text) usage_payload = usage_response.json() self.assertIn("categories", usage_payload) self.assertIn("storage_bytes", usage_payload) def test_high_frequency_intent_actions_prefer_direct_execute_entrypoints(self) -> None: expected_actions = { "create_assistant": "direct-create-assistant", "import_homepage": "direct-import-selected-account", "track_account": "direct-track-selected-account", "generate_copy": "direct-generate-copy", "ai_video": "direct-create-ai-video", "real_cut": "direct-create-real-cut", } for intent_key, expected_action_key in expected_actions.items(): with self.subTest(intent_key=intent_key): actions = self.oneliner.INTENT_ACTIONS[intent_key] self.assertTrue(actions) self.assertEqual(actions[0]["key"], expected_action_key) self.assertEqual(actions[0]["kind"], "ui_action") 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"]) explicit_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": { "source_job_id": "job_review_source", }, }, ) self.assertEqual(explicit_review_response.status_code, 200, explicit_review_response.text) explicit_review_payload = explicit_review_response.json() self.assertEqual(explicit_review_payload["recommended_action"]["action"], "open-review-edit") self.assertEqual(explicit_review_payload["recommended_action"]["job_id"], "job_review_source") 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-generated-copy-results") self.assertEqual(copy_payload["recommended_action"]["screen"], "playbook") self.assertEqual(copy_payload["recommended_action"]["platform"], "douyin") with patch.object( self.core, "create_content_source_sync_job", new=AsyncMock(return_value={"id": "job_sync_import", "title": "Import Sync Job"}), ): import_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "import-homepage", "project_id": self.ctx["project_id"], "platform": "douyin", "payload": {"source_url": "https://www.douyin.com/user/test-homepage"}, }, ) self.assertEqual(import_response.status_code, 200, import_response.text) import_payload = import_response.json() self.assertEqual(import_payload["recommended_action"]["action"], "open-job-detail") self.assertEqual(import_payload["recommended_action"]["screen"], "production") self.assertEqual(import_payload["recommended_action"]["job_id"], "job_sync_import") source_id = self._insert_content_source_account( platform="kuaishou", title="高分拆解账号", source_url="https://www.kuaishou.com/profile/top-video-account", ) video_source = self.core.create_content_source( account_id=self.ctx["member_id"], project_id=self.ctx["project_id"], source_kind="video_link", platform="kuaishou", source_url="https://www.kuaishou.com/video/top-video-1", title="高分作品 1", metadata={ "origin_content_source_id": source_id, "source_account_url": "https://www.kuaishou.com/profile/top-video-account", }, ) now = self.db_module.utc_now() knowledge_base_id = self._ensure_default_knowledge_base(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, ?, ?, ?, ?, ?, 'n8n', 'collector', '', '', ?, 'auto', 'completed', '', '', 'completed', '', '{}', ?, '', ?, ?) """, ( "job_video_1", self.ctx["member_id"], self.ctx["project_id"], knowledge_base_id, video_source["id"], "video_link", "analysis", "analysis_pipeline", "高分作品 1", '{"performance_score":91,"summary":"高分作品摘要"}', now, now, ), ) with patch.object(self.core, "call_model", new=AsyncMock(return_value='{"summary":"保留开头 3 秒抓人结构"}')): analyze_top_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "analyze-top-videos", "project_id": self.ctx["project_id"], "platform": "kuaishou", "payload": { "target_account_id": source_id, "top_video_count": 1, "min_score": 0, }, }, ) self.assertEqual(analyze_top_response.status_code, 200, analyze_top_response.text) analyze_top_payload = analyze_top_response.json() self.assertEqual(analyze_top_payload["recommended_action"]["action"], "select-account") self.assertEqual(analyze_top_payload["recommended_action"]["screen"], "discovery") self.assertEqual(analyze_top_payload["recommended_action"]["account_id"], source_id) def test_create_ai_video_action_passes_provider_and_model_through_oneliner(self) -> None: self._insert_completed_job(job_id="job_ai_video_source", title="AI Video Source Job") self._insert_assistant() captured_request: dict[str, Any] = {} async def fake_create_ai_video_job(request: Any, account: dict[str, Any]) -> dict[str, Any]: captured_request.update( { "project_id": request.project_id, "assistant_id": request.assistant_id, "source_job_id": request.source_job_id, "title": request.title, "brief": request.brief, "style": request.style, "shots": request.shots, "duration": request.duration, "video_provider": request.video_provider, "video_model": request.video_model, } ) return { "id": "job_ai_video_seedance", "title": request.title, "artifacts": { "source_job_id": request.source_job_id, "video_provider": request.video_provider, "video_model": request.video_model, }, } with patch.object(self.core, "create_ai_video_job", new=AsyncMock(side_effect=fake_create_ai_video_job)): response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "create-ai-video", "project_id": self.ctx["project_id"], "platform": "douyin", "payload": { "source_job_id": "job_ai_video_source", "title": "Seedance 2.0 测试任务", "brief": "做一条节奏强、镜头推进明确的短视频。", "style": "cinematic", "shots": 5, "duration": 6, "video_provider": "seedance2", "video_model": "seedance-2.0-pro", }, }, ) self.assertEqual(response.status_code, 200, response.text) payload = response.json() self.assertEqual(captured_request["source_job_id"], "job_ai_video_source") self.assertEqual(captured_request["video_provider"], "seedance2") self.assertEqual(captured_request["video_model"], "seedance-2.0-pro") self.assertEqual(payload["payload"]["job"]["artifacts"]["video_provider"], "seedance2") def test_import_video_link_and_text_actions_execute_through_oneliner(self) -> None: self._insert_assistant() captured_video_request: dict[str, Any] = {} captured_text_request: dict[str, Any] = {} async def fake_video_link_job(request: Any, account: dict[str, Any]) -> dict[str, Any]: captured_video_request.update( { "project_id": request.project_id, "knowledge_base_id": request.knowledge_base_id, "assistant_id": request.assistant_id, "title": request.title, "video_url": request.video_url, "language": request.language, } ) return {"id": "job_video_link_direct", "title": request.title or "视频链接分析任务"} async def fake_text_job(request: Any, account: dict[str, Any]) -> dict[str, Any]: captured_text_request.update( { "project_id": request.project_id, "knowledge_base_id": request.knowledge_base_id, "assistant_id": request.assistant_id, "title": request.title, "content": request.content, } ) return {"id": "job_text_direct", "title": request.title} with patch.object(self.core, "create_video_link_job", new=AsyncMock(side_effect=fake_video_link_job)): video_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "import-video-link", "project_id": self.ctx["project_id"], "platform": "douyin", "payload": { "title": "短视频链接导入", "video_url": "https://example.com/video/123", "assistant_id": "asst_member_default", "language": "zh-CN", }, }, ) self.assertEqual(video_response.status_code, 200, video_response.text) video_payload = video_response.json() self.assertEqual(captured_video_request["video_url"], "https://example.com/video/123") self.assertEqual(captured_video_request["assistant_id"], "asst_member_default") self.assertEqual(video_payload["recommended_action"]["action"], "open-job-detail") self.assertEqual(video_payload["recommended_action"]["job_id"], "job_video_link_direct") with patch.object(self.core, "create_text_job", new=AsyncMock(side_effect=fake_text_job)): text_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "import-text", "project_id": self.ctx["project_id"], "platform": "douyin", "payload": { "title": "文本导入", "content": "这是一段需要分析的文本。", "assistant_id": "asst_member_default", }, }, ) self.assertEqual(text_response.status_code, 200, text_response.text) text_payload = text_response.json() self.assertEqual(captured_text_request["title"], "文本导入") self.assertEqual(captured_text_request["assistant_id"], "asst_member_default") self.assertEqual(text_payload["recommended_action"]["action"], "open-job-detail") self.assertEqual(text_payload["recommended_action"]["job_id"], "job_text_direct") def test_direct_oneliner_actions_execute_real_account_and_agent_flows(self) -> None: source_id = self._insert_content_source_account( platform="kuaishou", title="快手测试账号", source_url="https://www.kuaishou.com/profile/test-account", ) candidate_id = self._insert_content_source_account( platform="kuaishou", title="快手候选账号", source_url="https://www.kuaishou.com/profile/candidate-account", ) captured_model_calls: list[dict[str, Any]] = [] async def fake_call_model(profile: dict[str, Any], *, system_prompt: str, user_prompt: str, temperature: float = 0.3, **_: Any) -> str: captured_model_calls.append( { "profile_id": profile["id"], "system_prompt": system_prompt, "user_prompt": user_prompt, "temperature": temperature, } ) return ( '{"executive_summary":"账号可继续做增长拆解","borrow_points":["封面结构稳定"],' '"risks":["更新频率偏低"],"next_actions":["继续同步并补日报"]}' ) with patch.object(self.core, "call_model", new=AsyncMock(side_effect=fake_call_model)): analyze_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "analyze-account", "project_id": self.ctx["project_id"], "platform": "kuaishou", "payload": { "target_account_id": source_id, "extra_focus": "重点看商业化动作", "max_videos": 4, }, }, ) self.assertEqual(analyze_response.status_code, 200, analyze_response.text) analyze_payload = analyze_response.json() self.assertEqual(analyze_payload["recommended_action"]["action"], "select-account") self.assertEqual(analyze_payload["recommended_action"]["screen"], "discovery") self.assertEqual(analyze_payload["recommended_action"]["account_id"], source_id) self.assertEqual(analyze_payload["payload"]["platform"], "kuaishou") self.assertEqual(analyze_payload["payload"]["account_id"], source_id) self.assertTrue(captured_model_calls) self.assertIn("重点看商业化动作", captured_model_calls[0]["user_prompt"]) track_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "track-account", "project_id": self.ctx["project_id"], "platform": "kuaishou", "payload": { "target_account_id": source_id, "note": "由主 Agent 直接加入跟踪", "refresh_now": True, }, }, ) self.assertEqual(track_response.status_code, 200, track_response.text) track_payload = track_response.json() self.assertEqual(track_payload["recommended_action"]["action"], "open-job-detail") self.assertEqual(track_payload["recommended_action"]["screen"], "production") self.assertTrue(track_payload["recommended_action"]["job_id"]) self.assertEqual(track_payload["payload"]["platform"], "kuaishou") self.assertEqual(track_payload["payload"]["account_id"], source_id) self.assertEqual(track_payload["payload"]["refresh"]["tracked_account_id"], source_id) self.assertTrue(track_payload["payload"]["refresh"]["sync_job_id"]) refresh_tracking_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "refresh-tracking", "project_id": self.ctx["project_id"], "platform": "kuaishou", "payload": {}, }, ) self.assertEqual(refresh_tracking_response.status_code, 200, refresh_tracking_response.text) refresh_tracking_payload = refresh_tracking_response.json() self.assertEqual(refresh_tracking_payload["recommended_action"]["action"], "open-job-detail") self.assertEqual(refresh_tracking_payload["recommended_action"]["screen"], "production") self.assertTrue(refresh_tracking_payload["recommended_action"]["job_id"]) self.assertEqual(refresh_tracking_payload["payload"]["platform"], "kuaishou") self.assertGreaterEqual(int(refresh_tracking_payload["payload"]["refresh"]["refreshed"] or 0), 1) mark_read_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "mark-tracking-read", "project_id": self.ctx["project_id"], "platform": "kuaishou", "payload": {"last_seen_at": "2026-04-05T00:00:00+00:00"}, }, ) self.assertEqual(mark_read_response.status_code, 200, mark_read_response.text) mark_read_payload = mark_read_response.json() self.assertEqual(mark_read_payload["recommended_action"]["action"], "goto-tracking") self.assertEqual(mark_read_payload["recommended_action"]["screen"], "tracking") self.assertEqual(mark_read_payload["payload"]["platform"], "kuaishou") self.assertEqual(mark_read_payload["payload"]["cursor"]["last_seen_at"], "2026-04-05T00:00:00+00:00") create_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "create-assistant", "project_id": self.ctx["project_id"], "platform": "kuaishou", "payload": { "name": "快手增长 Agent", "description": "负责承接快手平台增长动作", "generation_goal": "围绕快手账号分析和同步动作持续执行", }, }, ) self.assertEqual(create_response.status_code, 200, create_response.text) create_payload = create_response.json() self.assertEqual(create_payload["recommended_action"]["action"], "open-edit-assistant") self.assertEqual(create_payload["recommended_action"]["screen"], "playbook") self.assertTrue(create_payload["recommended_action"]["assistant_id"]) self.assertEqual(create_payload["payload"]["assistant"]["name"], "快手增长 Agent") similar_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "search-similar-accounts", "project_id": self.ctx["project_id"], "platform": "kuaishou", "payload": { "target_account_id": source_id, "max_candidates": 3, "extra_requirements": "优先找商业化和知识付费方向相近的账号", }, }, ) self.assertEqual(similar_response.status_code, 200, similar_response.text) similar_payload = similar_response.json() self.assertEqual(similar_payload["recommended_action"]["action"], "select-account") self.assertEqual(similar_payload["recommended_action"]["screen"], "discovery") self.assertTrue(similar_payload["recommended_action"]["account_id"]) self.assertNotEqual(similar_payload["recommended_action"]["account_id"], source_id) self.assertEqual(similar_payload["payload"]["platform"], "kuaishou") self.assertEqual(similar_payload["payload"]["source_account_id"], source_id) self.assertGreaterEqual(int(similar_payload["payload"]["search"]["candidate_count"] or 0), 1) save_benchmark_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "save-benchmark-link", "project_id": self.ctx["project_id"], "platform": "kuaishou", "payload": { "source_account_id": source_id, "note": "由主 Agent 直接加入对标库", }, }, ) self.assertEqual(save_benchmark_response.status_code, 200, save_benchmark_response.text) save_benchmark_payload = save_benchmark_response.json() self.assertEqual(save_benchmark_payload["recommended_action"]["action"], "select-account") self.assertEqual(save_benchmark_payload["recommended_action"]["screen"], "discovery") self.assertTrue(save_benchmark_payload["recommended_action"]["account_id"]) self.assertNotEqual(save_benchmark_payload["recommended_action"]["account_id"], source_id) self.assertEqual(save_benchmark_payload["payload"]["platform"], "kuaishou") self.assertEqual(save_benchmark_payload["payload"]["source_account_id"], source_id) self.assertTrue(save_benchmark_payload["payload"]["link"]["id"]) def test_direct_oneliner_similarity_and_benchmark_actions_execute_real_douyin_flows(self) -> None: source_id = self._insert_douyin_account( account_id="dyacct_source", profile_url="https://www.douyin.com/user/source-account", nickname="源账号", ) candidate_id = self._insert_douyin_account( account_id="dyacct_candidate", profile_url="https://www.douyin.com/user/candidate-account", nickname="候选账号", ) similar_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "search-similar-accounts", "project_id": self.ctx["project_id"], "platform": "douyin", "payload": { "target_account_id": source_id, "max_candidates": 3, }, }, ) self.assertEqual(similar_response.status_code, 200, similar_response.text) similar_payload = similar_response.json() self.assertEqual(similar_payload["recommended_action"]["action"], "select-account") self.assertEqual(similar_payload["recommended_action"]["screen"], "discovery") self.assertEqual(similar_payload["payload"]["platform"], "douyin") self.assertEqual(similar_payload["payload"]["source_account_id"], source_id) self.assertEqual(similar_payload["payload"]["search"]["top_candidate_account_id"], candidate_id) save_benchmark_response = self.client.post( "/v2/oneliner/actions/execute", headers=self.ctx["member_headers"], json={ "action_key": "save-benchmark-link", "project_id": self.ctx["project_id"], "platform": "douyin", "payload": { "source_account_id": source_id, "note": "由主 Agent 直接加入对标库", }, }, ) self.assertEqual(save_benchmark_response.status_code, 200, save_benchmark_response.text) save_benchmark_payload = save_benchmark_response.json() self.assertEqual(save_benchmark_payload["recommended_action"]["action"], "select-account") self.assertEqual(save_benchmark_payload["recommended_action"]["screen"], "discovery") self.assertEqual(save_benchmark_payload["payload"]["platform"], "douyin") self.assertEqual(save_benchmark_payload["payload"]["source_account_id"], source_id) self.assertEqual(save_benchmark_payload["payload"]["target_account_id"], candidate_id) self.assertTrue(save_benchmark_payload["payload"]["link"]["relation_id"]) def test_platform_agent_routes_are_live(self) -> None: save_profile = self.client.put( "/v2/platform-agents/douyin/profile", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "name": "抖音增长 Agent", "mission": "优先跟踪成交型账号并收口到今日动作。", "notes": "保持高频复盘。", "status": "active", "config": {"focus": "conversion"}, }, ) self.assertEqual(save_profile.status_code, 200, save_profile.text) profile_payload = save_profile.json() self.assertEqual(profile_payload["platform"], "douyin") self.assertEqual(profile_payload["name"], "抖音增长 Agent") self.assertEqual(profile_payload["config"]["focus"], "conversion") self.assertEqual(profile_payload["current_version"]["version_no"], 1) first_profile_version_id = profile_payload["current_version"]["id"] profile_versions = self.client.get( "/v2/platform-agents/douyin/profile/versions", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(profile_versions.status_code, 200, profile_versions.text) self.assertEqual(profile_versions.json()["count"], 1) memory_response = self.client.post( "/v2/platform-agents/douyin/memories", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "memory_key": "recent-pattern", "title": "近期有效模式", "summary": "成交口播账号在晚间更稳定。", "details": {"window": "night"}, "confidence": 0.88, }, ) self.assertEqual(memory_response.status_code, 200, memory_response.text) self.assertEqual(memory_response.json()["memory_key"], "recent-pattern") skill_response = self.client.post( "/v2/platform-agents/douyin/skills", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "skill_key": "douyin-benchmark", "name": "抖音对标拆解", "status": "draft", "method": {"summary": "先拆前三条高分作品。"}, "test_spec": {"summary": "看结构和钩子是否完整。"}, "last_result": {"summary": "首轮验证通过。"}, "success_count": 1, "failure_count": 0, "last_score": 0.72, }, ) self.assertEqual(skill_response.status_code, 200, skill_response.text) skill_payload = skill_response.json() self.assertEqual(skill_payload["skill_key"], "douyin-benchmark") list_agents = self.client.get( "/v2/platform-agents", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(list_agents.status_code, 200, list_agents.text) agents_payload = list_agents.json() douyin_agent = next(item for item in agents_payload["items"] if item["platform"] == "douyin") self.assertGreaterEqual(douyin_agent["memory_count"], 1) self.assertGreaterEqual(douyin_agent["skill_count"], 1) list_memories = self.client.get( "/v2/platform-agents/douyin/memories", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(list_memories.status_code, 200, list_memories.text) self.assertEqual(list_memories.json()["count"], 1) review_skill = self.client.post( f"/v2/platform-agents/douyin/skills/{skill_payload['id']}/review", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "accepted": True, "score": 0.93, "summary": "这条技能现在可以复用。", "review_notes": "通过验收。", }, ) self.assertEqual(review_skill.status_code, 200, review_skill.text) self.assertEqual(review_skill.json()["status"], "validated") update_profile = self.client.put( "/v2/platform-agents/douyin/profile", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "name": "抖音增长 Agent", "mission": "改成优先对标高互动账号。", "notes": "先压缩近期重点方向。", "status": "active", "config": {"focus": "engagement"}, "reason": "调整当前抖音平台策略", }, ) self.assertEqual(update_profile.status_code, 200, update_profile.text) self.assertEqual(update_profile.json()["current_version"]["version_no"], 2) rollback_profile = self.client.post( "/v2/platform-agents/douyin/profile/rollback", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "version_id": first_profile_version_id, "reason": "恢复到第一版平台 Agent 配置", }, ) self.assertEqual(rollback_profile.status_code, 200, rollback_profile.text) rollback_profile_payload = rollback_profile.json() self.assertEqual(rollback_profile_payload["current_version"]["rollback_from_version_id"], first_profile_version_id) self.assertEqual(rollback_profile_payload["config"]["focus"], "conversion") profile_audits = self.client.get( "/v2/platform-agents/douyin/profile/audits", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(profile_audits.status_code, 200, profile_audits.text) self.assertGreaterEqual(profile_audits.json()["count"], 3) versions = self.client.get( f"/v2/platform-agents/douyin/skills/{skill_payload['id']}/versions", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(versions.status_code, 200, versions.text) versions_payload = versions.json() self.assertGreaterEqual(versions_payload["count"], 2) rollback = self.client.post( f"/v2/platform-agents/douyin/skills/{skill_payload['id']}/rollback", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "version_id": versions_payload["items"][-1]["id"], }, ) self.assertEqual(rollback.status_code, 200, rollback.text) self.assertEqual(rollback.json()["skill_key"], "douyin-benchmark") self_check = self.client.post( "/v2/platform-agents/douyin/self-check", headers=self.ctx["member_headers"], json={"project_id": self.ctx["project_id"], "sample_limit": 2, "remember_summary": False}, ) self.assertEqual(self_check.status_code, 200, self_check.text) self_check_payload = self_check.json() self.assertEqual(self_check_payload["platform"], "douyin") self.assertIn("route_checks", self_check_payload) self.assertIn("score", self_check_payload) run_response = self.client.post( "/v2/oneliner/runs", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "source_screen": "playbook", "source_action_key": "platform-agent-handoff", "title": "验证平台 Agent 执行回写", "summary": "确认主 Agent 完成态会回写最近平台执行信息。", "intent_key": "governance_review", "platform": "douyin", "delivery_mode": "confirm_first", "plan": { "goal": "验证平台 Agent 执行回写", "summary": "检查主 Agent 完成态后平台 Agent 是否记录最近执行。", "steps": ["读取当前主配置", "读取当前平台 Agent", "生成执行结果"], }, }, ) self.assertEqual(run_response.status_code, 200, run_response.text) run_payload = run_response.json() confirm_response = self.client.post( f"/v2/oneliner/runs/{run_payload['id']}/confirm", headers=self.ctx["member_headers"], json={"note": "执行平台 Agent 回写验证"}, ) self.assertEqual(confirm_response.status_code, 200, confirm_response.text) detail_response = self.client.get( f"/v2/oneliner/runs/{run_payload['id']}", headers=self.ctx["member_headers"], ) self.assertEqual(detail_response.status_code, 200, detail_response.text) detail_payload = detail_response.json() self.assertEqual(detail_payload["run_status"], "done") self.assertTrue( (((detail_payload.get("result") or {}).get("execution_card") or {}).get("oneliner_profile_version") or {}).get("version_id") ) self.assertTrue( (((detail_payload.get("result") or {}).get("execution_card") or {}).get("platform_agent_profile") or {}).get("version_id") ) self.assertEqual( (((detail_payload.get("result") or {}).get("execution_card") or {}).get("platform_agent_profile") or {}).get("version_no"), rollback_profile_payload["current_version"]["version_no"], ) refreshed_agents = self.client.get( "/v2/platform-agents", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(refreshed_agents.status_code, 200, refreshed_agents.text) refreshed_douyin = next(item for item in refreshed_agents.json()["items"] if item["platform"] == "douyin") self.assertIn("recent_execution", refreshed_douyin) self.assertEqual(refreshed_douyin["current_version"]["version_no"], rollback_profile_payload["current_version"]["version_no"]) self.assertEqual(refreshed_douyin["recent_execution"]["run_id"], run_payload["id"]) self.assertEqual(refreshed_douyin["recent_execution"]["title"], "验证平台 Agent 执行回写") self.assertEqual(refreshed_douyin["recent_execution"]["goal"], "验证平台 Agent 执行回写") self.assertEqual(refreshed_douyin["recent_execution"]["intent_key"], "governance_review") self.assertEqual(refreshed_douyin["recent_execution"]["platform_scope"], "single_platform") self.assertEqual(refreshed_douyin["recent_execution"]["delivery_mode"], "hybrid") self.assertEqual(refreshed_douyin["recent_execution"]["source_action_key"], "platform-agent-handoff") self.assertGreaterEqual(refreshed_douyin["recent_execution"]["oneliner_profile_version_no"], 1) self.assertTrue(refreshed_douyin["recent_execution"]["oneliner_profile_version_id"]) self.assertTrue(refreshed_douyin["recent_execution"]["platform_agent_profile_version_id"]) self.assertEqual( refreshed_douyin["recent_execution"]["platform_agent_profile_version_no"], rollback_profile_payload["current_version"]["version_no"], ) self.assertEqual(refreshed_douyin["recent_execution"]["recommended_action"]["action"], "goto-playbook") self.assertEqual(refreshed_douyin["recent_execution"]["recommended_action"]["screen"], "playbook") self.assertEqual(refreshed_douyin["recent_execution"]["workstream_key"], "playbook") self.assertEqual(refreshed_douyin["recent_execution"]["workstream_label"], "Agent 治理") def test_admin_ops_routes_are_live(self) -> None: now = self.db_module.utc_now() job_id = "job_failed_admin_ops" with self.core.db.session() as conn: conn.execute("PRAGMA foreign_keys=OFF") conn.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 (?, ?, ?, '', '', '', '', ?, ?, ?, 'n8n', 'collector', '', '', ?, 'auto', 'failed', '', '', 'pending', ?, '{}', '{}', '', ?, ?) """, ( job_id, self.ctx["member_id"], self.ctx["project_id"], "text", "analysis", "analysis_pipeline", "Admin Ops Failed Job", "synthetic failure for admin ops", now, now, ), ) conn.execute("PRAGMA foreign_keys=ON") scan = self.client.post( "/v2/admin/ops/incidents/scan", headers=self.ctx["admin_headers"], ) self.assertEqual(scan.status_code, 200, scan.text) scan_payload = scan.json() self.assertGreaterEqual(scan_payload["count"], 1) incident = next(item for item in scan_payload["created_or_updated"] if item["source_type"] == "job") overview = self.client.get( "/v2/admin/ops/overview", headers=self.ctx["admin_headers"], ) self.assertEqual(overview.status_code, 200, overview.text) overview_payload = overview.json() self.assertGreaterEqual(overview_payload["incident_count"], 1) self.assertIn("integration_health", overview_payload) repair_plan = self.client.post( f"/v2/admin/ops/incidents/{incident['id']}/repair-plan", headers=self.ctx["admin_headers"], json={"scope": "tenant", "notes": "先生成最小修复计划。"}, ) self.assertEqual(repair_plan.status_code, 200, repair_plan.text) repair_payload = repair_plan.json() self.assertEqual(repair_payload["plan_scope"], "tenant") review = self.client.patch( f"/v2/admin/ops/incidents/{incident['id']}", headers=self.ctx["admin_headers"], json={"status": "watching", "review_notes": "继续观察后再执行。"}, ) self.assertEqual(review.status_code, 200, review.text) self.assertEqual(review.json()["status"], "watching") audit = self.client.post( f"/v2/admin/ops/fix-runs/{repair_payload['id']}/audit", headers=self.ctx["admin_headers"], json={"review_status": "approved", "review_notes": "方案通过。"}, ) self.assertEqual(audit.status_code, 200, audit.text) self.assertEqual(audit.json()["audit_status"], "approved") fix_runs = self.client.get( "/v2/admin/ops/fix-runs", headers=self.ctx["admin_headers"], ) self.assertEqual(fix_runs.status_code, 200, fix_runs.text) self.assertGreaterEqual(fix_runs.json()["count"], 1) def test_admin_governance_directory_lists_accounts_and_projects(self) -> None: response = self.client.get( "/v2/admin/oneliner/governance/directory", headers=self.ctx["admin_headers"], ) self.assertEqual(response.status_code, 200, response.text) payload = response.json() self.assertGreaterEqual(payload["count"], 2) member = next((item for item in payload["items"] if item["id"] == self.ctx["member_id"]), None) self.assertIsNotNone(member) assert member is not None self.assertEqual(member["project_count"], 1) self.assertEqual(member["projects"][0]["id"], self.ctx["project_id"]) def test_admin_override_versions_support_rollback(self) -> None: first_response = self.client.post( "/v2/admin/oneliner/governance/overrides", headers=self.ctx["admin_headers"], json={ "target_user_id": self.ctx["member_id"], "target_project_id": self.ctx["project_id"], "platform": "douyin", "title": "Override v1", "summary": "first override", "policy": {"actions": {"max_cards": 2}}, "reason": "first override", }, ) self.assertEqual(first_response.status_code, 200, first_response.text) first_version_id = first_response.json()["current_version"]["id"] second_response = self.client.post( "/v2/admin/oneliner/governance/overrides", headers=self.ctx["admin_headers"], json={ "target_user_id": self.ctx["member_id"], "target_project_id": self.ctx["project_id"], "platform": "douyin", "title": "Override v2", "summary": "second override", "policy": {"actions": {"max_cards": 5}}, "reason": "second override", }, ) self.assertEqual(second_response.status_code, 200, second_response.text) versions_before = self.client.get( "/v2/admin/oneliner/governance/overrides/versions", headers=self.ctx["admin_headers"], params={ "target_user_id": self.ctx["member_id"], "target_project_id": self.ctx["project_id"], "platform": "douyin", }, ) self.assertEqual(versions_before.status_code, 200, versions_before.text) self.assertEqual(versions_before.json()["count"], 2) rollback_response = self.client.post( "/v2/admin/oneliner/governance/overrides/rollback", headers=self.ctx["admin_headers"], json={ "target_user_id": self.ctx["member_id"], "target_project_id": self.ctx["project_id"], "platform": "douyin", "version_id": first_version_id, "reason": "rollback to v1", }, ) self.assertEqual(rollback_response.status_code, 200, rollback_response.text) rollback_payload = rollback_response.json() self.assertEqual(rollback_payload["current_version"]["rollback_from_version_id"], first_version_id) self.assertEqual(rollback_payload["effective_policy"]["actions"]["max_cards"], 2) versions_after = self.client.get( "/v2/admin/oneliner/governance/overrides/versions", headers=self.ctx["admin_headers"], params={ "target_user_id": self.ctx["member_id"], "target_project_id": self.ctx["project_id"], "platform": "douyin", }, ) self.assertEqual(versions_after.status_code, 200, versions_after.text) self.assertEqual(versions_after.json()["count"], 3) def test_user_global_versions_support_rollback_by_creating_new_version(self) -> None: first_response = self.client.put( "/v2/oneliner/governance/user/global", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "title": "Global strategy v1", "policy": {"tone": {"style": "analytical"}}, "reason": "first pass", }, ) self.assertEqual(first_response.status_code, 200, first_response.text) first_version_id = first_response.json()["current_version"]["id"] second_response = self.client.put( "/v2/oneliner/governance/user/global", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "title": "Global strategy v2", "policy": {"tone": {"style": "decisive"}}, "reason": "refine tone", }, ) self.assertEqual(second_response.status_code, 200, second_response.text) versions_before = self.client.get( "/v2/oneliner/governance/user/global/versions", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(versions_before.status_code, 200, versions_before.text) self.assertEqual(versions_before.json()["count"], 2) rollback_response = self.client.post( "/v2/oneliner/governance/user/global/rollback", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "version_id": first_version_id, "reason": "restore best baseline", }, ) self.assertEqual(rollback_response.status_code, 200, rollback_response.text) rollback_payload = rollback_response.json() self.assertEqual(rollback_payload["current_version"]["rollback_from_version_id"], first_version_id) self.assertEqual(rollback_payload["effective_policy"]["tone"]["style"], "analytical") versions_after = self.client.get( "/v2/oneliner/governance/user/global/versions", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(versions_after.status_code, 200, versions_after.text) self.assertEqual(versions_after.json()["count"], 3) def test_user_platform_versions_support_rollback_by_creating_new_version(self) -> None: first_response = self.client.put( "/v2/oneliner/governance/user/platforms/douyin", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "title": "Douyin strategy v1", "policy": {"douyin": {"benchmark_mode": "strict"}}, "reason": "first platform pass", }, ) self.assertEqual(first_response.status_code, 200, first_response.text) first_version_id = first_response.json()["current_version"]["id"] second_response = self.client.put( "/v2/oneliner/governance/user/platforms/douyin", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "title": "Douyin strategy v2", "policy": {"douyin": {"benchmark_mode": "aggressive"}}, "reason": "push harder", }, ) self.assertEqual(second_response.status_code, 200, second_response.text) versions_before = self.client.get( "/v2/oneliner/governance/user/platforms/douyin/versions", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(versions_before.status_code, 200, versions_before.text) self.assertEqual(versions_before.json()["count"], 2) rollback_response = self.client.post( "/v2/oneliner/governance/user/platforms/douyin/rollback", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "version_id": first_version_id, "reason": "restore previous platform strategy", }, ) self.assertEqual(rollback_response.status_code, 200, rollback_response.text) rollback_payload = rollback_response.json() self.assertEqual(rollback_payload["current_version"]["rollback_from_version_id"], first_version_id) self.assertEqual(rollback_payload["effective_policy"]["douyin"]["benchmark_mode"], "strict") versions_after = self.client.get( "/v2/oneliner/governance/user/platforms/douyin/versions", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(versions_after.status_code, 200, versions_after.text) self.assertEqual(versions_after.json()["count"], 3) def test_user_policy_audits_include_personal_and_admin_layers_for_project(self) -> None: self.client.put( "/v2/oneliner/governance/user/global", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "title": "Global strategy", "policy": {"tone": {"style": "analytical"}}, "reason": "personalize defaults", }, ) self.client.post( "/v2/admin/oneliner/governance/overrides", headers=self.ctx["admin_headers"], json={ "target_user_id": self.ctx["member_id"], "target_project_id": self.ctx["project_id"], "platform": "douyin", "title": "Admin override", "policy": {"actions": {"max_cards": 4}}, "reason": "contain drift", }, ) response = self.client.get( "/v2/oneliner/governance/user/audits", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"], "platform": "douyin"}, ) self.assertEqual(response.status_code, 200, response.text) payload = response.json() self.assertGreaterEqual(payload["count"], 2) scope_kinds = [item["scope_kind"] for item in payload["items"]] self.assertIn("user_global", scope_kinds) self.assertIn("admin_override", scope_kinds) first_item = payload["items"][0] self.assertIn("version", first_item) self.assertIn("scope", first_item) self.assertNotIn("actor_user_id", first_item) self.assertNotIn("policy", first_item["version"]) self.assertNotIn("reason", first_item["version"]) self.assertNotIn("actor_user_id", first_item["version"]) def test_admin_policy_audits_include_target_and_system_layers(self) -> None: self.client.put( "/v2/admin/oneliner/governance/system/main-agent", headers=self.ctx["admin_headers"], json={ "title": "System main", "policy": {"homepage": {"focus": "ops"}}, "reason": "seed system baseline", }, ) self.client.post( "/v2/admin/oneliner/governance/overrides", headers=self.ctx["admin_headers"], json={ "target_user_id": self.ctx["member_id"], "target_project_id": self.ctx["project_id"], "platform": "douyin", "title": "Admin override", "policy": {"actions": {"max_cards": 5}}, "reason": "focus target account", }, ) response = self.client.get( "/v2/admin/oneliner/governance/audits", headers=self.ctx["admin_headers"], params={ "target_user_id": self.ctx["member_id"], "target_project_id": self.ctx["project_id"], "platform": "douyin", "include_system": "1", }, ) self.assertEqual(response.status_code, 200, response.text) payload = response.json() self.assertGreaterEqual(payload["count"], 2) scope_kinds = [item["scope_kind"] for item in payload["items"]] self.assertIn("system_main", scope_kinds) self.assertIn("admin_override", scope_kinds) def test_non_admin_cannot_change_system_defaults(self) -> None: response = self.client.put( "/v2/admin/oneliner/governance/system/main-agent", headers=self.ctx["member_headers"], json={ "title": "Not allowed", "policy": {"tone": {"style": "rogue"}}, "reason": "should be blocked", }, ) self.assertEqual(response.status_code, 403, response.text) def test_oneliner_profile_versions_and_rollback_are_available(self) -> None: initial = self.client.get( "/v2/oneliner/profile", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(initial.status_code, 200, initial.text) initial_payload = initial.json() self.assertIn("current_version", initial_payload) initial_version_id = initial_payload["current_version"]["id"] update_one = self.client.put( "/v2/oneliner/profile", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "display_name": "增长总控 OneLiner", "assistant_id": "", "default_platform": "douyin", "long_term_goal": "围绕创业内容完成多平台增长", "notes": "先做抖音与小红书联动", "config": {"commercial_ready": True}, "reason": "对齐新的增长目标", }, ) self.assertEqual(update_one.status_code, 200, update_one.text) update_one_payload = update_one.json() first_saved_version_id = update_one_payload["current_version"]["id"] self.assertNotEqual(first_saved_version_id, initial_version_id) update_two = self.client.put( "/v2/oneliner/profile", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "display_name": "增长总控 OneLiner", "assistant_id": "", "default_platform": "xiaohongshu", "long_term_goal": "先把小红书对标拆解做深", "notes": "首页动作只保留一条主动作", "config": {"commercial_ready": True, "tenant_isolation_required": True}, "reason": "阶段性切到小红书主战场", }, ) self.assertEqual(update_two.status_code, 200, update_two.text) second_payload = update_two.json() second_version_id = second_payload["current_version"]["id"] self.assertNotEqual(second_version_id, first_saved_version_id) self.assertEqual(second_payload["default_platform"], "xiaohongshu") versions_response = self.client.get( "/v2/oneliner/profile/versions", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(versions_response.status_code, 200, versions_response.text) versions_payload = versions_response.json() self.assertGreaterEqual(versions_payload["count"], 3) rollback_response = self.client.post( "/v2/oneliner/profile/rollback", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "version_id": first_saved_version_id, "reason": "回到抖音主战场配置", }, ) self.assertEqual(rollback_response.status_code, 200, rollback_response.text) rollback_payload = rollback_response.json() self.assertEqual(rollback_payload["default_platform"], "douyin") self.assertEqual(rollback_payload["long_term_goal"], "围绕创业内容完成多平台增长") self.assertEqual(rollback_payload["current_version"]["rollback_from_version_id"], first_saved_version_id) audits_response = self.client.get( "/v2/oneliner/profile/audits", headers=self.ctx["member_headers"], params={"project_id": self.ctx["project_id"]}, ) self.assertEqual(audits_response.status_code, 200, audits_response.text) audits_payload = audits_response.json() self.assertGreaterEqual(audits_payload["count"], 3) action_keys = [item["action_key"] for item in audits_payload["items"]] self.assertIn("update-oneliner-profile", action_keys) self.assertIn("rollback-oneliner-profile", action_keys) def test_oneliner_message_execution_card_tracks_config_versions(self) -> None: profile_response = self.client.put( "/v2/oneliner/profile", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "display_name": "增长总控 OneLiner", "assistant_id": "", "default_platform": "douyin", "long_term_goal": "优先分析当前平台账号并收口到下一步动作", "notes": "验证消息卡里的配置追溯链", "config": {"analysis_mode": "fast"}, "reason": "给消息卡提供明确的主配置版本", }, ) self.assertEqual(profile_response.status_code, 200, profile_response.text) current_profile_version = profile_response.json()["current_version"] platform_response = self.client.put( "/v2/platform-agents/douyin/profile", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "name": "抖音增长 Agent", "mission": "优先分析当前账号和高分作品", "notes": "验证消息卡里的平台配置追溯链", "status": "active", "config": {"focus": "analysis"}, }, ) self.assertEqual(platform_response.status_code, 200, platform_response.text) current_platform_version = platform_response.json()["current_version"] session_response = self.client.post( "/v2/oneliner/sessions", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "preferred_platform": "douyin", "title": "消息卡配置追溯", }, ) self.assertEqual(session_response.status_code, 200, session_response.text) session_payload = session_response.json() message_response = self.client.post( f"/v2/oneliner/sessions/{session_payload['id']}/messages", headers=self.ctx["member_headers"], json={ "project_id": self.ctx["project_id"], "platform": "douyin", "content": "帮我创建 Agent", }, ) self.assertEqual(message_response.status_code, 200, message_response.text) payload = message_response.json() execution_card = (((payload.get("assistant_message") or {}).get("result")) or {}).get("execution_card") or {} self.assertEqual((execution_card.get("primary_action") or {}).get("key"), "direct-create-assistant") self.assertEqual((execution_card.get("oneliner_profile_version") or {}).get("version_id"), current_profile_version["id"]) self.assertEqual((execution_card.get("oneliner_profile_version") or {}).get("version_no"), current_profile_version["version_no"]) self.assertEqual((execution_card.get("platform_agent_profile") or {}).get("platform"), "douyin") self.assertEqual((execution_card.get("platform_agent_profile") or {}).get("version_id"), current_platform_version["id"]) self.assertEqual((execution_card.get("platform_agent_profile") or {}).get("version_no"), current_platform_version["version_no"])