diff --git a/src/session_recall/share/approval.py b/src/session_recall/share/approval.py
index 21fffd6..32d8701 100644
--- a/src/session_recall/share/approval.py
+++ b/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")
diff --git a/src/session_recall/share/cli.py b/src/session_recall/share/cli.py
index 2a7ca70..a2801bd 100644
--- a/src/session_recall/share/cli.py
+++ b/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))
diff --git a/src/session_recall/share/compose.py b/src/session_recall/share/compose.py
index 760b4b5..9c0d9ec 100644
--- a/src/session_recall/share/compose.py
+++ b/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"{t['text']}" for t in turns]
+ return ("\n" + "\n".join(lines) +
+ "\n\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 []) +
"\n"
f"What they are doing: {req.get('task', '(not stated)')}\n"
f"Problem and symptoms: {req.get('problem', '(not stated)')}\n"
@@ -71,13 +88,51 @@ 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
@@ -85,7 +140,7 @@ def make_composer(env: dict | None = None, client=None) -> Composer | None:
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
diff --git a/src/session_recall/share/envelope.py b/src/session_recall/share/envelope.py
index a982685..c085fb4 100644
--- a/src/session_recall/share/envelope.py
+++ b/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)
diff --git a/src/session_recall/share/notify.py b/src/session_recall/share/notify.py
index 6ba8cd2..41fbece 100644
--- a/src/session_recall/share/notify.py
+++ b/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 ` or `/no `"
+_USAGE = ("usage: reply to a preview with `/ok `, `/no `, "
+ "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:
diff --git a/src/session_recall/share/thread.py b/src/session_recall/share/thread.py
new file mode 100644
index 0000000..0912c25
--- /dev/null
+++ b/src/session_recall/share/thread.py
@@ -0,0 +1,108 @@
+"""Conversations: a question and everything that followed it.
+
+One-shot Q&A forces the asker to state the perfect question first time and gives
+the owner no way to say "which OS?" or to answer from memory when the index has
+nothing. A thread fixes both — and the owner writing a reply themselves is the
+most valuable path of all, because often the real answer was never in the index.
+
+The thread id lives inside the *encrypted* body, not the signed envelope, so the
+relay learns nothing about conversation structure — only that some address got
+mail. Each side keeps its own log; there is no shared server-side state.
+
+Approval rule, and the reason the log records who authored each turn:
+a machine-drafted answer needs /ok, while a message the owner typed is already
+authorized by the act of typing it. Authorship is the gate.
+"""
+
+import json
+import os
+import secrets
+import stat
+import time
+from dataclasses import dataclass, field, asdict
+from pathlib import Path
+
+THREADS_DIR = "threads"
+MAX_TURNS = 40 # a thread this long has become a channel, not a question
+IDLE_CLOSE_S = 14 * 86400
+
+
+@dataclass
+class Turn:
+ author: str # "peer" | "owner" | "worker"
+ text: str
+ ts: float
+ kind: str = "message" # message | question | answer | decline
+
+
+@dataclass
+class Thread:
+ id: str
+ peer_address: str
+ peer_name: str
+ created_at: float
+ turns: list = field(default_factory=list)
+ closed: bool = False
+
+ def append(self, author: str, text: str, kind: str = "message") -> None:
+ self.turns.append(asdict(Turn(author=author, text=text, ts=time.time(),
+ kind=kind)))
+
+ def context(self, limit: int = 6) -> list:
+ """Recent turns for the composer. Untrusted every round: an injection
+ planted in turn 1 is still sitting here on turn 9, so callers must fence
+ this the same way they fence a fresh request."""
+ return self.turns[-limit:]
+
+ def should_close(self, now: float | None = None) -> bool:
+ now = time.time() if now is None else now
+ if len(self.turns) >= MAX_TURNS:
+ return True
+ last = self.turns[-1]["ts"] if self.turns else self.created_at
+ return now - last > IDLE_CLOSE_S
+
+
+def new_id() -> str:
+ return secrets.token_hex(6)
+
+
+def _path(share_dir: Path, thread_id: str) -> Path:
+ return share_dir / THREADS_DIR / f"{thread_id}.json"
+
+
+def save(share_dir: Path, thread: Thread) -> None:
+ d = share_dir / THREADS_DIR
+ d.mkdir(parents=True, exist_ok=True)
+ d.chmod(0o700)
+ path = _path(share_dir, thread.id)
+ fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
+ stat.S_IRUSR | stat.S_IWUSR)
+ with os.fdopen(fd, "w") as f:
+ json.dump(asdict(thread), f, indent=2, ensure_ascii=False)
+
+
+def load(share_dir: Path, thread_id: str) -> Thread | None:
+ path = _path(share_dir, thread_id)
+ if not path.exists():
+ return None
+ return Thread(**json.loads(path.read_text()))
+
+
+def open_or_create(share_dir: Path, thread_id: str, peer_address: str,
+ peer_name: str) -> Thread:
+ """Threads are created by whichever side speaks first; the other side
+ materialises the same id on receipt."""
+ existing = load(share_dir, thread_id)
+ if existing is not None:
+ return existing
+ return Thread(id=thread_id, peer_address=peer_address, peer_name=peer_name,
+ created_at=time.time())
+
+
+def listing(share_dir: Path) -> list[Thread]:
+ d = share_dir / THREADS_DIR
+ if not d.is_dir():
+ return []
+ threads = [Thread(**json.loads(p.read_text())) for p in d.glob("*.json")]
+ return sorted(threads, key=lambda t: t.turns[-1]["ts"] if t.turns
+ else t.created_at, reverse=True)
diff --git a/src/session_recall/share/worker.py b/src/session_recall/share/worker.py
index 6505759..7454fbb 100644
--- a/src/session_recall/share/worker.py
+++ b/src/session_recall/share/worker.py
@@ -30,6 +30,7 @@
from nacl.encoding import RawEncoder
from nacl import hash as nacl_hash
+from . import thread as thread_mod
from .ask import retrieval_query
from .crypto import canonical
from .envelope import Incoming, ShareState, open_incoming
@@ -62,6 +63,7 @@ class Candidate:
version: str = "" # /ok must quote this — approval of an exact blob
problem: str = "" # symptoms the asker reported; last for compatibility
composed: bool = False # True when an LLM wrote `text` from the fragments
+ thread: str = "" # conversation this answer belongs to
def compute_version(self) -> str:
digest = nacl_hash.blake2b(
@@ -95,7 +97,8 @@ def _digest(chunks: list) -> str:
def build_candidate(incoming: Incoming, searcher: Searcher,
- allowed_projects: list[str], composer=None) -> Candidate:
+ allowed_projects: list[str], composer=None,
+ turns: list | None = None) -> Candidate:
body = {"question": str(incoming.body.get("question", ""))[:2000],
"task": str(incoming.body.get("task", ""))[:500],
"problem": str(incoming.body.get("problem", ""))[:1000]}
@@ -111,15 +114,27 @@ def build_candidate(incoming: Incoming, searcher: Searcher,
text = ("(nothing found within shareable scope — "
"see `session-recall share allow`)")
else:
- written = composer(body, chunks) if composer else None
+ written = composer(body, chunks, turns) if composer else None
text, composed = (written, True) if written else (_digest(chunks), False)
+ # the scanner runs on the sources too: a composed answer can look clean
+ # while the model was reading secret-adjacent material, and the owner
+ # should know to open that one locally
+ findings = [asdict(f) for f in scan(text)]
+ seen = {(f["kind"], f["excerpt"]) for f in findings}
+ for c in chunks:
+ for f in scan(c["snippet"]):
+ if (f.kind, f.excerpt) not in seen:
+ seen.add((f.kind, f.excerpt))
+ findings.append({**asdict(f), "in": "source"})
+
cand = Candidate(
id=secrets.token_hex(4), peer_name=incoming.peer.name,
peer_address=incoming.peer.address, question=body["question"],
task=body["task"], problem=body["problem"],
reply_nonce=incoming.nonce, created_at=time.time(), text=text,
- chunks=chunks, findings=[asdict(f) for f in scan(text)], composed=composed)
+ chunks=chunks, findings=findings, composed=composed,
+ thread=str(incoming.body.get("thread", ""))[:32])
cand.version = cand.compute_version()
return cand
@@ -134,8 +149,18 @@ def poll_once(identity: Identity, trust: TrustStore, state: ShareState,
incoming = open_incoming(identity, trust, state, raw)
if incoming is None or incoming.kind != "req":
continue
+ thread_id = str(incoming.body.get("thread", ""))[:32] or thread_mod.new_id()
+ convo = thread_mod.open_or_create(share_dir, thread_id,
+ incoming.peer.address, incoming.peer.name)
+ if convo.closed or convo.should_close():
+ convo.closed = True # a thread this old is a standing channel
+ thread_mod.save(share_dir, convo)
+ continue
cand = build_candidate(incoming, searcher, trust.allowed_projects(),
- composer=composer)
+ composer=composer, turns=convo.context())
+ cand.thread = thread_id
+ convo.append("peer", cand.question, kind="question")
+ thread_mod.save(share_dir, convo)
_write_candidate(share_dir, cand)
out.append(cand)
return out
diff --git a/tests/test_share_envelope.py b/tests/test_share_envelope.py
index 804567b..f3b5615 100644
--- a/tests/test_share_envelope.py
+++ b/tests/test_share_envelope.py
@@ -34,7 +34,7 @@ def test_request_roundtrip(world):
assert got is not None
assert got.kind == "req"
assert got.body == {"question": "how did you fix the CI?", "task": "debug",
- "problem": ""}
+ "problem": "", "thread": ""}
assert got.peer.name == "egor"
@@ -46,7 +46,7 @@ def test_response_roundtrip(world):
egor_state = ShareState(maxim_trust.path.parent / "egor-state.json")
got_resp = open_incoming(egor, egor_trust, egor_state, resp)
assert got_resp.kind == "resp"
- assert got_resp.body == {"text": "the answer"}
+ assert got_resp.body == {"text": "the answer", "thread": "", "sources": 0}
assert got_resp.in_reply_to == got.nonce
diff --git a/tests/test_share_threads.py b/tests/test_share_threads.py
new file mode 100644
index 0000000..9b80372
--- /dev/null
+++ b/tests/test_share_threads.py
@@ -0,0 +1,144 @@
+"""Conversations: follow-ups without re-asking, and the owner answering in
+their own words — which is the whole point, because the real answer is often
+not in the index at all."""
+
+from dataclasses import dataclass
+
+import pytest
+
+from session_recall.share import approval, compose, thread as thread_mod
+from session_recall.share import identity as identity_mod
+from session_recall.share.envelope import ShareState, make_request, open_incoming
+from session_recall.share.transport import InMemoryTransport
+from session_recall.share.trust import Peer, TrustStore
+from session_recall.share.worker import load_candidate, poll_once
+
+
+@dataclass
+class FakeAnchor:
+ session_id: str = "sess-1234567890"
+ uuid: str = "u1"
+ role: str = "assistant"
+ snippet: str = "запинили mcp<2 в PR #13"
+ score: float = 0.9
+ project: str = "session-recall"
+ when: int = 1785000000
+ source: str = "claude"
+
+
+@pytest.fixture
+def world(tmp_path):
+ maxim = identity_mod.create(tmp_path / "maxim", "maxim")
+ egor = identity_mod.create(tmp_path / "egor", "egor")
+ mt = TrustStore(tmp_path / "maxim" / "trust.json")
+ b = egor.public_bundle()
+ mt.add(Peer(name=b["name"], address=b["address"],
+ sign_pk=b["sign_pk"], box_pk=b["box_pk"]))
+ mt.allow_project("session-recall")
+ et = TrustStore(tmp_path / "egor" / "trust.json")
+ mb = maxim.public_bundle()
+ et.add(Peer(name=mb["name"], address=mb["address"],
+ sign_pk=mb["sign_pk"], box_pk=mb["box_pk"]))
+ return {"maxim": maxim, "egor": egor, "mt": mt, "et": et,
+ "state": ShareState(tmp_path / "maxim" / "state.json"),
+ "transport": InMemoryTransport(),
+ "mdir": tmp_path / "maxim", "edir": tmp_path / "egor"}
+
+
+def _ask(w, question="как чинили CI?", thread=""):
+ w["transport"].post_mail(w["maxim"].address, make_request(
+ w["egor"], w["maxim"].public_bundle(), question,
+ task="поднимаю relay", problem="ModuleNotFoundError", thread=thread))
+ return poll_once(w["maxim"], w["mt"], w["state"], w["transport"],
+ lambda q, k: [FakeAnchor()], w["mdir"])
+
+
+def test_first_ask_opens_a_thread(world):
+ cand = _ask(world)[0]
+ assert cand.thread
+ convo = thread_mod.load(world["mdir"], cand.thread)
+ assert convo.peer_name == "egor"
+ assert convo.turns[0]["author"] == "peer"
+
+
+def test_follow_up_joins_the_same_thread(world):
+ first = _ask(world)[0]
+ second = _ask(world, "а пин или миграция?", thread=first.thread)[0]
+ assert second.thread == first.thread
+ convo = thread_mod.load(world["mdir"], first.thread)
+ assert len(convo.turns) == 2
+
+
+def test_history_reaches_the_composer(world):
+ first = _ask(world)[0]
+ seen = {}
+ world["transport"].post_mail(world["maxim"].address, make_request(
+ world["egor"], world["maxim"].public_bundle(), "а точнее?",
+ thread=first.thread))
+
+ def composer(req, chunks, turns=None):
+ seen["turns"] = turns
+ return "ответ с учётом предыдущего"
+
+ poll_once(world["maxim"], world["mt"], world["state"], world["transport"],
+ lambda q, k: [FakeAnchor()], world["mdir"], composer=composer)
+ assert seen["turns"], "the composer must see what was already asked"
+ assert "как чинили CI?" in seen["turns"][0]["text"]
+
+
+def test_owner_can_answer_in_their_own_words(world):
+ """No approval step — typing it IS the authorization."""
+ cand = _ask(world)[0]
+ ok = approval.own_message(world["maxim"], world["mt"], world["mdir"],
+ world["transport"], cand.thread,
+ "там дело было не в пине, а в версии ноды")
+ assert ok
+ inbox = world["transport"].fetch_mail(world["egor"].address)
+ got = open_incoming(world["egor"], world["et"],
+ ShareState(world["edir"] / "state.json"), inbox[0])
+ assert got.body["text"].startswith("там дело было")
+ assert got.body["thread"] == cand.thread
+
+
+def test_owner_message_fails_closed_on_revoked_peer(world):
+ cand = _ask(world)[0]
+ world["mt"].revoke("egor")
+ assert approval.own_message(world["maxim"], world["mt"], world["mdir"],
+ world["transport"], cand.thread, "ответ") is False
+ assert world["transport"].fetch_mail(world["egor"].address) == []
+
+
+def test_closed_thread_refuses_new_turns(world):
+ cand = _ask(world)[0]
+ convo = thread_mod.load(world["mdir"], cand.thread)
+ convo.closed = True
+ thread_mod.save(world["mdir"], convo)
+ assert _ask(world, "ещё вопрос", thread=cand.thread) == []
+ assert approval.own_message(world["maxim"], world["mt"], world["mdir"],
+ world["transport"], cand.thread, "ответ") is False
+
+
+def test_thread_closes_after_too_many_turns(world):
+ convo = thread_mod.Thread(id="abc", peer_address="x", peer_name="egor",
+ created_at=0.0)
+ for i in range(thread_mod.MAX_TURNS):
+ convo.append("peer", f"q{i}")
+ assert convo.should_close()
+
+
+# -- what actually crosses the wire ------------------------------------------
+def test_response_carries_no_transcript_or_session_ids(world):
+ cand = _ask(world)[0]
+ approval.approve(world["mdir"], cand.id, cand.version)
+ approval.dispatch(world["maxim"], world["mt"], world["mdir"],
+ world["transport"])
+ inbox = world["transport"].fetch_mail(world["egor"].address)
+ got = open_incoming(world["egor"], world["et"],
+ ShareState(world["edir"] / "state.json"), inbox[0])
+ assert got.body["sources"] == 1 # grounded-vs-guess signal survives
+ assert "session_id" not in got.body and "refs" not in got.body
+ assert "sess-1234567890" not in str(got.body)
+
+
+def test_composer_is_told_not_to_cite_sessions():
+ assert "Never cite session ids" in compose._SYSTEM
diff --git a/tests/test_share_worker.py b/tests/test_share_worker.py
index 508f5e0..260fbe4 100644
--- a/tests/test_share_worker.py
+++ b/tests/test_share_worker.py
@@ -159,7 +159,7 @@ def test_composer_writes_the_answer(world):
task="поднимаю relay", problem="ModuleNotFoundError fastmcp"))
seen = {}
- def composer(req, chunks):
+ def composer(req, chunks, turns=None):
seen.update(req)
return "запинили mcp<2, смотри session-recall · 07876709"
@@ -177,7 +177,7 @@ def test_composer_failure_falls_back_to_digest(world):
_ask(egor, maxim, transport)
cand = poll_once(maxim, trust, state, transport,
lambda q, k: [_anchor("session-recall")], sdir,
- composer=lambda req, chunks: None)[0]
+ composer=lambda req, chunks, turns=None: None)[0]
assert cand.composed is False
assert "how we fixed it" in cand.text
@@ -200,7 +200,7 @@ def test_secret_in_composed_answer_still_flagged(world):
_ask(egor, maxim, transport)
cand = poll_once(maxim, trust, state, transport,
lambda q, k: [_anchor("session-recall")], sdir,
- composer=lambda req, chunks: "ключ был AKIAIOSFODNN7EXAMPLE")[0]
+ composer=lambda req, chunks, turns=None: "ключ был AKIAIOSFODNN7EXAMPLE")[0]
assert any(f["kind"] == "aws-access-key" for f in cand.findings)