feat(ops-run): carry harness_profile_ref and attribution refs

ACTIVITY-WP-0032-T02 / ACT-ADR-006. ops_runs grows harness_profile_ref (text,
indexed) and execution_refs (jsonb), migration 0008, threaded through the
emission path, queue projection, and run artefacts.

The important part is the enforcement, not the columns. ACT-ADR-006 says
approach_hint must never override, synthesize, or fall back from an absent or
invalid harness_profile_ref — a silent fallback would reintroduce the
claim-time routing failure of 2026-08-17. resolve_execution_selector never
consults the hint: a malformed ref raises even when a hint is present, and
ACTIVITY_CORE_REQUIRE_HARNESS_PROFILE turns an absent ref into an error once
definitions have migrated.

Validation is structural only, since the glas-harness catalogue is
authoritative and must not be mirrored. Requiring the <id>@<version> pin is
worth doing locally: GlasProfiles.resolve matches an unpinned ref against every
version and refuses it as ambiguous, so the pin converts a late failure into an
emission-time error without knowing any profile id.

Migration verified on real PostgreSQL 16: upgrade, downgrade, re-upgrade, and a
legacy-shaped row still inserts and stays claimable.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/activity_core/activities.py

@@ -495,6 +495,8 @@ async def emit_tasks(payload: dict) -> list[str]:
                             session,
                             spec,
                             approach_hint=spec_dict.get("approach_hint"),
