from __future__ import annotations import importlib import os import sys import tempfile import unittest 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.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 = [ "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_messages", "oneliner_sessions", "oneliner_profiles", "auth_tokens", "projects", "accounts", "model_profiles", ] for table in tables: try: self.core.db.execute(f"DELETE FROM {table}") except Exception: continue 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 _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.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"], "当前焦点") event_types = [item["event_type"] for item in payload["events"]] self.assertIn("run.progress", event_types) self.assertIn("run.done", event_types) 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_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)