feat: implement ISSUE-WP-0005 connector mapping and scope alignment

Add MappingService (work_record_uuid ↔ external id), project/map CLI with
outward push-status, optional TaskSpec work_record fields, boundary-sync
policy, and packaging/capability framing cleanup. Mark WP-0005 finished.

issue_core/api/ingest.py

@@ -21,6 +21,7 @@ from ..backends.gitea import GiteaBackend
 from ..backends.local import LocalSQLiteBackend
 from ..cli.utils import get_config_dir, load_backend_configs
 from ..core.interfaces import BackendFactory, IssueBackend
+from ..core.mapping import MappingService
 from ..core.models import Issue, IssueState, Label
 from .auth import require_api_key
 from .schemas import BackendName, TaskIngestionRequest, TaskIngestionResponse
@@ -87,6 +88,12 @@ def _build_issue(payload: TaskIngestionRequest, backend_type: str) -> Issue:
         ingestion_meta["due_at"] = (now + timedelta(days=payload.due_in_days)).isoformat()
 
     sync_metadata: Dict[str, Any] = {"ingestion": ingestion_meta}
+    if payload.work_record_uuid:
+        sync_metadata["mapping"] = {
+            "work_record_uuid": payload.work_record_uuid,
+            "work_record_id": payload.work_record_id,
+            "work_record_kind": payload.work_record_kind,
+        }
 
     return Issue(
         id="",
@@ -102,15 +109,65 @@ def _build_issue(payload: TaskIngestionRequest, backend_type: str) -> Issue:
     )
 
 
+def _mapping_backend_name(backend_type: str) -> str:
+    return "sqlite" if backend_type == "local" else backend_type
+
+
+def _upsert_mapping_if_requested(
+    payload: TaskIngestionRequest,
+    *,
+    backend_type: str,
+    issue_id: str,
+    issue_url: Optional[str],
+) -> None:
+    if not payload.work_record_uuid:
+        return
+    svc = MappingService()
+    svc.connect(str(get_config_dir() / "mappings.db"))
+    try:
+        svc.upsert(
+            work_record_uuid=payload.work_record_uuid,
+            work_record_id=payload.work_record_id,
+            work_record_kind=payload.work_record_kind,
+            backend=_mapping_backend_name(backend_type),
+            external_id=issue_id,
+            external_url=issue_url,
+            target_repo=payload.target_repo,
+            direction="outward",
+        )
+    finally:
+        svc.disconnect()
+
+
 @router.post(
     "/issues/",
     response_model=TaskIngestionResponse,
     status_code=status.HTTP_201_CREATED,
     dependencies=[Depends(require_api_key)],
-    summary="Ingest a task from an external emitter (e.g. activity-core).",
+    summary="Create an external tracker issue (intentional emit; optional work-record mapping).",
 )
 async def ingest_task(payload: TaskIngestionRequest) -> TaskIngestionResponse:
     backend, backend_type = _resolve_backend()
+    backend_name: BackendName = _BACKEND_TYPE_TO_NAME.get(backend_type, backend_type)  # type: ignore[assignment]
+    mapping_backend = _mapping_backend_name(backend_type)
+
+    # Idempotent: if work_record_uuid already mapped, return existing issue.
+    if payload.work_record_uuid:
+        map_svc = MappingService()
+        map_svc.connect(str(get_config_dir() / "mappings.db"))
+        try:
+            existing = map_svc.get_active_by_uuid(
+                payload.work_record_uuid, backend=mapping_backend
+            )
+            if existing:
+                return TaskIngestionResponse(
+                    issue_id=existing.external_id,
+                    issue_url=existing.external_url,
+                    backend=backend_name,
+                )
+        finally:
+            map_svc.disconnect()
+
     draft = _build_issue(payload, backend_type)
     try:
         created: Issue = backend.create_issue(draft)
@@ -131,7 +188,20 @@ async def ingest_task(payload: TaskIngestionRequest) -> TaskIngestionResponse:
     issue_url: Optional[str] = None
     if created.sync_metadata:
         issue_url = created.sync_metadata.get("url") or created.sync_metadata.get("html_url")
