feat(tower): retrieve_work_item_by_identifier — UUID lookup by 'PROJECT-N'

CHANGELOG.md

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **`retrieve_work_item_by_identifier`** MCP tool on tower — resolves a
+  human-readable identifier (`<PROJECT_IDENTIFIER>-<N>`, e.g. `COINEX-72`) to
+  its full work-item record with UUID. Counterpart to `pickup_issue` for
+  interactive use when the operator has only the canonical identifier (from a
+  Plane browse URL or a comment) and no UUID. SDLC-pipeline agents continue
+  to use `pickup_issue` — they receive UUIDs via the spawn prompt.
+- `PlaneClient.retrieve_issue_by_sequence_id` — underlying lookup, walks
+  `list_issues` and short-circuits on first match (Plane v1 silently ignores
+  `?sequence_id=` as a query filter).
+
 ## [0.1.0] — 2026-05-03
 
 ### Added

src/plane_conductor/mcp_tower.py

@@ -512,6 +512,65 @@ async def pickup_issue(
 # ----- find_artifact_by_label / list_sub_issues ----------------------------
 
 
+_IDENTIFIER_RE = re.compile(r"^([A-Z][A-Z0-9]*)-(\d+)$")
+
+
+@mcp.tool()
+async def retrieve_work_item_by_identifier(
+    identifier: str,
+    *,
+    workspace: str | None = None,
+) -> dict[str, Any]:
+    """Resolve a human-readable identifier `<PROJECT_IDENTIFIER>-<N>` to its UUID.
+
+    Counterpart to `pickup_issue` for interactive use — when the operator
+    has only the canonical identifier (e.g. from a Plane browse URL or a
+    comment reference) and no UUID. SDLC-pipeline agents already receive
+    UUIDs via the spawn prompt and should keep using `pickup_issue`.
+
+    Args:
+      identifier: full identifier like `"COINEX-72"` (case-insensitive
+        prefix). Must match `^[A-Z][A-Z0-9]*-\\d+$`.
+      workspace: explicit workspace slug. Optional — when the identifier
+        prefix matches a registered `project_identifier`, the workspace is
+        auto-resolved.
+
+    Returns the same shape as `pickup_issue` plus a canonical
+    `identifier` field. Raises `ValueError` on a malformed identifier or
+    when no work item with that sequence exists in the resolved project.
+    """
+    cleaned = identifier.strip().upper()
+    m = _IDENTIFIER_RE.match(cleaned)
+    if not m:
+        raise ValueError(
+            f"identifier {identifier!r} must match '<PROJECT_IDENTIFIER>-<N>' (e.g. 'COINEX-72')"
+        )
+    project_identifier, seq_str = m.group(1), m.group(2)
+    sequence_id = int(seq_str)
+
+    ctx = _registry().resolve(workspace=workspace, project_identifier=project_identifier)
+    async with await _client_for(ctx) as plane:
+        issue = await plane.retrieve_issue_by_sequence_id(ctx.config.project_id, sequence_id)
+    if issue is None:
+        raise ValueError(
+            f"work item {cleaned} not found in workspace {ctx.config.workspace_slug!r}"
+        )
+    return {
+        "id": issue.get("id"),
+        "sequence_id": issue.get("sequence_id"),
+        "identifier": f"{ctx.project_identifier}-{issue.get('sequence_id')}",
+        "name": issue.get("name"),
+        "project_id": str(ctx.config.project_id),
+        "project_identifier": ctx.project_identifier,
+        "workspace_slug": ctx.config.workspace_slug,
+        "url": f"{ctx.config.plane_base_url}/{ctx.config.workspace_slug}/projects/"
+        f"{ctx.config.project_id}/issues/{issue.get('id')}/",
+        "parent": issue.get("parent"),
+        "labels": issue.get("labels") or [],
+        "state": issue.get("state"),
+    }
+
+
 @mcp.tool()
 async def find_artifact_by_label(
     role: str,

src/plane_conductor/plane_client.py

@@ -358,6 +358,28 @@ async def list_issues(
                 params["fields"] = fields
         return out
 
+    async def retrieve_issue_by_sequence_id(
+        self,
+        project_id: str | UUID,
+        sequence_id: int,
+        *,
+        fields: str | None = LIST_ISSUES_DEFAULT_FIELDS,
+    ) -> dict[str, Any] | None:
+        """Resolve a project-scoped sequence_id (the `<N>` in `<IDENT>-<N>`) to
+        the full work-item record. Returns `None` if no issue with that
+        sequence exists.
+
+        Plane v1 silently ignores `?sequence_id=` as a query filter (same
+        category as `?parent=`), so we walk pages via `list_issues` and
+        short-circuit on first match. Typical projects have ≤ a few hundred
+        issues — at most two paginated round-trips.
+        """
+        issues = await self.list_issues(project_id, fields=fields)
+        for issue in issues:
+            if issue.get("sequence_id") == sequence_id:
+                return issue
+        return None
+
     async def get_project(self, project_id: str | UUID) -> dict[str, Any]:
         """Fetch a single project (identifier, name, ...). Wrap the GET so the
         tower doesn't reach into `_request` directly."""

tests/test_mcp_tower.py

@@ -43,6 +43,7 @@
     post_comment,
     post_review,
     request_handoff,
+    retrieve_work_item_by_identifier,
 )
 
 # ---------------------------------------------------------------------------
@@ -192,6 +193,99 @@ async def test_pickup_issue_by_uuid(
     assert result["project_identifier"] == "TEST"
 
 
+# ---------------------------------------------------------------------------
+# retrieve_work_item_by_identifier
+# ---------------------------------------------------------------------------
+
+
+@respx.mock
+async def test_retrieve_work_item_by_identifier_happy_path(
+    registry: TowerRegistry, ctx: WorkspaceContext, project_id: str
+) -> None:
+    """The canonical use case: operator types 'TEST-42', tower resolves
+    workspace via the project_identifier prefix and returns the full record."""
+    list_url = (
+        f"https://plane.test/api/v1/workspaces/{ctx.config.workspace_slug}/"
+        f"projects/{project_id}/issues/"
+    )
+    respx.get(list_url).mock(
+        return_value=httpx.Response(
+            200,
+            json={
+                "results": [
+                    {"id": "uuid-41", "sequence_id": 41, "name": "Older"},
+                    {
+                        "id": ROOT_UUID,
+                        "sequence_id": 42,
+                        "name": "Add user dashboard",
+                        "parent": None,
+                        "labels": [],
+                        "state": "state-uuid",
+                    },
+                ],
+                "next_cursor": None,
+            },
+        )
+    )
+    result = await retrieve_work_item_by_identifier(identifier="TEST-42")
+    assert result["id"] == ROOT_UUID
+    assert result["sequence_id"] == 42
+    assert result["identifier"] == "TEST-42"
+    assert result["workspace_slug"] == ctx.config.workspace_slug
+    assert result["project_identifier"] == "TEST"
+
+
+@respx.mock
+async def test_retrieve_work_item_by_identifier_case_insensitive(
+    registry: TowerRegistry, ctx: WorkspaceContext, project_id: str
+) -> None:
+    """Operators paste identifiers from many sources — accept lowercase too."""
+    list_url = (
+        f"https://plane.test/api/v1/workspaces/{ctx.config.workspace_slug}/"
+        f"projects/{project_id}/issues/"
+    )
+    respx.get(list_url).mock(
+        return_value=httpx.Response(
+            200,
+            json={"results": [{"id": ROOT_UUID, "sequence_id": 42}], "next_cursor": None},
+        )
+    )
+    result = await retrieve_work_item_by_identifier(identifier="test-42")
+    assert result["sequence_id"] == 42
+
+
+async def test_retrieve_work_item_by_identifier_malformed_raises(
+    registry: TowerRegistry,
+) -> None:
+    """Identifier shape is part of the contract — reject early with a clear message."""
+    with pytest.raises(ValueError, match="must match"):
+        await retrieve_work_item_by_identifier(identifier="not-an-identifier")
+    with pytest.raises(ValueError, match="must match"):
+        await retrieve_work_item_by_identifier(identifier="42")
+    with pytest.raises(ValueError, match="must match"):
+        await retrieve_work_item_by_identifier(identifier="TEST-")
+
+
+@respx.mock
+async def test_retrieve_work_item_by_identifier_not_found_raises(
+    registry: TowerRegistry, ctx: WorkspaceContext, project_id: str
+) -> None:
+    """When the prefix resolves but the sequence_id isn't in the project,
+    callers get a ValueError naming the identifier they asked for."""
+    list_url = (
+        f"https://plane.test/api/v1/workspaces/{ctx.config.workspace_slug}/"
+        f"projects/{project_id}/issues/"
+    )
+    respx.get(list_url).mock(
+        return_value=httpx.Response(
+            200,
+            json={"results": [{"id": "uuid-1", "sequence_id": 1}], "next_cursor": None},
+        )
+    )
+    with pytest.raises(ValueError, match="TEST-999 not found"):
+        await retrieve_work_item_by_identifier(identifier="TEST-999")
+
+
 # ---------------------------------------------------------------------------
 # find_artifact_by_label
 # ---------------------------------------------------------------------------

tests/test_plane_client.py

@@ -211,6 +211,48 @@ async def test_list_issues_forwards_fields_on_pagination(client: PlaneClient) ->
     await client.aclose()
 
 
+@respx.mock
+async def test_retrieve_issue_by_sequence_id_finds_match(client: PlaneClient) -> None:
+    """First-match short-circuit — the iteration stops as soon as the sequence_id
+    is found, but the underlying list_issues walks pages until then."""
+    url = f"{BASE}/api/v1/workspaces/{SLUG}/projects/{PROJECT}/issues/"
+    respx.get(url).mock(
+        return_value=httpx.Response(
+            200,
+            json={
+                "results": [
+                    {"id": "uuid-71", "sequence_id": 71, "name": "Other"},
+                    {"id": "uuid-72", "sequence_id": 72, "name": "Target"},
+                    {"id": "uuid-73", "sequence_id": 73, "name": "Newer"},
+                ],
+                "next_cursor": None,
+            },
+        )
+    )
+    issue = await client.retrieve_issue_by_sequence_id(PROJECT, 72)
+    assert issue is not None
+    assert issue["id"] == "uuid-72"
+    assert issue["name"] == "Target"
+    await client.aclose()
+
+
+@respx.mock
+async def test_retrieve_issue_by_sequence_id_returns_none_when_missing(
+    client: PlaneClient,
+) -> None:
+    """Caller-friendly contract: missing sequence_id yields None, not an exception."""
+    url = f"{BASE}/api/v1/workspaces/{SLUG}/projects/{PROJECT}/issues/"
+    respx.get(url).mock(
+        return_value=httpx.Response(
+            200,
+            json={"results": [{"id": "uuid-1", "sequence_id": 1}], "next_cursor": None},
+        )
+    )
+    issue = await client.retrieve_issue_by_sequence_id(PROJECT, 9999)
+    assert issue is None
+    await client.aclose()
+
+
 @respx.mock
 async def test_request_retries_on_429_with_retry_after(client: PlaneClient) -> None:
     """429 with `Retry-After: N` must sleep N seconds and retry. Plane's