session-memory: scoping + proposals + active registry (WP-0007 T04)

distribute/proposals.py: Scope-aware targeting (FR-X2, empty axis = any), render
distributable (approved+distribution_ready) patterns into a proposals/ tree
mirroring target paths — proposed not applied (FR-X3, HITL), idempotent on re-run.
ActiveRegistry (FR-X4) records which pattern+version is proposed in which
(repo,flavor). 6 new tests; suite 123/123.

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

session_memory/distribute/proposals.py

@@ -0,0 +1,136 @@
+"""Scoping, proposed-not-applied output, and the active-pattern registry
+(PRD §6.4 FR-X2/FR-X3/FR-X4; T04).
+
+* **Scope (FR-X2):** a pattern lands in a target environment only if the target's
+  repo/domain/flavor are within the pattern's :class:`Scope` (an empty scope list
+  means "unrestricted on that axis").
+* **Proposed, not applied (FR-X3):** rendered artifacts are written under a
+  ``proposals/`` tree mirroring the target path — a reviewable diff a human applies,
+  never auto-written into the live file. Re-running upserts each pattern's block in
+  place (idempotent), so proposals don't accumulate duplicates.
+* **Active-pattern registry (FR-X4):** a JSON record of which pattern (and version)
+  is proposed/active in which (repo, flavor) environment.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from ..curate.schema import SolutionPattern
+from .base import upsert_block
+from .registry import get_distributor
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+@dataclass(frozen=True)
+class Target:
+    """An environment a pattern could be distributed to."""
+
+    repo: str
+    domain: str = ""
+    flavor: str = "claude"
+
+
+def applies(pattern: SolutionPattern, target: Target) -> bool:
+    """True if ``target`` is within the pattern's scope (empty axis == any)."""
+    sc = pattern.scope
+    if sc.repos and target.repo not in sc.repos:
+        return False
+    if sc.domains and target.domain and target.domain not in sc.domains:
+        return False
+    if sc.flavors and target.flavor not in sc.flavors:
+        return False
+    return True
+
+
+def is_distributable(pattern: SolutionPattern) -> bool:
+    return pattern.status == "approved" and pattern.distribution_ready
+
+
+class ActiveRegistry:
+    """JSON record of patterns proposed/active per (repo, flavor) — FR-X4."""
+
+    def __init__(self, path: str) -> None:
+        self.path = path
+        self._entries: dict[str, dict] = {}
+        if os.path.exists(path):
+            with open(path, encoding="utf-8") as fh:
+                for e in json.load(fh):
+                    self._entries[self._key(e["pattern_id"], e["repo"], e["flavor"])] = e
+
+    @staticmethod
+    def _key(pid: str, repo: str, flavor: str) -> str:
+        return f"{pid}|{repo}|{flavor}"
+
+    def record(self, pid: str, repo: str, flavor: str, version: str,
+               status: str = "proposed") -> None:
+        self._entries[self._key(pid, repo, flavor)] = {
+            "pattern_id": pid, "repo": repo, "flavor": flavor,
+            "version": version, "status": status, "updated_at": _now(),
+        }
+
+    def entries(self) -> list[dict]:
+        return [self._entries[k] for k in sorted(self._entries)]
+
+    def save(self) -> None:
+        os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)
+        with open(self.path, "w", encoding="utf-8") as fh:
+            json.dump(self.entries(), fh, indent=2, sort_keys=True)
+            fh.write("\n")
+
+
+@dataclass
+class ProposalResult:
+    proposals: list = None        # (repo, flavor, pattern_id, proposal_path)
+    files_written: list = None    # absolute proposal paths
+    skipped_not_distributable: list = None  # pattern ids
+
+    def __post_init__(self):
+        self.proposals = self.proposals or []
+        self.files_written = self.files_written or []
+        self.skipped_not_distributable = self.skipped_not_distributable or []
+
+
+def propose(patterns: list[SolutionPattern], targets: list[Target], out_dir: str,
+            registry: ActiveRegistry) -> ProposalResult:
+    """Render in-scope, distributable patterns into per-target proposal files."""
+    result = ProposalResult()
+    pending: dict[str, str] = {}  # proposal path -> accumulated content
+
+    for p in patterns:
+        if not is_distributable(p):
+            result.skipped_not_distributable.append(p.id)
+            continue
+        for t in targets:
+            dist = get_distributor(t.flavor)
+            if dist is None or not applies(p, t):
+                continue
+            art = dist.render(p)
+            path = os.path.join(out_dir, t.repo, art.target_path)
+            if path not in pending:
+                pending[path] = _read(path)
+            pending[path] = upsert_block(pending[path], p.id, art.content)
+            registry.record(p.id, t.repo, t.flavor, p.version)
+            result.proposals.append((t.repo, t.flavor, p.id, path))
+
+    for path, content in pending.items():
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "w", encoding="utf-8") as fh:
+            fh.write(content if content.endswith("\n") else content + "\n")
+        result.files_written.append(path)
+
+    registry.save()
+    return result
+
+
+def _read(path: str) -> str:
+    if os.path.exists(path):
+        with open(path, encoding="utf-8") as fh:
+            return fh.read()
+    return ""