-    backend_name: BackendName = _BACKEND_TYPE_TO_NAME.get(backend_type, backend_type)  # type: ignore[assignment]
+
+    try:
+        _upsert_mapping_if_requested(
+            payload,
+            backend_type=backend_type,
+            issue_id=issue_id,
+            issue_url=issue_url,
+        )
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(exc),
+        ) from exc
+
     return TaskIngestionResponse(
         issue_id=issue_id,
         issue_url=issue_url,

issue_core/api/schemas.py

@@ -34,10 +34,25 @@ class TaskIngestionRequest(BaseModel):
         min_length=1,
         description=(
             "Activity event UUID, or a stable scheduler/source key when no "
-            "event row exists."
+            "event row exists. Not a work-record UUID."
         ),
     )
     activity_definition_id: str = Field(..., min_length=1)
+    work_record_uuid: Optional[str] = Field(
+        default=None,
+        description=(
+            "Optional work-record UUIDv7. When set, ingestion upserts a "
+            "mapping row (work_record_uuid ↔ backend issue id)."
+        ),
+    )
+    work_record_id: Optional[str] = Field(
+        default=None,
+        description="Optional canonical work-record id (human-facing).",
+    )
+    work_record_kind: Optional[str] = Field(
+        default=None,
+        description="Optional work-record kind (task, intake, …).",
+    )
 
 
 class TaskIngestionResponse(BaseModel):

issue_core/cli/main.py

@@ -12,6 +12,7 @@ from .commands import issue_group
 from .backend_commands import backend_group
 from .sync_commands import sync_group
 from .serve_command import serve_command
+from .map_commands import map_group, project_command
 from .. import __version__
 
 
@@ -23,22 +24,19 @@ from .. import __version__
 @click.pass_context
 def cli(ctx, config, backend, verbose):
     """
-    Universal Issue Tracking System
+    External issue-tracker connector (Gitea/Forgejo, local SQLite).
 
-    A backend-agnostic issue tracking tool that works with local SQLite,
-    Gitea, GitHub, and other issue tracking systems.
+    Not the origin of fleet work records — see INTENT.md. Use project/map
+    to link work-record UUIDs to external issues when a tracker is in use.
 
     Examples:
-        issue list                    # List all issues
-        issue create "Bug in parser"  # Create new issue
-        issue show 42                 # Show issue #42
-        issue close 42                # Close issue #42
-
-        backend add local ~/.issues   # Add local backend
-        backend add gitea myrepo      # Add Gitea backend
-
-        sync pull gitea               # Sync from Gitea
-        sync push gitea               # Sync to Gitea
+        issue list
+        issue create "Tracker note"
+        issue project <work-record-uuid> --title "..."
+        issue map show --uuid <uuid>
+        issue map push-status --uuid <uuid> --status progress
+        backend add local ~/.issues
+        sync pull gitea
     """
     # Ensure the object exists
     ctx.ensure_object(dict)
@@ -54,6 +52,8 @@ cli.add_command(issue_group, name='issue')
 cli.add_command(backend_group, name='backend')
 cli.add_command(sync_group, name='sync')
 cli.add_command(serve_command)
+cli.add_command(map_group, name='map')
+cli.add_command(project_command, name='project')
 
 
 # Convenience aliases - direct issue commands

issue_core/cli/map_commands.py

