feat(share): conversations, owner-authored replies, local-CLI composer

src/session_recall/share/approval.py

@@ -119,6 +119,27 @@ def preview(cand: Candidate, redact: bool | None = None,
     return "\n".join(out) if plain else "\n\n".join(out)
 
 
+def own_message(identity: Identity, trust: TrustStore, share_dir: Path,
+                transport, thread_id: str, text: str) -> bool:
+    """The owner answering in their own words. No approval step: they authored
+    it, which IS the authorization — a second confirmation on your own typing
+    is ritual, not a gate. Everything else (revoked peer, closed thread) still
+    fails closed."""
+    from . import thread as thread_mod
+    convo = thread_mod.load(share_dir, thread_id)
+    if convo is None or convo.closed or not text.strip():
+        return False
+    peer = trust.get_by_address(convo.peer_address)
+    if peer is None:
+        return False
+    transport.post_mail(peer.address,
+                        make_response(identity, peer, text.strip(),
+                                      in_reply_to="", thread=thread_id))
+    convo.append("owner", text.strip())
+    thread_mod.save(share_dir, convo)
+    return True
+
+
 def approve(share_dir: Path, cand_id: str, version: str) -> Candidate | None:
     """None unless `version` matches the stored candidate exactly."""
     cand = load_candidate(share_dir, cand_id)
@@ -169,7 +190,9 @@ def dispatch(identity: Identity, trust: TrustStore, share_dir: Path,
             continue
         transport.post_mail(peer.address,
                             make_response(identity, peer, cand.text,
-                                          in_reply_to=cand.reply_nonce))
+                                          in_reply_to=cand.reply_nonce,
+                                          thread=cand.thread,
+                                          sources=len(cand.chunks)))
         # keep the audit trail honest: a delivered decline is not an answer
         set_status(share_dir, cand.id,
                    "sent" if cand.status == "approved" else "declined-sent")

src/session_recall/share/cli.py

@@ -60,6 +60,10 @@ def add_parser(sub) -> None:
     akp.add_argument("--doing", required=True, help="what you are working on")
     akp.add_argument("--problem", required=True, help="what broke, with symptoms")
     akp.add_argument("--want", required=True, help="what you want to know")
+    rpp = ssub.add_parser("reply", help="continue a conversation you started")
+    rpp.add_argument("thread")
+    rpp.add_argument("text", nargs="+")
+    ssub.add_parser("threads", help="list conversations")
     ssub.add_parser("fetch", help="collect answers peers have sent you")
     okp = ssub.add_parser("approve", help="approve a candidate locally (no TG)")
     okp.add_argument("id")
@@ -187,13 +191,47 @@ def run(args: argparse.Namespace) -> int:
         print(f"\n--- candidate answer (v{cand.version}) ---\n{cand.text}")
         return 0
 
-    if cmd in ("ask", "fetch"):
+    if cmd == "threads":
+        from . import thread as thread_mod
+        convos = thread_mod.listing(sdir)
+        if not convos:
+            print("no conversations yet")
+            return 0
+        for t in convos:
+            mark = "closed " if t.closed else ""
+            last = t.turns[-1]["text"][:60] if t.turns else "(empty)"
+            print(f"{mark}{t.id}  {t.peer_name}  {len(t.turns)} turns  {last}")
+        return 0
+
+    if cmd in ("ask", "reply", "fetch"):
+        from . import thread as thread_mod
         from .ask import AskTooThin, validate
         from .envelope import ShareState, make_request, open_incoming
         transport = from_env(os.environ, identity=ident)
         if transport is None:
             print(_TRANSPORT_HINT)
             return 1
+
+        if cmd == "reply":
+            convo = thread_mod.load(sdir, args.thread)
+            if convo is None:
+                print(f"no conversation {args.thread!r} — see: session-recall share threads")
+                return 1
+            if convo.closed or convo.should_close():
+                print(f"conversation {convo.id} is closed — start a new ask")
+                return 1
+            peer = trust.get_by_address(convo.peer_address)
+            if peer is None:
+                print(f"{convo.peer_name} is no longer trusted")
+                return 1
+            text = " ".join(args.text).strip()
+            transport.post_mail(peer.address, make_request(
+                ident, peer, text, thread=convo.id))
+            convo.append("owner", text)
+            thread_mod.save(sdir, convo)
+            print(f"sent into {convo.id}; they approve before anything comes back")
+            return 0
+
         if cmd == "ask":
             peer = (trust.get_by_address(args.peer)
                     or next((p for p in trust.peers() if p.name == args.peer), None))

src/session_recall/share/compose.py