tests/test_distribute_proposals.py

@@ -0,0 +1,79 @@
+"""Scoping + proposals + active registry tests (WP-0007 T04)."""
+
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from session_memory.curate.schema import Resolution, Scope, SolutionPattern  # noqa: E402
+from session_memory.distribute.proposals import (  # noqa: E402
+    ActiveRegistry,
+    Target,
+    applies,
+    propose,
+)
+
+
+def _pattern(pid="sp-x", repos=None, flavors=None, status="approved", ready=True):
+    return SolutionPattern(
+        id=pid, name="Read before edit", version="1.0.0", polarity="problem",
+        problem="edit before read", resolutions=[Resolution(summary="read first")],
+        scope=Scope(repos=repos or [], flavors=flavors or []),
+        status=status, distribution_ready=ready,
+    )
+
+
+def test_applies_respects_scope():
+    p = _pattern(repos=["agentic-resources"], flavors=["claude"])
+    assert applies(p, Target("agentic-resources", flavor="claude"))
+    assert not applies(p, Target("other-repo", flavor="claude"))
+    assert not applies(p, Target("agentic-resources", flavor="codex"))
+
+
+def test_empty_scope_is_unrestricted():
+    assert applies(_pattern(), Target("any", flavor="grok"))
+
+
+def test_propose_writes_scoped_proposal_files(tmp_path):
+    out = str(tmp_path / "proposals")
+    reg = ActiveRegistry(str(tmp_path / "active.json"))
+    p = _pattern(flavors=["claude"])
+    res = propose([p], [Target("agentic-resources", flavor="claude"),
+                        Target("agentic-resources", flavor="codex")], out, reg)
+    # only claude target is in scope
+    assert len(res.proposals) == 1
+    path = os.path.join(out, "agentic-resources", "CLAUDE.md")
+    assert os.path.exists(path)
+    assert "BEGIN helix-forge pattern:sp-x" in open(path).read()
+
+
+def test_not_distributable_skipped(tmp_path):
+    reg = ActiveRegistry(str(tmp_path / "active.json"))
+    prov = _pattern(status="provisional", ready=False)
+    res = propose([prov], [Target("r", flavor="claude")], str(tmp_path / "p"), reg)
+    assert res.proposals == []
+    assert "sp-x" in res.skipped_not_distributable
+
+
+def test_proposals_idempotent_on_rerun(tmp_path):
+    out = str(tmp_path / "proposals")
+    reg_path = str(tmp_path / "active.json")
+    p = _pattern()
+    propose([p], [Target("r", flavor="claude")], out, ActiveRegistry(reg_path))
+    propose([p], [Target("r", flavor="claude")], out, ActiveRegistry(reg_path))
+    content = open(os.path.join(out, "r", "CLAUDE.md")).read()
+    assert content.count("BEGIN helix-forge pattern:sp-x") == 1  # no duplication
+
+
+def test_active_registry_records_environment(tmp_path):
+    reg_path = str(tmp_path / "active.json")
+    reg = ActiveRegistry(reg_path)
+    propose([_pattern()], [Target("r", domain="helix_forge", flavor="claude")],
+            str(tmp_path / "p"), reg)
+    reg2 = ActiveRegistry(reg_path)  # reload from disk
+    entries = reg2.entries()
+    assert len(entries) == 1
+    assert entries[0]["pattern_id"] == "sp-x"
+    assert entries[0]["repo"] == "r"
+    assert entries[0]["flavor"] == "claude"
+    assert entries[0]["status"] == "proposed"

workplans/AGENTIC-WP-0007-session-memory-phase3.md

@@ -71,7 +71,7 @@ expressible for all. Unit-tested.
 
 ```task
 id: AGENTIC-WP-0007-T04
-status: todo
+status: done
 priority: high
 state_hub_task_id: "2c690f29-2aee-460a-b9cd-3566018f6b3c"
 ```