fix(yunxiao): shorten synced GitHub card titles

.github/workflows/yunxiao-github-sync.yml

@@ -15,6 +15,7 @@ on:
         options:
           - preflight
           - backfill
+          - rename-titles
       state:
         description: 回填 GitHub 数据范围
         required: false
@@ -24,7 +25,7 @@ on:
           - open
           - all
       apply:
-        description: 回填时是否实际写入云效
+        description: 回填或标题迁移时是否实际写入云效
         required: false
         type: boolean
         default: false
@@ -79,7 +80,7 @@ jobs:
 
           if [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
             args=(--mode "${{ inputs.mode }}" --state "${{ inputs.state }}")
-            if [ "${{ inputs.mode }}" = "backfill" ]; then
+            if [ "${{ inputs.mode }}" = "backfill" ] || [ "${{ inputs.mode }}" = "rename-titles" ]; then
               if [ "${{ inputs.apply }}" = "true" ]; then
                 args+=(--apply)
               else

docs/github-yunxiao-sync.md

@@ -0,0 +1,41 @@
+# GitHub → 云效标题同步
+
+`Sync GitHub to Yunxiao` 工作流将 GitHub Issue 和社区 PR 同步到仓库变量指定的云效项目。
+
+卡片标题使用短格式：
+
+- `[Issue #594] 原始 Issue 标题`
+- `[PR #766] 原始 PR 标题`
+
+卡片描述首行保留 `GitHub: https://github.com/owner/repository/issues/594`（PR 使用
+`/pull/766`）。这条来源链接用于区分仓库、类型和编号，请保留该行。
+脚本兼容原先的 `[GitHub owner/repository Issue #594]` 和 `[GitHub Issue #594]`
+标题；后续生命周期事件会复用已有卡片、缩短前缀并保留当前标题正文。来源缺失或同一来源对应多张卡片时，
+同步会报错，避免误改或重复创建。
+
+## 整理已有卡片标题
+
+在 GitHub Actions 中打开 `Sync GitHub to Yunxiao`，选择 **Run workflow**：
+
+1. `mode` 选择 `rename-titles`，`apply` 保持关闭，先检查运行日志中的预览统计。
+2. 确认 `failed=0` 和 `would_rename` 数量后，再运行相同模式并勾选 `apply`。
+3. 检查 `renamed`、`skipped`、`failed` 统计，并回查云效列表。
+
+该模式只整理目标云效项目中已存在、来源匹配当前 GitHub 仓库的卡片（含已关闭项），
+仅写入 `subject`。不会创建历史卡片，也不会修改状态、迭代、负责人、参与者或描述。
+`state` 选项只用于 `backfill`，不限制标题整理范围。
+整理重复执行不会再次改写已经使用短标题的卡片；写前回读发现标题或来源变更时会跳过并报错，
+应重新预览后再处理。回读与写入之间仍有短暂竞争窗口，整理期间请避免同时编辑卡片标题。
+
+本地对应命令（需要安全注入既有云效凭据和仓库映射变量）：
+
+```sh
+python scripts/yunxiao_github_sync.py --mode rename-titles --dry-run
+python scripts/yunxiao_github_sync.py --mode rename-titles --apply
+```
+
+`--dry-run` 优先于 `--apply`。只为改标题时不要使用 `backfill --state all --apply`，
+它会同步状态和人员等字段，并可能创建此前未同步的历史工作项。
+
+云效 API 参考：[搜索工作项](https://help.aliyun.com/zh/yunxiao/developer-reference/searchworkitems)、
+[更新工作项](https://help.aliyun.com/zh/yunxiao/developer-reference/updateworkitem)。

scripts/test_yunxiao_github_sync.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
 import importlib.util
+import io
 import json
 import os
 import tempfile
 import unittest
+from contextlib import redirect_stdout
 from pathlib import Path
 from unittest.mock import patch
 
@@ -16,6 +18,68 @@
 SPEC.loader.exec_module(MODULE)
 
 
+REPOSITORY = "MemTensor/memmy-agent"
+CFG = {
+    "org": "org", "project_id": "project", "workitem_category": "Req",
+    "type_id": "type", "assignee_id": "assignee", "priority_id": "priority",
+    "parent_id": "", "sprint_id": "sprint", "participant_ids": ["participant"],
+    "statuses": {"待处理": "pending", "已取消": "cancelled"},
+}
+
+
+def source_item(number=42, *, kind="issue", repository=REPOSITORY):
+    return {
+        "number": number, "title": "Fix scheduler", "state": "open", "labels": [],
+        "html_url": MODULE.source_url(kind, number, repository),
+        "created_at": "2026-09-10T00:00:00Z", "body": "Details",
+    }
+
+
+def source_card(identifier="card", *, number=42, kind="issue", repository=REPOSITORY, form="short"):
+    item = source_item(number, kind=kind, repository=repository)
+    prefix = {
+        "short": f"[{MODULE.SOURCE_LABELS[kind]} #{number}]",
+        "long": MODULE.build_source_key(kind, item, repository),
+        "legacy": MODULE.legacy_source_key(kind, item),
+    }[form]
+    return {
+        "id": identifier, "subject": f"{prefix} Fix scheduler",
+        "description": MODULE.item_description(item), "status": "closed-custom",
+        "sprint": "manual-sprint", "assignedTo": "manual-owner",
+    }
+
+
+class MemoryTransport:
+    """A stateful API fixture: subject searches see changes made by PUT."""
+
+    def __init__(self, cards=()):
+        self.cards = {card["id"]: dict(card) for card in cards}
+        self.calls = []
+        self.omit_description = False
+
+    def __call__(self, method, path, body=None):
+        self.calls.append((method, path, body))
+        if path.endswith("workitems:search"):
+            key = json.loads(body["conditions"])["conditionGroups"][0][0]["value"][0]
+            cards = [dict(card) for card in self.cards.values() if key in card["subject"]]
+            offset = (body["page"] - 1) * body["perPage"]
+            cards = cards[offset:offset + body["perPage"]]
+            if self.omit_description:
+                for card in cards:
+                    card.pop("description", None)
+            return cards
+        if method == "GET":
+            return dict(self.cards[path.rsplit("/", 1)[-1]])
+        if method == "PUT":
+            self.cards[path.rsplit("/", 1)[-1]].update(body)
+            return None
+        if method == "POST" and path.endswith("/workitems"):
+            identifier = f"new-{len(self.cards)}"
+            self.cards[identifier] = {"id": identifier, **body}
+            return {"id": identifier}
+        raise AssertionError((method, path, body))
+
+
 class YunxiaoSyncTests(unittest.TestCase):
     def test_source_key_includes_repository(self) -> None:
         item = {"number": 42, "title": "Fix scheduler"}
@@ -24,11 +88,14 @@ def test_source_key_includes_repository(self) -> None:
             "[GitHub MemTensor/memmy-agent Issue #42]",
         )
 
-    def test_title_includes_source_key(self) -> None:
+    def test_title_keeps_type_and_number_without_repository_prefix(self) -> None:
         item = {"number": 42, "title": "Fix scheduler"}
         self.assertEqual(
             MODULE.build_title("issue", item, "MemTensor/memmy-agent"),
-            "[GitHub MemTensor/memmy-agent Issue #42] Fix scheduler",
+            "[Issue #42] Fix scheduler",
+        )
+        self.assertEqual(
+            MODULE.build_title("pr", item, REPOSITORY), "[PR #42] Fix scheduler"
         )
 
     def test_source_status(self) -> None:
@@ -153,18 +220,23 @@ def transport(method, path, body=None):
             client=MODULE.YunxiaoClient(transport),
         )
         self.assertEqual(result, "created")
-        self.assertEqual(calls[2][2]["parentId"], "directory")
-        self.assertEqual(calls[2][2]["sprint"], "sprint")
-        self.assertEqual(calls[2][2]["participants"], ["participant-a", "participant-b"])
-        self.assertIn("Details", calls[2][2]["description"])
+        self.assertEqual(calls[-1][2]["parentId"], "directory")
+        self.assertEqual(calls[-1][2]["sprint"], "sprint")
+        self.assertEqual(calls[-1][2]["participants"], ["participant-a", "participant-b"])
+        self.assertIn("Details", calls[-1][2]["description"])
+        self.assertEqual(calls[-1][2]["subject"], "[Issue #1] Test")
 
     def test_reopened_item_updates_to_pending(self) -> None:
         calls = []
 
         def transport(method, path, body=None):
             calls.append((method, path, body))
+            existing = source_card("existing", number=2, form="long")
+            existing["subject"] = "[GitHub MemTensor/memmy-agent Issue #2] Reopen"
             if "search" in path:
-                return {"data": {"workitems": [{"id": "existing"}]}}
+                return {"data": {"workitems": [existing]}}
+            if method == "GET":
+                return existing
             return {}
 
         cfg = {
@@ -202,6 +274,7 @@ def transport(method, path, body=None):
             calls[-1][2],
             {
                 "status": "pending",
+                "subject": "[Issue #2] Reopen",
                 "assignedTo": "assignee",
                 "participants": ["participant-a", "participant-b"],
             },
@@ -249,6 +322,184 @@ def transport(method, path, body=None):
         self.assertEqual(result, "created")
         self.assertEqual(calls[-1][2]["status"], "cancelled")
 
+    def find_existing(self, transport, *, kind="issue", number=42):
+        return MODULE.find_existing_workitem(
+            "org", "project", kind, source_item(number, kind=kind), REPOSITORY,
+            MODULE.YunxiaoClient(transport), category="Req",
+        )
+
+    def sync(self, transport):
+        return MODULE.sync_one(
+            "org", CFG, "issue", source_item(), repository=REPOSITORY,
+            apply=True, label_ids={}, days_to_finish=7, create_closed=False,
+            client=MODULE.YunxiaoClient(transport),
+        )
+
+    def rename(self, transport, *, apply=False):
+        output = io.StringIO()
+        with (
+            patch.dict(os.environ, {"GITHUB_REPOSITORY": REPOSITORY}),
+            patch.object(MODULE, "configuration_from_environment", return_value=CFG),
+            patch.object(MODULE.time, "sleep"),
+            redirect_stdout(output),
+        ):
+            result = MODULE.rename_titles(MODULE.YunxiaoClient(transport), apply=apply)
+        return result, json.loads(output.getvalue())["titles"]
+
+    def test_description_source_requires_the_leading_source_line(self):
+        url = MODULE.source_url("issue", 42, REPOSITORY)
+        self.assertEqual(MODULE.description_source_url(f"GitHub: {url}\n\nBody"), url.casefold())
+        self.assertEqual(
+            MODULE.description_source_url(f'<p>GitHub: <a href="{url}">{url}</a></p><p>Body</p>'),
+            url.casefold(),
+        )
+        self.assertIsNone(MODULE.description_source_url(f"Body mentions\nGitHub: {url}"))
+        self.assertIsNone(MODULE.description_source_url(f"GitHub: {url}#issuecomment-1"))
+
+    def test_sync_after_create_and_after_rename_reuses_the_same_card(self):
+        transport = MemoryTransport()
+        self.assertEqual(self.sync(transport), "created")
+        self.assertEqual(self.sync(transport), "updated-status")
+        self.assertEqual(len(transport.cards), 1)
+        self.assertEqual(transport.cards["new-0"]["subject"], "[Issue #42] Fix scheduler")
+        for form in ("long", "legacy"):
+            with self.subTest(form=form):
+                transport = MemoryTransport([source_card(form=form)])
+                self.assertEqual(self.sync(transport), "updated-status")
+                self.assertEqual(self.sync(transport), "updated-status")
+                self.assertEqual(list(transport.cards), ["card"])
+                self.assertEqual(transport.cards["card"]["subject"], "[Issue #42] Fix scheduler")
+
+    def test_same_number_from_another_repository_is_not_selected(self):
+        transport = MemoryTransport([
+            source_card("foreign", repository="Other/repository"),
+            source_card("target"),
+        ])
+        self.assertEqual(self.find_existing(transport), "target")
+
+    def test_sync_preserves_the_human_edited_title_suffix(self):
+        for form in ("long", "legacy", "short"):
+            with self.subTest(form=form):
+                card = source_card(form=form)
+                card["subject"] = card["subject"].replace("Fix scheduler", "Human-maintained title")
+                transport = MemoryTransport([card])
+                self.assertEqual(self.sync(transport), "updated-status")
+                self.assertEqual(transport.cards["card"]["subject"], "[Issue #42] Human-maintained title")
+
+    def test_issue_and_pr_with_same_number_are_separate(self):
+        transport = MemoryTransport([source_card("issue"), source_card("pr", kind="pr")])
+        self.assertEqual(self.find_existing(transport, kind="pr"), "pr")
+
+    def test_missing_search_description_uses_workitem_detail(self):
+        transport = MemoryTransport([source_card()])
+        transport.omit_description = True
+        self.assertEqual(self.find_existing(transport), "card")
+        self.assertTrue(any(method == "GET" for method, _, _ in transport.calls))
+
+    def test_matching_candidate_on_a_later_page_is_found(self):
+        cards = [source_card(f"foreign-{i}", repository=f"Other/repository-{i}") for i in range(4)]
+        transport = MemoryTransport([*cards, source_card("target")])
+        with patch.object(MODULE, "SEARCH_PAGE_SIZE", 2):
+            self.assertEqual(self.find_existing(transport), "target")
+        self.assertIn(3, [body["page"] for _, path, body in transport.calls if path.endswith(":search")])
+
+    def test_duplicates_across_title_formats_block_writes(self):
+        transport = MemoryTransport([source_card("a"), source_card("b", form="long")])
+        with self.assertRaisesRegex(MODULE.PreflightError, "多个"):
+            self.sync(transport)
+        self.assertFalse(any(method == "PUT" for method, _, _ in transport.calls))
+
+    def test_unverifiable_matching_card_blocks_duplicate_creation(self):
+        card = source_card()
+        card["description"] = "Manual text without a source line"
+        transport = MemoryTransport([card])
+        with self.assertRaisesRegex(MODULE.PreflightError, "缺少可验证"):
+            self.sync(transport)
+        self.assertEqual(len(transport.cards), 1)
+
+    def test_malformed_search_response_does_not_look_like_no_matches(self):
+        with self.assertRaisesRegex(MODULE.PreflightError, "无法解析"):
+            self.find_existing(lambda *args: {"unexpected": []})
+
+    def test_search_stops_at_the_api_pagination_window(self):
+        transport = MemoryTransport([source_card(str(i)) for i in range(3)])
+        with (
+            patch.object(MODULE, "SEARCH_PAGE_SIZE", 2),
+            patch.object(MODULE, "SEARCH_WINDOW", 2),
+            self.assertRaisesRegex(MODULE.PreflightError, "分页窗口"),
+        ):
+            self.find_existing(transport)
+
+    def test_title_cleanup_preview_has_no_writes(self):
+        transport = MemoryTransport([source_card(form="long")])
+        before = json.dumps(transport.cards)
+        result, stats = self.rename(transport)
+        self.assertEqual(result, 0)
+        self.assertEqual(stats["would_rename"], 1)
+        self.assertEqual(stats["renamed"], 0)
+        self.assertEqual(json.dumps(transport.cards), before)
+        self.assertTrue(all(method in ("GET", "POST") and not path.endswith("/workitems")
+                            for method, path, _ in transport.calls))
+
+    def test_title_cleanup_collects_all_pages_then_changes_subject_only(self):
+        cards = [source_card(str(i), number=i + 1, form="long") for i in range(5)]
+        cards.append(source_card("legacy", number=6, form="legacy"))
+        cards.append(source_card("foreign", repository="Other/repo", form="legacy"))
+        transport = MemoryTransport(cards)
+        with patch.object(MODULE, "SEARCH_PAGE_SIZE", 2):
+            result, stats = self.rename(transport, apply=True)
+        self.assertEqual(result, 0)
+        self.assertEqual(stats["renamed"], 6)
+        self.assertEqual(stats["skipped"], 1)
+        for before in cards:
+            after = transport.cards[before["id"]]
+            for field in ("description", "status", "sprint", "assignedTo"):
+                self.assertEqual(before[field], after[field])
+        writes = [body for method, _, body in transport.calls if method == "PUT"]
+        self.assertEqual(len(writes), 6)
+        self.assertTrue(all(set(body) == {"subject"} for body in writes))
+        transport.calls.clear()
+        _, repeated = self.rename(transport, apply=True)
+        self.assertEqual(repeated["renamed"], 0)
+        self.assertFalse(any(method == "PUT" for method, _, _ in transport.calls))
+
+    def test_title_cleanup_duplicate_sources_are_not_renamed(self):
+        transport = MemoryTransport([source_card("a", form="long"), source_card("b", form="legacy")])
+        result, stats = self.rename(transport, apply=True)
+        self.assertEqual(result, 1)
+        self.assertEqual(stats["failed"], 2)
+        self.assertFalse(any(method == "PUT" for method, _, _ in transport.calls))
+
+    def test_title_cleanup_detects_duplicate_already_short_cards(self):
+        transport = MemoryTransport([source_card("a"), source_card("b", form="long")])
+        result, stats = self.rename(transport, apply=True)
+        self.assertEqual(result, 1)
+        self.assertEqual(stats["failed"], 2)
+        self.assertEqual(stats["renamed"], 0)
+        self.assertFalse(any(method == "PUT" for method, _, _ in transport.calls))
+
+    def test_title_cleanup_does_not_overwrite_a_concurrent_edit(self):
+        transport = MemoryTransport([source_card(form="long")])
+
+        def concurrent_transport(method, path, body=None):
+            if method == "GET":
+                transport.cards["card"]["subject"] = "Manually edited title"
+            return transport(method, path, body)
+
+        result, stats = self.rename(concurrent_transport, apply=True)
+        self.assertEqual(result, 1)
+        self.assertEqual(stats["failed"], 1)
+        self.assertFalse(any(method == "PUT" for method, _, _ in transport.calls))
+
+    def test_explicit_dry_run_overrides_apply_for_title_cleanup(self):
+        with (
+            patch.dict(os.environ, {"YUNXIAO_TOKEN": "test-token"}),
+            patch.object(MODULE.sys, "argv", ["sync", "--mode", "rename-titles", "--apply", "--dry-run"]),
+            patch.object(MODULE, "rename_titles", return_value=0) as rename,
+        ):
+            self.assertEqual(MODULE.main(), 0)
+        self.assertFalse(rename.call_args.kwargs["apply"])
+
 
 if __name__ == "__main__":
     unittest.main()