@@ -16,10 +16,13 @@
 """
 
 import os
+import subprocess
+import tempfile
 from typing import Callable
 
 MODEL = "claude-opus-5"
 MAX_TOKENS = 4000
+CLI_TIMEOUT_S = 180
 
 Composer = Callable[[dict, list], str | None]
 
@@ -31,11 +34,16 @@
 
 Structure the answer as:
 1. A direct answer to what they want to know, in a few sentences.
-2. "Где смотреть" / "Where to look": the specific fragments carrying the \
-detail — name the project and session id shown on each fragment, and any file, \
-PR, command or error string they contain.
+2. Anything in the fragments the asker can act on or open for themselves: pull \
+request or issue numbers, commit hashes, file paths, package versions, exact \
+commands, error strings. Quote them precisely — these are what make an answer \
+usable instead of merely reassuring.
 3. What the fragments do NOT answer, if anything, stated plainly.
 
+Never cite session ids, transcript locations, or dates of the owner's own \
+work — the asker has no access to those and they reveal nothing useful. Cite \
+artifacts that exist outside the owner's machine.
+
 Rules:
 - Ground every claim in the fragments. Never invent file names, commands, \
 versions, or outcomes that are not there.
@@ -59,9 +67,18 @@ def _fragments(chunks: list) -> str:
     return "\n".join(parts)
 
 
-def _prompt(req: dict, chunks: list) -> str:
+def _history(turns: list) -> str:
+    if not turns:
+        return ""
+    lines = [f"<turn author=\"{t['author']}\">{t['text']}</turn>" for t in turns]
+    return ("<earlier_in_this_conversation>\n" + "\n".join(lines) +
+            "\n</earlier_in_this_conversation>\n\n")
+
+
+def _prompt(req: dict, chunks: list, turns: list | None = None) -> str:
     return (
         "A colleague is asking about work recorded in these fragments.\n\n"
+        + _history(turns or []) +
         "<request>\n"
         f"What they are doing: {req.get('task', '(not stated)')}\n"
         f"Problem and symptoms: {req.get('problem', '(not stated)')}\n"
@@ -71,21 +88,59 @@ def _prompt(req: dict, chunks: list) -> str:
         "Write the answer.")
 
 
-def make_composer(env: dict | None = None, client=None) -> Composer | None:
-    """None means "no composer configured" — the caller keeps the deterministic
-    digest. `client` is injectable so tests never touch the network."""
-    env = os.environ if env is None else env
-    if (env.get("SESSION_RECALL_COMPOSE") or "none").strip().lower() != "claude":
-        return None
+def _cli_composer(runner=None) -> Composer:
+    """Compose through the locally installed `claude` CLI.
 
