refactor(agent): separate tool surface from operation registry

backend/app/cli.py

@@ -16,7 +16,7 @@
 from app.ops import get_operation_schema, list_operations
 from app.bridge_cli import main as bridge_main
 from app.runtime_paths import runtime_data_dir, runtime_uploads_dir
-from app.services.agent_skill_registry import catalog, registry_snapshot, resolve_skill
+from app.services.agent_skill_registry import (\n    agent_operation_names,\n    catalog,\n    registry_snapshot,\n    resolve_skill,\n)
 from app.services.agent_host_registry import host_capability_matrix
 from app.services.operation_projection import (
     confirm_operation_proposal,
@@ -107,6 +107,12 @@ def main(argv: Optional[list[str]] = None) -> int:
                 {
                     "ok": True,
                     "operation_count": len(list_operations()),
+                    "operation_registry_count": len(list_operations()),
+                    "agent_tool_count": len(agent_operation_names()),
+                    "featured_tool_count": len(agent_operation_names(featured_only=True)),
+                    "internal_operation_count": (
+                        len(list_operations()) - len(agent_operation_names())
+                    ),
                     "returned_count": len(operations),
                     "selector": selector,
                     "operations": operations,
@@ -192,11 +198,19 @@ def _build_parser() -> JsonArgumentParser:
     manifest_selector.add_argument("--group", default="", help="Show compact Operation summaries for one group.")
     manifest_selector.add_argument("--all", dest="all_operations", action="store_true", help="Show the full developer audit manifest.")
 
-    ops = sub.add_parser("ops", help="Discover atomic internal operations.", add_help=False)
+    ops = sub.add_parser(
+        "ops",
+        help="Discover Agent-facing OfferU tools; --all is the raw Operation Registry audit.",
+        add_help=False,
+    )
     ops.add_argument("--pretty", action="store_true", help="Pretty-print JSON.")
     ops_selector = ops.add_mutually_exclusive_group()
     ops_selector.add_argument("--skill", default="", help="List compact Operations for one Skill.")
-    ops_selector.add_argument("--group", default="", help="List compact Operations for one group.")
+    ops_selector.add_argument(
+        "--group",
+        default="",
+        help="List Agent-exposed Operations for one group; internal/legacy entries stay hidden.",
+    )
     ops_selector.add_argument("--all", dest="all_operations", action="store_true", help="List every full Operation schema for auditing.")
 
     schema = sub.add_parser("schema", help="Show one operation schema.", add_help=False)
@@ -266,6 +280,13 @@ def _doctor() -> dict[str, Any]:
     provider_health = _doctor_provider_health()
     data_safety = _doctor_data_safety()
     frontend_health = _doctor_frontend_health()
+    agent_names = agent_operation_names()
+    featured_names = agent_operation_names(featured_only=True)
+    agent_schemas = [
+        operation
+        for operation in operation_schemas
+        if str(operation.get("name") or "") in agent_names
+    ]
     return {
         "ok": True,
         "service": "OfferU CLI",
@@ -723,14 +744,13 @@ def _select_operations(
         known_groups = {str(operation.get("group") or "ungrouped") for operation in operations}
         if clean_group not in known_groups:
             raise ValueError(f"未知能力组: {clean_group}")
+        # Group discovery is an Agent surface, not a raw Registry dump.  UI-only,
+        # migration, diagnostic and compatibility Operations stay reachable by
+        # exact schema/run name and by the explicit --all developer audit path.
+        selected_names = agent_operation_names()
         selector = f"group:{clean_group}"
     elif default_featured and not all_operations:
-        selected_names = {
-            str(operation_name)
-            for skill_item in catalog()
-            if skill_item.get("featured")
-            for operation_name in skill_item.get("allowed_tools", [])
-        }
+        selected_names = agent_operation_names(featured_only=True)
         selector = "featured_skills"
     if selected_names is not None:
         operations = [operation for operation in operations if operation.get("name") in selected_names]
@@ -816,10 +836,18 @@ def _manifest(*, skill: str = "", group: str = "", all_operations: bool = False)
             "raw_api_capability": False,
             "side_effect_labels": sorted({effect for op in operation_schemas for effect in op.get("side_effects", [])}),
         },
+        # operation_count is retained for backwards compatibility.  New callers
+        # should distinguish the governed Registry from the model-facing Tool
+        # surface instead of treating every Operation as an Agent tool.
         "operation_count": len(operation_schemas),
+        "operation_registry_count": len(operation_schemas),
+        "agent_tool_count": len(agent_names),
+        "featured_tool_count": len(featured_names),
+        "internal_operation_count": len(operation_schemas) - len(agent_names),
         "returned_count": len(operations),
         "selector": selector,
         "groups": _groups(operation_schemas),
+        "agent_groups": _groups(agent_schemas),
         "operations": operations,
         "skill_registry": skills,
         # Host × Skill support matrix (full/limited/unsupported) so an external

backend/app/ops.py

@@ -30,7 +30,6 @@
 )
 from app.services.privacy_hygiene import (
     get_privacy_hygiene_status,
-    get_synthetic_email_test_data_status,
     purge_synthetic_email_test_data,
     scrub_legacy_email_notification_bodies,
 )
@@ -92,7 +91,6 @@
     get_hosted_executor_session,
     get_career_task,
     get_career_task_result,
-    get_agent_provider_health,
     get_agent_connections,
     get_local_agent_capability_matrix,
     get_local_agent_capability_report,
@@ -217,7 +215,6 @@
     duplicate_resume_template,
     extract_interview_questions,
     generate_html_resume,
-    generate_legacy_cover_letter,
     generate_legacy_interview_answer,
     import_jobs_to_application_table,
     import_latest_extension_batch_to_application_table,
@@ -889,11 +886,6 @@ class ResumeTemplateDuplicateInput(_StrictOperationInput):
     new_name: str = Field(min_length=1, max_length=300)
 
 
-class LegacyCoverLetterInput(_StrictOperationInput):
-    job_id: int = Field(gt=0)
-    resume_id: int = Field(gt=0)
-
-
 class GenerateHtmlResumeInput(_StrictOperationInput):
     profile_id: int = Field(gt=0)
     template_id: int = Field(gt=0)
@@ -2575,14 +2567,6 @@ async def _search_jobs_via_sources(
         input_model=DelegateCareerTaskInput,
         version="2026-08-27",
     ),
-    "get_agent_provider_health": Operation(
-        name="get_agent_provider_health",
-        fn=get_agent_provider_health,
-        description="读取一个 Agent Runtime provider 的脱敏可用性、认证与阻塞状态。",
-        group="agent_runtime",
-        input_model=ProviderHealthInput,
-        version="2026-08-27",
-    ),
     "list_agent_provider_health": Operation(
         name="list_agent_provider_health",
         fn=list_agent_provider_health,
@@ -3124,13 +3108,6 @@ async def _search_jobs_via_sources(
         group="governance",
         version="2026-08-31",
     ),
-    "get_synthetic_email_test_data_status": Operation(
-        name="get_synthetic_email_test_data_status",
-        fn=get_synthetic_email_test_data_status,
-        description="读取严格限定的合成邮箱测试数据计数，不返回账号地址或邮件内容。",
-        group="governance",
-        version="2026-08-31",
-    ),
     "scrub_legacy_email_notification_bodies": Operation(
         name="scrub_legacy_email_notification_bodies",
         fn=scrub_legacy_email_notification_bodies,
@@ -3520,14 +3497,6 @@ async def _search_jobs_via_sources(
             side_effects=("llm", "write"),
             input_model=GenerateLegacyInterviewAnswerInput,
         ),
-        "generate_legacy_cover_letter": Operation(
-            name="generate_legacy_cover_letter",
-            fn=generate_legacy_cover_letter,
-            description="兼容旧投递接口生成求职信草稿。",
-            group="applications",
-            side_effects=("llm",),
-            input_model=LegacyCoverLetterInput,
-        ),
         "create_resume_template": Operation(
             name="create_resume_template",
             fn=create_resume_template,

backend/app/routes/applications.py

@@ -376,7 +376,7 @@ async def create_application(data: ApplicationCreate):
 
 @router.post("/generate")
 async def generate(data: GenerateRequest):
-    return await _execute_operation("generate_legacy_cover_letter", data.model_dump())
+    return await _execute_operation("generate_cover_letter", data.model_dump())
 
 
 @router.get("/stats")

backend/app/services/agent_operations.py

@@ -1097,12 +1097,6 @@ async def delegate_career_task(
     )
 
 
-async def get_agent_provider_health(provider_id: str) -> dict:
-    from app.services.agent_provider_health import get_provider_health
-
-    return await get_provider_health(provider_id)
-
-
 async def list_agent_provider_health() -> dict:
     from app.services.agent_provider_health import list_provider_health
 

backend/app/services/agent_skill_registry.py

@@ -134,6 +134,26 @@ def catalog() -> list[dict[str, Any]]:
     return [skill.summary() for skill in sorted(skills, key=lambda item: (item.order, item.id))]
 
 
+def agent_operation_names(*, featured_only: bool = False) -> set[str]:
+    """Return Operations intentionally exposed through Agent Skills.
+
+    The Operation Registry is the governed execution/control plane and includes
+    UI-only, migration, diagnostic and compatibility Operations.  It is not the
+    same thing as the model-facing Tool surface.
+
+    This helper is the single projection boundary used by CLI discovery:
+    - all Skill allowlists -> complete Agent Tool surface;
+    - featured Skills only -> default compact discovery surface.
+    """
+
+    names: set[str] = set()
+    for skill in catalog():
+        if featured_only and not bool(skill.get("featured")):
+            continue
+        names.update(str(name) for name in skill.get("allowed_tools") or [] if str(name))
+    return names
+
+
 def registry_snapshot(operation_schemas: list[dict[str, Any]] | None = None) -> dict[str, Any]:
     skills = catalog()
     if operation_schemas is not None:

backend/app/services/legacy_operations.py

@@ -534,34 +534,6 @@ async def generate_legacy_interview_answer(question_id: int) -> dict[str, Any]:
         }
 
 
-async def generate_legacy_cover_letter(job_id: int, resume_id: int) -> dict[str, Any]:
-    from app.agents.cover_letter import generate_cover_letter
-
-    async with async_session() as db:
-        job = (await db.execute(select(Job).where(Job.id == job_id))).scalar_one_or_none()
-        if not job:
-            raise ValueError("Job not found")
-        resume = (
-            await db.execute(select(Resume).where(Resume.id == resume_id))
-        ).scalar_one_or_none()
-        if not resume:
-            raise ValueError("Resume not found")
-
-        content = resume.content_json or {}
-        resume_text = f"姓名: {content.get('name', '')}\n"
-        resume_text += f"技能: {content.get('skills', '')}\n"
-        for experience in content.get("experience", []):
-            resume_text += (
-                f"工作经历: {experience.get('company', '')} - "
-                f"{experience.get('position', '')}\n"
-                f"  描述: {experience.get('description', '')}\n"
-            )
-        return await generate_cover_letter(
-            jd=job.raw_description or job.summary,
-            resume=resume_text,
-        )
-
-
 async def create_resume_template(
     name: str,
     thumbnail_url: str = "",

backend/tests/test_agent_tool_surface.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import inspect
+import unittest
+
+from app.cli import _manifest, _select_operations
+from app.ops import OPERATIONS
+from app.routes.applications import generate
+from app.services.agent_skill_registry import agent_operation_names
+
+
+class AgentToolSurfaceTests(unittest.TestCase):
+    def test_operation_registry_is_not_the_agent_tool_catalog(self) -> None:
+        agent_tools = agent_operation_names()
+        featured_tools = agent_operation_names(featured_only=True)
+
+        self.assertTrue(agent_tools)
+        self.assertTrue(featured_tools)
+        self.assertTrue(featured_tools.issubset(agent_tools))
+        self.assertTrue(agent_tools.issubset(OPERATIONS))
+        self.assertLess(len(agent_tools), len(OPERATIONS))
+
+    def test_group_discovery_hides_internal_and_legacy_operations(self) -> None:
+        rows, selector = _select_operations(group="applications")
+        names = {str(row.get("name") or "") for row in rows}
+
+        self.assertEqual(selector, "group:applications")
+        self.assertIn("list_applications", names)
+        self.assertIn("preview_application_action", names)
+        self.assertNotIn("create_application_table", names)
+        self.assertNotIn("create_legacy_application", names)
+        self.assertNotIn("update_legacy_application", names)
+
+    def test_explicit_all_remains_the_registry_audit_escape_hatch(self) -> None:
+        rows, selector = _select_operations(all_operations=True)
+        names = {str(row.get("name") or "") for row in rows}
+
+        self.assertEqual(selector, "all")
+        self.assertIn("create_application_table", names)
+        self.assertIn("create_legacy_application", names)
+        self.assertEqual(names, set(OPERATIONS))
+
+    def test_manifest_reports_registry_and_agent_counts_separately(self) -> None:
+        payload = _manifest()
+
+        self.assertEqual(payload["operation_registry_count"], len(OPERATIONS))
+        self.assertEqual(payload["agent_tool_count"], len(agent_operation_names()))
+        self.assertEqual(
+            payload["featured_tool_count"],
+            len(agent_operation_names(featured_only=True)),
+        )
+        self.assertEqual(
+            payload["internal_operation_count"],
+            len(OPERATIONS) - len(agent_operation_names()),
+        )
+        self.assertEqual(payload["returned_count"], 0)
+
+    def test_resume_group_excludes_route_level_crud(self) -> None:
+        payload = _manifest(group="resume")
+        names = {str(row.get("name") or "") for row in payload["operations"]}
+
+        self.assertIn("prepare_resume_optimization", names)
+        self.assertIn("export_resume_pdf", names)
+        self.assertNotIn("create_resume_template", names)
+        self.assertNotIn("create_resume_section", names)
+        self.assertNotIn("save_resume_draft_record", names)
+
+    def test_dead_and_duplicate_registry_entries_are_removed(self) -> None:
+        self.assertNotIn("get_agent_provider_health", OPERATIONS)
+        self.assertNotIn("get_synthetic_email_test_data_status", OPERATIONS)
+        self.assertNotIn("generate_legacy_cover_letter", OPERATIONS)
+
+    def test_legacy_http_cover_letter_route_uses_canonical_operation(self) -> None:
+        source = inspect.getsource(generate)
+        self.assertIn('"generate_cover_letter"', source)
+        self.assertNotIn("generate_legacy_cover_letter", source)
+
+
+if __name__ == "__main__":
+    unittest.main()