+                            harness_profile_ref=spec_dict.get("harness_profile_ref"),
+                            execution_refs=spec_dict.get("execution_refs"),
                         )
                         if ops_id is not None:
                             activity.logger.info(

src/activity_core/glas_profile.py

@@ -0,0 +1,131 @@
+"""Harness profile selection for queued ops runs (ACT-ADR-006).
+
+activity-core names an **approved, versioned harness profile** on a governed
+ops_run and passes the attribution references through. It does not resolve the
+profile, select a rein, or mirror the glas-harness catalogue — that catalogue is
+authoritative and Glas exposes no network validation service, so a local copy
+would drift and produce a second, disagreeing opinion about what is executable.
+
+What we can check locally is *structure*, and one structural property matters a
+great deal: ``GlasProfiles.resolve`` (glas_harness/profiles.py) treats a ref
+without ``@version`` as matching every version of that id and raises
+``AmbiguousProfileError`` when more than one exists. Requiring the pin here
+turns a late ambiguity failure into an emission-time error, without knowing a
+single profile id.
+
+Everything else — whether a well-formed ref names a profile that exists, is
+enabled, and matches the contract version — is the execution-side Glas
+resolver's mandatory fail-closed check, by design.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from typing import Any
+
+# `harness.agent-dev-local@1.0.0` — id, then a pinned version.
+_PROFILE_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+_PROFILE_VERSION_PATTERN = re.compile(r"^[0-9]+(\.[0-9]+)*([A-Za-z0-9.\-+]*)$")
+
+# Passed through from workforce/leadership vocabulary (info-tech-canon). We
+# carry these; we do not author or interpret them.
+EXECUTION_REF_KEYS: tuple[str, ...] = (
+    "correlation_id",
+    "assignment_ref",
+    "role_ref",
+    "duty_ref",
+    "goal_refs",
+    "resource_envelope_refs",
+)
+_LIST_REF_KEYS = frozenset({"goal_refs", "resource_envelope_refs"})
+
+
+class ProfileRefError(ValueError):
+    """A harness profile reference is absent or structurally unusable."""
+
+
+def require_harness_profile() -> bool:
+    """Whether a governed ops_run must carry a profile ref.
+
+    Off during migration (ACT-ADR-006 coexistence): definitions adopt the ref
+    one at a time. Turn on once inventory shows no caller depends on
+    ``approach_hint`` for runtime selection.
+    """
+    return os.getenv("ACTIVITY_CORE_REQUIRE_HARNESS_PROFILE", "").strip().lower() in {
+        "1",
+        "true",
+        "yes",
+        "on",
+    }
+
+
+def validate_profile_ref(raw: Any) -> str:
+    """Return a normalised, pinned profile ref or raise ProfileRefError."""
+    if raw is None or not isinstance(raw, str) or not raw.strip():
+        raise ProfileRefError("harness_profile_ref is empty")
+
+    ref = raw.strip()
+    if any(character.isspace() for character in ref):
+        raise ProfileRefError(f"harness_profile_ref contains whitespace: {ref!r}")
+
+    profile_id, separator, version = ref.partition("@")
+    if not separator:
+        # Glas would match every version of this id and refuse as ambiguous.
+        raise ProfileRefError(
+            f"harness_profile_ref must pin a version as '<id>@<version>': {ref!r}"
+        )
+    if not _PROFILE_ID_PATTERN.match(profile_id):
+        raise ProfileRefError(f"harness_profile_ref has an invalid id: {ref!r}")
+    if not _PROFILE_VERSION_PATTERN.match(version):
+        raise ProfileRefError(f"harness_profile_ref has an invalid version: {ref!r}")
+    return ref
+
+
+def resolve_execution_selector(
+    harness_profile_ref: Any,
+    approach_hint: Any = None,
+) -> tuple[str | None, str | None]:
+    """Return ``(harness_profile_ref, approach_hint)`` for a queued ops_run.
+
+    ACT-ADR-006 §2: the two coexist with **distinct semantics**. The profile ref
+    is the authoritative execution-constellation selector; ``approach_hint`` is
+    a legacy definition-matching hint only. It must never override, synthesize,
+    or fall back from an absent or invalid profile ref — a silent fallback would
+    reintroduce exactly the claim-time routing failure this replaces.
+
+    So the hint is carried through untouched and is never consulted here.
+    """
+    hint = approach_hint if isinstance(approach_hint, str) and approach_hint.strip() else None
+
+    if harness_profile_ref is None or (
+        isinstance(harness_profile_ref, str) and not harness_profile_ref.strip()
+    ):
+        if require_harness_profile():
+            raise ProfileRefError(
+                "governed ops_run has no harness_profile_ref "
+                "(approach_hint cannot substitute for one)"
+            )
+        return None, hint
+
+    # Malformed refs always raise, migration flag or not: a ref that was meant
+    # to route this run and cannot must not silently degrade to the legacy hint.
+    return validate_profile_ref(harness_profile_ref), hint
+
+
+def normalise_execution_refs(raw: Any) -> dict[str, Any]:
+    """Keep only the contract's attribution refs, as strings / lists of strings."""
+    if not isinstance(raw, dict):
+        return {}
+
+    refs: dict[str, Any] = {}
+    for key in EXECUTION_REF_KEYS:
+        value = raw.get(key)
+        if key in _LIST_REF_KEYS:
+            if isinstance(value, (list, tuple)):
+                items = [str(item).strip() for item in value if str(item).strip()]
+                if items:
+                    refs[key] = items
+        elif isinstance(value, str) and value.strip():
+            refs[key] = value.strip()
+    return refs

src/activity_core/ops_run_queue.py

@@ -11,6 +11,10 @@ from sqlalchemy import Select, and_, func, select, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from activity_core.glas_profile import (
+    normalise_execution_refs,
+    resolve_execution_selector,
+)
 from activity_core.orm import OpsRun
 from activity_core.rules.models import TaskSpec
 
@@ -92,6 +96,8 @@ def ops_run_to_dict(row: OpsRun) -> dict[str, Any]:
         "source_id": row.source_id,
         "triggering_event_id": row.triggering_event_id,
         "approach_hint": row.approach_hint,
+        "harness_profile_ref": row.harness_profile_ref,
+        "execution_refs": dict(row.execution_refs or {}),
         "result": dict(row.result or {}),
         "created_at": row.created_at.isoformat() if row.created_at else None,
         "updated_at": row.updated_at.isoformat() if row.updated_at else None,
@@ -103,6 +109,8 @@ async def create_ops_run_from_spec(
     spec: TaskSpec,
     *,
     approach_hint: str | None = None,
+    harness_profile_ref: str | None = None,
+    execution_refs: dict | None = None,
 ) -> uuid.UUID | None:
     """Insert ops_run if queue enabled; return id or None if disabled/duplicate."""
     if not ops_run_queue_enabled():
@@ -115,6 +123,10 @@ async def create_ops_run_from_spec(
     except ValueError:
         return None
 
+    # ACT-ADR-006: raises on a malformed ref, and on an absent one once
+    # ACTIVITY_CORE_REQUIRE_HARNESS_PROFILE is enabled. Never falls back to the hint.
+    profile_ref, hint = resolve_execution_selector(harness_profile_ref, approach_hint)
+
     key = build_idempotency_key(spec)
     now = datetime.now(timezone.utc)
     stmt = (
@@ -133,7 +145,9 @@ async def create_ops_run_from_spec(
             source_type=spec.source_type or "rule",
             source_id=spec.source_id or "",
             triggering_event_id=spec.triggering_event_id or "",
-            approach_hint=approach_hint,
+            approach_hint=hint,
+            harness_profile_ref=profile_ref,
+            execution_refs=normalise_execution_refs(execution_refs),
             result={},
             created_at=now,
             updated_at=now,

src/activity_core/orm.py

@@ -170,6 +170,12 @@ class OpsRun(Base):
     source_id: Mapped[str] = mapped_column(Text, nullable=False)
     triggering_event_id: Mapped[str] = mapped_column(Text, nullable=False, index=True)
     approach_hint: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # ACT-ADR-006: authoritative execution-constellation selector. Legacy
+    # `approach_hint` above must never substitute for an absent/invalid one.
+    harness_profile_ref: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # Attribution refs carried through from workforce/leadership vocabulary;
+    # activity-core does not author or interpret them.
+    execution_refs: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
     result: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()

src/activity_core/run_artifacts.py

@@ -153,6 +153,8 @@ def _ops_summary(row: OpsRun) -> dict[str, Any]:
         "triggering_event_id": row.triggering_event_id,
         "source_id": row.source_id,
         "approach_hint": row.approach_hint,
+        "harness_profile_ref": row.harness_profile_ref,
+        "execution_refs": dict(row.execution_refs or {}),
         "created_at": row.created_at.isoformat() if row.created_at else None,
         "updated_at": row.updated_at.isoformat() if row.updated_at else None,
         "result": {