@@ -0,0 +1,421 @@
+"""
+CLI for work-record ↔ external issue mapping (connector role).
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from typing import Optional
+from uuid import UUID
+
+import click
+
+from ..core.mapping import MappingService, map_work_record_status_to_issue_state
+from ..core.models import Issue, IssueState, Label
+from .utils import echo_error, echo_info, echo_success, get_backend, get_config_dir
+
+
+def _default_mapping_db() -> str:
+    return str(get_config_dir() / "mappings.db")
+
+
+def _open_mapping(db_path: Optional[str] = None) -> MappingService:
+    svc = MappingService()
+    svc.connect(db_path or _default_mapping_db())
+    return svc
+
+
+def _looks_like_uuid(value: str) -> bool:
+    try:
+        UUID(value)
+        return True
+    except (ValueError, TypeError, AttributeError):
+        return False
+
+
+def _split_ref(ref: str) -> tuple:
+    """Return (work_record_uuid|None, work_record_id|None)."""
+    if _looks_like_uuid(ref):
+        return ref, None
+    return None, ref
+
+
+def _backend_label(backend) -> str:
+    """Map backend_type to mapping.backend name."""
+    bt = getattr(backend, "backend_type", None) or "local"
+    return "sqlite" if bt == "local" else bt
+
+
+def _issue_external_id(issue: Issue) -> str:
+    return issue.id or str(issue.number)
+
+
+def _issue_url(issue: Issue) -> Optional[str]:
+    if issue.sync_metadata:
+        return issue.sync_metadata.get("url") or issue.sync_metadata.get("html_url")
+    return None
+
+
+def _print_mapping(mapping, as_json: bool = False) -> None:
+    if as_json:
+        click.echo(json.dumps(mapping.to_dict(), indent=2))
+    else:
+        click.echo(
+            f"{mapping.work_record_uuid}  "
+            f"{mapping.work_record_id or '-'}  "
+            f"{mapping.backend}:{mapping.external_id}  "
+            f"[{mapping.status}/{mapping.direction}]"
+        )
+        if mapping.external_url:
+            click.echo(f"  url: {mapping.external_url}")
+
+
+@click.command("project")
+@click.argument("work_record_ref")
+@click.option("--title", "-t", help="Issue title (required if creating new)")
+@click.option("--description", "-d", default="", help="Issue description")
+@click.option("--kind", "work_record_kind", default="task", help="Work-record kind")
+@click.option(
+    "--canonical-id",
+    help="Canonical work-record id (if ref is a UUID)",
+)
+@click.option("--label", "-l", multiple=True, help="Labels")
+@click.option("--target-repo", help="Optional target_repo denorm")
+@click.option("--mapping-db", type=click.Path(), help="Override mappings.db path")
+@click.option("--format", "output_format", type=click.Choice(["text", "json"]), default="text")
+@click.pass_context
+def project_command(
+    ctx,
+    work_record_ref,
+    title,
+    description,
+    work_record_kind,
+    canonical_id,
+    label,
+    target_repo,
+    mapping_db,
+    output_format,
+):
+    """Project a work record to the default tracker backend (idempotent)."""
+    wr_uuid, wr_id = _split_ref(work_record_ref)
+    if wr_uuid and canonical_id:
+        wr_id = canonical_id
+    if not wr_uuid and not wr_id:
+        raise click.ClickException("work_record_ref is required")
+
+    # For mapping we need a UUID key. If only canonical id given, derive a
+    # stable namespace UUID so bookkeeping still works until hub UUID is known.
+    if not wr_uuid:
+        # Deterministic UUIDv5 from canonical id — not a hub UUIDv7; operators
+        # should prefer real hub UUIDs when available.
+        from uuid import uuid5, NAMESPACE_URL
+
+        wr_uuid = str(uuid5(NAMESPACE_URL, f"work-record:{wr_id}"))
+        echo_info(
+            f"No UUID given; using deterministic key {wr_uuid} for canonical id {wr_id}"
+        )
+
+    backend = get_backend(ctx)
+    backend_name = _backend_label(backend)
+    svc = _open_mapping(mapping_db)
+    try:
+        existing = svc.get_active_by_uuid(wr_uuid, backend=backend_name)
+        if existing:
+            if output_format == "json":
+                click.echo(json.dumps({"mapping": existing.to_dict(), "created": False}, indent=2))
+            else:
+                echo_info("Active mapping already exists (idempotent).")
+                _print_mapping(existing)
+            return
+
+        if not title:
+            title = f"[projection] {wr_id or wr_uuid}"
+
+        now = datetime.now(timezone.utc)
+        labels = [Label(name=n) for n in label]
+        labels.append(Label(name=f"work_record:{wr_id or wr_uuid}"))
+        draft = Issue(
+            id="",
+            number=0,
+            title=title,
+            description=description or "",
+            state=IssueState.OPEN,
+            created_at=now,
+            updated_at=now,
+            labels=labels,
+            backend_type=getattr(backend, "backend_type", backend_name),
+            sync_metadata={
+                "mapping": {
+                    "work_record_uuid": wr_uuid,
+                    "work_record_id": wr_id,
+                    "work_record_kind": work_record_kind,
+                }
+            },
+        )
+        created = backend.create_issue(draft)
+        ext_id = _issue_external_id(created)
+        mapping = svc.upsert(
+            work_record_uuid=wr_uuid,
+            work_record_id=wr_id,
+            work_record_kind=work_record_kind,
+            backend=backend_name,
+            external_id=ext_id,
+            external_url=_issue_url(created),
+            target_repo=target_repo,
+            direction="outward",
+        )
+        if output_format == "json":
+            click.echo(
+                json.dumps(
+                    {
+                        "mapping": mapping.to_dict(),
+                        "created": True,
+                        "issue_id": ext_id,
+                        "issue_number": created.number,
+                    },
+                    indent=2,
+                )
+            )
+        else:
+            echo_success(f"Projected → {backend_name}:{ext_id} (#{created.number})")
+            _print_mapping(mapping)
+    finally:
+        svc.disconnect()
+        try:
+            backend.disconnect()
+        except Exception:
+            pass
+
+
+@click.group("map")
+def map_group():
+    """Work-record ↔ external issue mapping."""
+
+
+@map_group.command("link")
+@click.argument("work_record_ref")
+@click.argument("external_id")
+@click.option("--kind", "work_record_kind", default="task")
+@click.option("--canonical-id", help="Canonical id when ref is UUID")
+@click.option("--url", "external_url", help="External issue URL")
+@click.option("--target-repo")
+@click.option("--mapping-db", type=click.Path())
+@click.option("--format", "output_format", type=click.Choice(["text", "json"]), default="text")
+@click.pass_context
+def map_link(
+    ctx,
+    work_record_ref,
+    external_id,
+    work_record_kind,
+    canonical_id,
+    external_url,
+    target_repo,
+    mapping_db,
+    output_format,
+):
+    """Link an existing external issue to a work record (no create)."""
+    wr_uuid, wr_id = _split_ref(work_record_ref)
+    if wr_uuid and canonical_id:
+        wr_id = canonical_id
+    if not wr_uuid and wr_id:
+        from uuid import uuid5, NAMESPACE_URL
+
+        wr_uuid = str(uuid5(NAMESPACE_URL, f"work-record:{wr_id}"))
+
+    backend = get_backend(ctx)
+    backend_name = _backend_label(backend)
+    # Verify issue exists when possible
+    issue = backend.get_issue(str(external_id))
+    if issue is None:
+        try:
+            issue = backend.get_issue_by_number(int(external_id))
+        except (TypeError, ValueError):
+            issue = None
+    if issue is None:
+        echo_info("Issue not found on backend; linking by id only.")
+        ext_id = str(external_id)
+        url = external_url
+    else:
+        ext_id = _issue_external_id(issue)
+        url = external_url or _issue_url(issue)
+
+    svc = _open_mapping(mapping_db)
+    try:
+        mapping = svc.upsert(
+            work_record_uuid=wr_uuid,
+            work_record_id=wr_id,
+            work_record_kind=work_record_kind,
+            backend=backend_name,
+            external_id=ext_id,
+            external_url=url,
+            target_repo=target_repo,
+            direction="linked",
+        )
+        if output_format == "json":
+            click.echo(json.dumps(mapping.to_dict(), indent=2))
+        else:
+            echo_success("Linked.")
+            _print_mapping(mapping)
+    finally:
+        svc.disconnect()
+        try:
+            backend.disconnect()
+        except Exception:
+            pass
+
+
+@map_group.command("show")
+@click.option("--uuid", "work_record_uuid", help="Work-record UUIDv7")
+@click.option("--id", "work_record_id", help="Canonical work-record id")
+@click.option("--external", "external_id", help="External issue id")
+@click.option("--backend", "backend_name", help="Backend name for external lookup")
+@click.option("--mapping-db", type=click.Path())
+@click.option("--format", "output_format", type=click.Choice(["text", "json"]), default="text")
+@click.pass_context
+def map_show(
+    ctx, work_record_uuid, work_record_id, external_id, backend_name, mapping_db, output_format
+):
+    """Show active mapping by UUID, canonical id, or external id."""
+    if not any([work_record_uuid, work_record_id, external_id]):
+        raise click.ClickException("Provide --uuid, --id, or --external")
+    if external_id and not backend_name:
+        backend = get_backend(ctx)
+        backend_name = _backend_label(backend)
+        try:
+            backend.disconnect()
+        except Exception:
+            pass
+
+    svc = _open_mapping(mapping_db)
+    try:
+        mapping = svc.resolve(
+            work_record_uuid=work_record_uuid,
+            work_record_id=work_record_id,
+            backend=backend_name,
+            external_id=external_id,
+        )
+        if not mapping:
+            raise click.ClickException("No active mapping found")
+        _print_mapping(mapping, as_json=(output_format == "json"))
+    finally:
+        svc.disconnect()
+
+
+@map_group.command("detach")
+@click.option("--uuid", "work_record_uuid", help="Work-record UUID")
+@click.option("--id", "work_record_id", help="Canonical work-record id")
+@click.option("--backend", "backend_name", help="Limit detach to one backend")
+@click.option("--mapping-db", type=click.Path())
+@click.option("--format", "output_format", type=click.Choice(["text", "json"]), default="text")
+def map_detach(work_record_uuid, work_record_id, backend_name, mapping_db, output_format):
+    """Detach (soft) an active mapping."""
+    if not work_record_uuid and not work_record_id:
+        raise click.ClickException("Provide --uuid or --id")
+    svc = _open_mapping(mapping_db)
+    try:
+        mapping = svc.detach(
+            work_record_uuid=work_record_uuid,
+            work_record_id=work_record_id,
+            backend=backend_name,
+        )
+        if not mapping:
+            raise click.ClickException("No active mapping to detach")
+        if output_format == "json":
+            click.echo(json.dumps(mapping.to_dict(), indent=2))
+        else:
+            echo_success("Detached.")
+            _print_mapping(mapping)
+    finally:
+        svc.disconnect()
+
+
+@map_group.command("list")
+@click.option("--limit", type=int, default=50)
+@click.option("--mapping-db", type=click.Path())
+@click.option("--format", "output_format", type=click.Choice(["text", "json"]), default="text")
+def map_list(limit, mapping_db, output_format):
+    """List active mappings."""
+    svc = _open_mapping(mapping_db)
+    try:
+        rows = svc.list_active(limit=limit)
+        if output_format == "json":
+            click.echo(json.dumps([r.to_dict() for r in rows], indent=2))
+        else:
+            if not rows:
+                click.echo("No active mappings.")
+                return
+            for m in rows:
+                _print_mapping(m)
+    finally:
+        svc.disconnect()
+
+
+@map_group.command("push-status")
+@click.option("--uuid", "work_record_uuid", help="Work-record UUID")
+@click.option("--id", "work_record_id", help="Canonical work-record id")
+@click.option(
+    "--status",
+    "work_status",
+    required=True,
+    help="Fleet work-record status (e.g. progress, done) or tracker state",
+)
+@click.option("--mapping-db", type=click.Path())
+@click.option("--format", "output_format", type=click.Choice(["text", "json"]), default="text")
+@click.pass_context
+def map_push_status(
+    ctx, work_record_uuid, work_record_id, work_status, mapping_db, output_format
+):
+    """Outward-only: set external issue state from work-record status (v1)."""
+    if not work_record_uuid and not work_record_id:
+        raise click.ClickException("Provide --uuid or --id")
+    try:
+        issue_state = map_work_record_status_to_issue_state(work_status)
+    except ValueError as exc:
+        raise click.ClickException(str(exc)) from exc
+
+    svc = _open_mapping(mapping_db)
+    backend = get_backend(ctx)
+    try:
+        mapping = svc.resolve(
+            work_record_uuid=work_record_uuid,
+            work_record_id=work_record_id,
+            backend=_backend_label(backend),
+        )
+        if not mapping:
+            raise click.ClickException("No active mapping found")
+
+        issue = backend.get_issue(mapping.external_id)
+        if issue is None:
+            try:
+                issue = backend.get_issue_by_number(int(mapping.external_id))
+            except (TypeError, ValueError):
+                issue = None
+        if issue is None:
+            raise click.ClickException(f"External issue {mapping.external_id} not found")
+
+        issue.state = IssueState.from_string(issue_state)
+        issue.updated_at = datetime.now(timezone.utc)
+        updated = backend.update_issue(issue)
+        svc.mark_pushed(mapping.id)
+
+        payload = {
+            "mapping": mapping.to_dict(),
+            "work_record_status": work_status,
+            "issue_state": issue_state,
+            "external_id": mapping.external_id,
+            "issue_number": updated.number,
+        }
+        if output_format == "json":
+            click.echo(json.dumps(payload, indent=2))
+        else:
+            echo_success(
+                f"Pushed status {work_status!r} → issue state {issue_state!r} "
+                f"on {mapping.backend}:{mapping.external_id}"
+            )
+    finally:
+        svc.disconnect()
+        try:
+            backend.disconnect()
+        except Exception:
+            pass

issue_core/core/mapping.py

@@ -0,0 +1,381 @@
+"""
+Work-record UUID ↔ external issue id mapping.
+
+Fleet bookkeeping for the connector role (INTENT / work-record canon).
+Storage is a dedicated SQLite file so mapping works regardless of whether
+the active CRUD backend is local or Gitea.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+
+ACTIVE = "active"
+DETACHED = "detached"
+SUPERSEDED = "superseded"
+
+
+@dataclass
+class WorkRecordMapping:
+    """One projection link between a work record and a backend issue."""
+
+    id: str
+    work_record_uuid: str
+    work_record_id: Optional[str]
+    work_record_kind: Optional[str]
+    backend: str
+    external_id: str
+    external_url: Optional[str]
+    target_repo: Optional[str]
+    direction: str  # outward | linked
+    status: str  # active | detached | superseded
+    created_at: str
+    updated_at: str
+    last_pushed_at: Optional[str] = None
+    last_pulled_at: Optional[str] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "id": self.id,
+            "work_record_uuid": self.work_record_uuid,
+            "work_record_id": self.work_record_id,
+            "work_record_kind": self.work_record_kind,
+            "backend": self.backend,
+            "external_id": self.external_id,
+            "external_url": self.external_url,
+            "target_repo": self.target_repo,
+            "direction": self.direction,
+            "status": self.status,
+            "created_at": self.created_at,
+            "updated_at": self.updated_at,
+            "last_pushed_at": self.last_pushed_at,
+            "last_pulled_at": self.last_pulled_at,
+        }
+
+
+# Work-record task status → tracker IssueState (outward v1).
+# See docs/boundary-sync-and-status-mapping.md
+TASK_STATUS_TO_ISSUE_STATE = {
+    "wait": "open",
+    "todo": "open",
+    "progress": "in_progress",
+    "done": "closed",
+    "cancel": "closed",
+    # Already tracker-ish values (pass-through)
+    "open": "open",
+    "closed": "closed",
+    "in_progress": "in_progress",
+    "blocked": "blocked",
+}
+
+
+def map_work_record_status_to_issue_state(status: str) -> str:
+    """Map fleet work-record status (or tracker state) to IssueState value."""
+    key = (status or "").strip().lower().replace("-", "_")
+    if key not in TASK_STATUS_TO_ISSUE_STATE:
+        raise ValueError(
+            f"Unknown status {status!r}; expected one of "
+            f"{sorted(set(TASK_STATUS_TO_ISSUE_STATE))}"
+        )
+    return TASK_STATUS_TO_ISSUE_STATE[key]
+
+
+class MappingService:
+    """CRUD for work_record_issue_map rows."""
+
+    def __init__(self, db_path: Optional[str] = None):
+        self.db_path = db_path or "mappings.db"
+        self.connection: Optional[sqlite3.Connection] = None
+
+    def connect(self, db_path: Optional[str] = None) -> None:
+        if db_path:
+            self.db_path = db_path
+        Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
+        self.connection = sqlite3.connect(self.db_path)
+        self.connection.row_factory = sqlite3.Row
+        self.connection.execute("PRAGMA foreign_keys = ON")
+        self._initialize_schema()
+
+    def disconnect(self) -> None:
+        if self.connection:
+            self.connection.close()
+            self.connection = None
+
+    def _initialize_schema(self) -> None:
+        assert self.connection is not None
+        self.connection.executescript(
+            """
+            CREATE TABLE IF NOT EXISTS work_record_issue_map (
+                id TEXT PRIMARY KEY,
+                work_record_uuid TEXT NOT NULL,
+                work_record_id TEXT,
+                work_record_kind TEXT,
+                backend TEXT NOT NULL,
+                external_id TEXT NOT NULL,
+                external_url TEXT,
+                target_repo TEXT,
+                direction TEXT NOT NULL DEFAULT 'outward',
+                status TEXT NOT NULL DEFAULT 'active'
+                    CHECK (status IN ('active', 'detached', 'superseded')),
+                created_at TEXT NOT NULL,
+                updated_at TEXT NOT NULL,
+                last_pushed_at TEXT,
+                last_pulled_at TEXT
+            );
+            CREATE UNIQUE INDEX IF NOT EXISTS idx_map_active_uuid_backend
+                ON work_record_issue_map (work_record_uuid, backend)
+                WHERE status = 'active';
+            CREATE UNIQUE INDEX IF NOT EXISTS idx_map_active_backend_external
+                ON work_record_issue_map (backend, external_id)
+                WHERE status = 'active';
+            CREATE INDEX IF NOT EXISTS idx_map_work_record_id
+                ON work_record_issue_map (work_record_id);
+            """
+        )
+        self.connection.commit()
+
+    def _require_conn(self) -> sqlite3.Connection:
+        if not self.connection:
+            raise RuntimeError("MappingService is not connected")
+        return self.connection
+
+    def _row_to_mapping(self, row: sqlite3.Row) -> WorkRecordMapping:
+        return WorkRecordMapping(
+            id=row["id"],
+            work_record_uuid=row["work_record_uuid"],
+            work_record_id=row["work_record_id"],
+            work_record_kind=row["work_record_kind"],
+            backend=row["backend"],
+            external_id=row["external_id"],
+            external_url=row["external_url"],
+            target_repo=row["target_repo"],
+            direction=row["direction"],
+            status=row["status"],
+            created_at=row["created_at"],
+            updated_at=row["updated_at"],
+            last_pushed_at=row["last_pushed_at"],
+            last_pulled_at=row["last_pulled_at"],
+        )
+
+    def get_active_by_uuid(
+        self, work_record_uuid: str, backend: Optional[str] = None
+    ) -> Optional[WorkRecordMapping]:
+        conn = self._require_conn()
+        if backend:
+            cur = conn.execute(
+                """
+                SELECT * FROM work_record_issue_map
+                WHERE work_record_uuid = ? AND backend = ? AND status = 'active'
+                """,
+                (work_record_uuid, backend),
+            )
+        else:
+            cur = conn.execute(
+                """
+                SELECT * FROM work_record_issue_map
+                WHERE work_record_uuid = ? AND status = 'active'
+                ORDER BY updated_at DESC
+                """,
+                (work_record_uuid,),
+            )
+        row = cur.fetchone()
+        return self._row_to_mapping(row) if row else None
+
+    def get_active_by_canonical_id(
+        self, work_record_id: str, backend: Optional[str] = None
+    ) -> Optional[WorkRecordMapping]:
+        conn = self._require_conn()
+        if backend:
+            cur = conn.execute(
+                """
+                SELECT * FROM work_record_issue_map
+                WHERE work_record_id = ? AND backend = ? AND status = 'active'
+                """,
+                (work_record_id, backend),
+            )
+        else:
+            cur = conn.execute(
+                """
+                SELECT * FROM work_record_issue_map
+                WHERE work_record_id = ? AND status = 'active'
+                ORDER BY updated_at DESC
+                """,
+                (work_record_id,),
+            )
+        row = cur.fetchone()
+        return self._row_to_mapping(row) if row else None
+
+    def get_active_by_external(
+        self, backend: str, external_id: str
+    ) -> Optional[WorkRecordMapping]:
+        conn = self._require_conn()
+        cur = conn.execute(
+            """
+            SELECT * FROM work_record_issue_map
+            WHERE backend = ? AND external_id = ? AND status = 'active'
+            """,
+            (backend, str(external_id)),
+        )
+        row = cur.fetchone()
+        return self._row_to_mapping(row) if row else None
+
+    def resolve(
+        self,
+        *,
+        work_record_uuid: Optional[str] = None,
+        work_record_id: Optional[str] = None,
+        backend: Optional[str] = None,
+        external_id: Optional[str] = None,
+    ) -> Optional[WorkRecordMapping]:
+        if work_record_uuid:
+            return self.get_active_by_uuid(work_record_uuid, backend=backend)
+        if work_record_id:
+            return self.get_active_by_canonical_id(work_record_id, backend=backend)
+        if backend and external_id is not None:
+            return self.get_active_by_external(backend, external_id)
+        raise ValueError("Provide work_record_uuid, work_record_id, or backend+external_id")
+
+    def upsert(
+        self,
+        *,
+        work_record_uuid: str,
+        backend: str,
+        external_id: str,
+        work_record_id: Optional[str] = None,
+        work_record_kind: Optional[str] = None,
+        external_url: Optional[str] = None,
+        target_repo: Optional[str] = None,
+        direction: str = "outward",
+    ) -> WorkRecordMapping:
+        """Idempotent active mapping for (uuid, backend)."""
+        conn = self._require_conn()
+        now = datetime.now(timezone.utc).isoformat()
+        existing = self.get_active_by_uuid(work_record_uuid, backend=backend)
+        if existing:
+            if existing.external_id != str(external_id):
+                raise ValueError(
+                    f"Active mapping for {work_record_uuid} on {backend} already "
+                    f"points at external_id={existing.external_id!r}, not {external_id!r}"
+                )
+            conn.execute(
+                """
+                UPDATE work_record_issue_map SET
+                    work_record_id = COALESCE(?, work_record_id),
+                    work_record_kind = COALESCE(?, work_record_kind),
+                    external_url = COALESCE(?, external_url),
+                    target_repo = COALESCE(?, target_repo),
+                    updated_at = ?
+                WHERE id = ?
+                """,
+                (
+                    work_record_id,
+                    work_record_kind,
+                    external_url,
+                    target_repo,
+                    now,
+                    existing.id,
+                ),
+            )
+            conn.commit()
+            refreshed = self.get_active_by_uuid(work_record_uuid, backend=backend)
+            assert refreshed is not None
+            return refreshed
+
+        # External id already mapped to a different UUID?
+        conflict = self.get_active_by_external(backend, external_id)
+        if conflict and conflict.work_record_uuid != work_record_uuid:
+            raise ValueError(
+                f"external_id={external_id!r} on {backend} already mapped to "
+                f"{conflict.work_record_uuid}"
+            )
+
+        mapping_id = str(uuid.uuid4())
+        conn.execute(
+            """
+            INSERT INTO work_record_issue_map (
+                id, work_record_uuid, work_record_id, work_record_kind,
+                backend, external_id, external_url, target_repo,
+                direction, status, created_at, updated_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?)
+            """,
+            (
+                mapping_id,
+                work_record_uuid,
+                work_record_id,
+                work_record_kind,
+                backend,
+                str(external_id),
+                external_url,
+                target_repo,
+                direction,
+                now,
+                now,
+            ),
+        )
+        conn.commit()
+        created = self.get_active_by_uuid(work_record_uuid, backend=backend)
+        assert created is not None
+        return created
+
+    def detach(
+        self,
+        *,
+        work_record_uuid: Optional[str] = None,
+        work_record_id: Optional[str] = None,
+        backend: Optional[str] = None,
+    ) -> Optional[WorkRecordMapping]:
+        conn = self._require_conn()
+        mapping = self.resolve(
+            work_record_uuid=work_record_uuid,
+            work_record_id=work_record_id,
+            backend=backend,
+        )
+        if not mapping:
+            return None
+        now = datetime.now(timezone.utc).isoformat()
+        conn.execute(
+            """
+            UPDATE work_record_issue_map
+            SET status = 'detached', updated_at = ?
+            WHERE id = ?
+            """,
+            (now, mapping.id),
+        )
+        conn.commit()
+        cur = conn.execute(
+            "SELECT * FROM work_record_issue_map WHERE id = ?", (mapping.id,)
+        )
+        row = cur.fetchone()
+        return self._row_to_mapping(row) if row else None
+
+    def mark_pushed(self, mapping_id: str) -> None:
+        conn = self._require_conn()
+        now = datetime.now(timezone.utc).isoformat()
+        conn.execute(
+            """
+            UPDATE work_record_issue_map
+            SET last_pushed_at = ?, updated_at = ?
+            WHERE id = ?
+            """,
+            (now, now, mapping_id),
+        )
+        conn.commit()
+
+    def list_active(self, limit: int = 100) -> List[WorkRecordMapping]:
+        conn = self._require_conn()
+        cur = conn.execute(
+            """
+            SELECT * FROM work_record_issue_map
+            WHERE status = 'active'
+            ORDER BY updated_at DESC
+            LIMIT ?
+            """,
+            (limit,),
+        )
+        return [self._row_to_mapping(r) for r in cur.fetchall()]