+    Costs nothing beyond the existing subscription and needs no API key, but the
+    CLI is a full agent by default, so the invocation strips it to text-in /
+    text-out: `--tools ""` removes every built-in tool and `--strict-mcp-config`
+    ignores every configured MCP server. The prompt goes on argv, never through
+    a shell — no shell means no quoting bug can turn retrieved text into a
+    command. The working directory is an empty temp dir so no project's
+    CLAUDE.md is discovered.
+
+    Known consequence, deliberately left to the operator: print mode writes a
+    session transcript under CLAUDE_CONFIG_DIR, which session-recall indexes by
+    default. Point CLAUDE_CONFIG_DIR at a scratch directory (or exclude it from
+    indexing) before real use, or a peer's question re-enters your own index and
+    later surfaces as if it were your own past work.
+    """
+    def run(args, cwd):
+        return subprocess.run(args, cwd=cwd, capture_output=True, text=True,
+                              timeout=CLI_TIMEOUT_S)
+
+    runner = runner or run
+
+    def compose(req: dict, chunks: list, turns: list | None = None) -> str | None:
+        if not chunks:
+            return None
+        with tempfile.TemporaryDirectory() as empty:
+            try:
+                done = runner([
+                    "claude", "-p",
+                    "--tools", "",
+                    "--strict-mcp-config",
+                    "--system-prompt", _SYSTEM,
+                    _prompt(req, chunks, turns),
+                ], empty)
+            except (OSError, subprocess.SubprocessError):
+                return None
+        if getattr(done, "returncode", 1) != 0:
+            return None
+        return (done.stdout or "").strip() or None
+
+    return compose
+
+
+def _api_composer(client=None) -> Composer | None:
     if client is None:
         try:
             import anthropic
         except ImportError:
             return None
         client = anthropic.Anthropic()
 
-    def compose(req: dict, chunks: list) -> str | None:
+    def compose(req: dict, chunks: list, turns: list | None = None) -> str | None:
         if not chunks:
             return None
         try:
@@ -95,7 +150,8 @@ def compose(req: dict, chunks: list) -> str | None:
                 betas=["server-side-fallback-2026-07-01"],
                 fallbacks="default",
                 system=_SYSTEM,
-                messages=[{"role": "user", "content": _prompt(req, chunks)}],
+                messages=[{"role": "user",
+                           "content": _prompt(req, chunks, turns)}],
             )
         except Exception:
             return None          # provider down → deterministic digest, never a gap
@@ -105,3 +161,16 @@ def compose(req: dict, chunks: list) -> str | None:
         return text.strip() or None
 
     return compose
+
+
+def make_composer(env: dict | None = None, client=None, runner=None) -> Composer | None:
+    """None means "no composer configured" — the caller keeps the deterministic
+    digest. `client`/`runner` are injectable so tests never touch the network or
+    spawn a process."""
+    env = os.environ if env is None else env
+    engine = (env.get("SESSION_RECALL_COMPOSE") or "none").strip().lower()
+    if engine in ("claude-cli", "cli"):
+        return _cli_composer(runner=runner)
+    if engine in ("claude", "api", "claude-api"):
+        return _api_composer(client=client)
+    return None

src/session_recall/share/envelope.py

@@ -98,17 +98,26 @@ def _sealed(identity: Identity, peer_box_pk: str, kind: str, body: dict,
 
 
 def make_request(identity: Identity, peer: Peer | dict, question: str,
-                 task: str = "", problem: str = "") -> bytes:
+                 task: str = "", problem: str = "", thread: str = "") -> bytes:
+    """`thread` rides inside the encrypted body on purpose: the relay must not
+    learn which envelopes belong to the same conversation."""
     box_pk = peer.box_pk if isinstance(peer, Peer) else peer["box_pk"]
     address = peer.address if isinstance(peer, Peer) else peer["address"]
     return _sealed(identity, box_pk, "req",
-                   {"question": question, "task": task, "problem": problem},
+                   {"question": question, "task": task, "problem": problem,
+                    "thread": thread},
                    address, None)
 
 
 def make_response(identity: Identity, peer: Peer, text: str,
-                  in_reply_to: str) -> bytes:
-    return _sealed(identity, peer.box_pk, "resp", {"text": text},
+                  in_reply_to: str, thread: str = "", sources: int = 0) -> bytes:
+    """Raw transcript never crosses. What goes out is the answer — the same
+    thing the owner would have said after reading their own history — plus how
+    many fragments it rests on, so the asker can tell a grounded answer from a
+    guess. Session ids and project names stay home: the asker could not read
+    them anyway, and they are exactly the metadata worth not leaking."""
+    return _sealed(identity, peer.box_pk, "resp",
+                   {"text": text, "thread": thread, "sources": sources},
                    peer.address, in_reply_to)
 
 

src/session_recall/share/notify.py

@@ -25,7 +25,8 @@
 from .worker import Searcher, poll_once
 
 _REF_RE = re.compile(r"^\[([0-9a-f]{8}) v([0-9a-f]{8})\]")
-_USAGE = "usage: reply to a preview with `/ok <version>` or `/no <reason>`"
+_USAGE = ("usage: reply to a preview with `/ok <version>`, `/no <reason>`, "
+          "or just write your own answer and it goes as-is")
 
 
 def _cand_ref(message: dict) -> tuple[str, str] | None:
@@ -60,13 +61,26 @@ def _say_preview(self, cand) -> None:
     def _handle(self, message: dict) -> None:
         text = (message.get("text") or "").strip()
         mid = message.get("message_id")
-        if not (text.startswith("/ok") or text.startswith("/no")):
+        if not text:
             return
         ref = _cand_ref(message)
         if ref is None:
-            return self._say(_USAGE, reply_to=mid)
+            # a command needs a preview to point at; anything else is noise
+            return self._say(_USAGE, reply_to=mid) if text.startswith("/") else None
         cand_id, _preview_version = ref  # the reply names the candidate…
 
+        if not text.startswith("/"):
+            # plain text replied into a thread is the owner answering in their
+            # own words — authorship is the authorization, so it just goes
+            from .worker import load_candidate
+            cand = load_candidate(self.share_dir, cand_id)
+            if cand is None or not cand.thread:
+                return self._say(f"nothing to reply to under {cand_id}", reply_to=mid)
+            ok = approval.own_message(self.identity, self.trust, self.share_dir,
+                                      self.transport, cand.thread, text)
+            return self._say(f"sent to {cand.peer_name}" if ok else
+                             "not sent: thread closed or peer revoked", reply_to=mid)
+
         if text.startswith("/ok"):
             parts = text.split()
             if len(parts) != 2: