diff --git a/frontend/src/api/snippets.ts b/frontend/src/api/snippets.ts new file mode 100644 index 0000000..6b4a025 --- /dev/null +++ b/frontend/src/api/snippets.ts @@ -0,0 +1,117 @@ +import { apiGet, apiPost, apiPatch, apiDelete } from "@/api/client"; + +/** One canonical location of a reusable thing. A snippet that unifies several + * one-offs carries several — one per call site. */ +export interface SnippetLocation { + repo: string; + path: string; + symbol: string; +} + +/** Structured fields parsed out of a snippet note (mirrors the backend + * `parse_snippet_fields` — see services/snippets.py). `repo`/`path`/`symbol` + * mirror the first location for back-compat; `locations` is the full list. */ +export interface SnippetFields { + name: string; + when_to_use: string; + signature: string; + language: string; + repo: string; + path: string; + symbol: string; + locations: SnippetLocation[]; + code: string; +} + +/** A full snippet record: the note dict plus the parsed `snippet` sub-object, + * as returned by the backend `snippet_to_dict`. */ +export interface Snippet { + id: number; + title: string; + body: string; + tags: string[]; + note_type: string; + project_id: number | null; + permission?: string; + created_at: string; + updated_at: string; + snippet: SnippetFields; + systems?: { id: number; name: string }[]; + /** Set when another user owns this record; `owner` is their username and + * `permission` your access level on it. */ + shared?: boolean; + owner?: string | null; +} + +/** Lightweight list item from the knowledge preview feed. Note: the `snippet` + * field here is a truncated *body preview* (the knowledge feed's naming), not + * the parsed fields above. */ +export interface SnippetListItem { + id: number; + title: string; + tags: string[]; + note_type: string; + snippet: string; + created_at: string; + updated_at: string; + /** Present only when the record belongs to someone else — a suggestion from + * them, not one of your own. Absent means it's yours. */ + shared?: boolean; + owner?: string | null; +} + +/** Create/update payload — discrete fields the backend serializes into the + * note (title/body/tags). Empty strings are applied; omitted keys are left + * unchanged on update. */ +export interface SnippetInput { + name: string; + code: string; + language?: string; + signature?: string; + when_to_use?: string; + locations?: SnippetLocation[]; + tags?: string[]; + project_id?: number | null; + system_ids?: number[]; + /** Create only: record it even though a near-duplicate already exists. */ + force?: boolean; +} + +export async function listSnippets( + params: { q?: string; tag?: string; projectId?: number | null } = {}, +): Promise<{ snippets: SnippetListItem[]; total: number }> { + const qs = new URLSearchParams(); + if (params.q) qs.set("q", params.q); + if (params.tag) qs.set("tag", params.tag); + if (params.projectId) qs.set("project_id", String(params.projectId)); + const query = qs.toString(); + return apiGet(`/api/snippets${query ? `?${query}` : ""}`); +} + +export async function getSnippet(id: number): Promise { + return apiGet(`/api/snippets/${id}`); +} + +export async function createSnippet(data: SnippetInput): Promise { + return apiPost("/api/snippets", data); +} + +export async function updateSnippet( + id: number, + data: Partial, +): Promise { + return apiPatch(`/api/snippets/${id}`, data); +} + +export async function deleteSnippet(id: number): Promise { + return apiDelete(`/api/snippets/${id}`); +} + +/** Unify `sourceIds` into the canonical snippet `targetId`. Returns the merged + * survivor plus `merged_ids` — the sources actually folded in and trashed. */ +export async function mergeSnippets( + targetId: number, + sourceIds: number[], +): Promise { + return apiPost(`/api/snippets/${targetId}/merge`, { source_ids: sourceIds }); +} diff --git a/frontend/src/components/AppHeader.vue b/frontend/src/components/AppHeader.vue index 06e41b3..37c567b 100644 --- a/frontend/src/components/AppHeader.vue +++ b/frontend/src/components/AppHeader.vue @@ -48,6 +48,7 @@ router.afterEach(() => { Dashboard Browse Projects + Snippets Rulebooks @@ -93,6 +94,7 @@ router.afterEach(() => { Dashboard Browse Projects + Snippets Rulebooks Shared
diff --git a/frontend/src/router/index.ts b/frontend/src/router/index.ts index 87d1771..43de4dd 100644 --- a/frontend/src/router/index.ts +++ b/frontend/src/router/index.ts @@ -69,6 +69,26 @@ const router = createRouter({ name: "note-edit", component: () => import("@/views/NoteEditorView.vue"), }, + { + path: "/snippets", + name: "snippets", + component: () => import("@/views/SnippetListView.vue"), + }, + { + path: "/snippets/new", + name: "snippet-new", + component: () => import("@/views/SnippetEditorView.vue"), + }, + { + path: "/snippets/:id", + name: "snippet-view", + component: () => import("@/views/SnippetDetailView.vue"), + }, + { + path: "/snippets/:id/edit", + name: "snippet-edit", + component: () => import("@/views/SnippetEditorView.vue"), + }, { path: "/graph", name: "graph", diff --git a/frontend/src/types/note.ts b/frontend/src/types/note.ts index 517c542..01f269f 100644 --- a/frontend/src/types/note.ts +++ b/frontend/src/types/note.ts @@ -3,7 +3,7 @@ import type { System } from "@/api/systems"; export type TaskStatus = "todo" | "in_progress" | "done" | "cancelled"; export type TaskPriority = "none" | "low" | "medium" | "high"; export type TaskKind = "work" | "plan" | "issue"; -export type NoteType = "note" | "process"; +export type NoteType = "note" | "process" | "snippet"; export interface Note { id: number; diff --git a/frontend/src/views/KnowledgeView.vue b/frontend/src/views/KnowledgeView.vue index 307813b..7d5d54d 100644 --- a/frontend/src/views/KnowledgeView.vue +++ b/frontend/src/views/KnowledgeView.vue @@ -27,6 +27,10 @@ interface KnowledgeItem { project_id: number | null; created_at: string; updated_at: string; + // Set only when another user owns this record — their suggestion, not one of + // yours. Absent means it's yours. + shared?: boolean; + owner?: string | null; // Task-specific status?: string; priority?: string; @@ -437,6 +441,11 @@ onUnmounted(() => {
{{ tag }}
+ by {{ item.owner ?? "another user" }} {{ formatDate(item.updated_at) }} @@ -839,6 +848,16 @@ onUnmounted(() => { color: var(--color-muted); } .k-card-date { font-size: 0.72rem; color: var(--color-text-secondary); white-space: nowrap; opacity: 0.7; } +/* Only rendered for a record another user owns, so an unmarked card is + unambiguously the viewer's own. */ +.shared-tag { + font-size: 0.68rem; + padding: 0.08rem 0.35rem; + border-radius: 4px; + white-space: nowrap; + background: color-mix(in srgb, var(--color-text-secondary) 15%, transparent); + color: var(--color-text-secondary); +} /* ── Task card ──────────────────────────────────────────── */ .k-card-task { diff --git a/frontend/src/views/SnippetDetailView.vue b/frontend/src/views/SnippetDetailView.vue new file mode 100644 index 0000000..6718ac5 --- /dev/null +++ b/frontend/src/views/SnippetDetailView.vue @@ -0,0 +1,342 @@ + + + + + diff --git a/frontend/src/views/SnippetEditorView.vue b/frontend/src/views/SnippetEditorView.vue new file mode 100644 index 0000000..7420b6d --- /dev/null +++ b/frontend/src/views/SnippetEditorView.vue @@ -0,0 +1,613 @@ + + + + + diff --git a/frontend/src/views/SnippetListView.vue b/frontend/src/views/SnippetListView.vue new file mode 100644 index 0000000..c5849cb --- /dev/null +++ b/frontend/src/views/SnippetListView.vue @@ -0,0 +1,618 @@ + + + + + diff --git a/plugin/.claude-plugin/plugin.json b/plugin/.claude-plugin/plugin.json index dc90135..69161cc 100644 --- a/plugin/.claude-plugin/plugin.json +++ b/plugin/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "scribe", - "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.", - "version": "0.1.12", + "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.", + "version": "0.1.16", "author": { "name": "Bryan Van Deusen" }, "mcpServers": { "scribe": { diff --git a/plugin/README.md b/plugin/README.md index b6ab7ff..9a76a44 100644 --- a/plugin/README.md +++ b/plugin/README.md @@ -7,8 +7,9 @@ instance into a first-class Claude Code extension: rulebook (the `scribe` server). - **Session-start push channel** — a `SessionStart` hook injects your always-on rules + active-project context so Scribe surfaces *without being asked*. -- **Universal process-skills** — brainstorm, systematic-debugging, TDD, - writing-plans, verification, receiving-code-review (replaces superpowers). +- **Universal process-skills** — using-scribe, writing-plans, + systematic-debugging, verification, brainstorming, reusing-code (record and + recall reusable code as snippets). Replaces superpowers. - **Your Scribe Processes as skills** — saved Processes are synced into local `~/.claude/skills/scribe-proc-*` stubs that auto-surface by relevance; the stub fetches the live procedure via `get_process`. Refreshed each session and diff --git a/plugin/hooks/scribe_static_context.md b/plugin/hooks/scribe_static_context.md index 180ecf1..dd5a313 100644 --- a/plugin/hooks/scribe_static_context.md +++ b/plugin/hooks/scribe_static_context.md @@ -37,6 +37,11 @@ for the operator's work, and as your own working memory across sessions. moment it's complete. When you **fix** something — even in passing — record it as its own issue (`create_task(kind="issue")`), not as a work-log line on an unrelated open task. +- **Reuse before rebuilding** — before writing a new helper/utility/component, + search recorded **snippets** (reusable code recorded once for recall) and + reuse the prior art instead of re-solving it; when you build something + reusable, record it with `create_snippet` (name, code, when-to-reach-for-it, + location) so a later session is offered it, not left to write it again. - Do **not** keep the operator's rules, plans, or project notes in local memory / CLAUDE.md in parallel with Scribe — Scribe holds the single copy. - **Compact at clean seams** — because you record as you go, a context diff --git a/plugin/skills/reusing-code/SKILL.md b/plugin/skills/reusing-code/SKILL.md new file mode 100644 index 0000000..afb097a --- /dev/null +++ b/plugin/skills/reusing-code/SKILL.md @@ -0,0 +1,85 @@ +--- +name: reusing-code +description: Use when you're about to write a helper, utility, hook, or reusable component — search recorded snippets FIRST so prior art is reused instead of re-solved. And the moment you build or notice something reusable, record it as a snippet so a later session finds it. Triggers on "write a util/helper", "I need a function that…", "let me add a component", or just having built something worth reusing. +--- + +# Reusing code — recall before you rebuild + +Reusable code is worth writing once. Scribe stores **snippets** — a named, +reusable function or component recorded with its language, signature, canonical +location (repo · path · symbol), a one-line *"when to reach for it,"* and the +code itself — so prior art can surface *before* it's re-written as a one-off. +Snippets are ordinary embedded notes, so a recorded one also surfaces on its own +through recall/auto-inject; this skill is the active reflex around that. + +## Before you write a new helper — search first + +- About to write a utility, hook, formatter, adapter, or a reusable component? + **Search snippets before writing it.** `list_snippets(q="…")` (or a plain + `search`) — a matching one may already exist, in this project or another. + `list_snippets` searches every project by default; that's deliberate, since a + helper you need here was quite possibly written somewhere else. Narrow with + `project_id` only when you specifically want this project's own. +- If a snippet fits, pull it in full with `get_snippet(id)` and reuse it — its + `location` points at the reference implementation. Adapt, don't re-derive. +- If auto-inject already surfaced a snippet title that looks relevant, that's + your cue to `get_snippet` it rather than start from scratch. + +## The moment you build something reusable — record it + +- Just wrote (or noticed) a helper, hook, pattern, or component worth repeating? + Record it with `create_snippet` while it's fresh: + - **name** — what it's called, e.g. `useDebouncedRef`. + - **code** — the implementation. + - **when_to_use** — one sharp line on when to reach for it. This becomes part + of the title, so it's what a later recall menu shows — make it earn the pull. + - **language**, **signature**, and **location** (`repo` / `path` / `symbol`) + so the recorded copy points back at the canonical source. + - **project_id** / **system_ids** to associate it with the work it belongs to. +- Record the *reference* implementation, not every call site — one good entry + per reusable thing. If it already exists, `update_snippet` it instead of + recording a second copy (the create gate will flag a near-duplicate anyway). + +## A shared snippet is a suggestion, not a standard + +Scribe is multi-user, so a search can return snippets other people own. Those +come back marked `shared: true` with an `owner`. + +- Read one as **that person's suggestion**, not as the way things are done here. + Judge the code on its merits before reaching for it. +- Say whose it is when you propose it — "there's a snippet from *alex* that does + this" — so the operator can weigh the source, not just the code. +- Don't treat it as the house pattern, and don't build on it at scale, without + the operator agreeing to adopt it. +- Snippets shared directly with the operator only appear when you search for + them, never in a plain `list_snippets` — so anything ambient is genuinely + theirs. + +## Keep the record honest + +A recorded snippet is offered as prior art on every matching turn, so a wrong +one costs more than a missing one. + +- Details gone stale — a renamed symbol, a moved file, a signature that's + changed? Fix it with `update_snippet`. Passing an **empty string** clears a + field, so a wrong signature or location can be removed, not just written over. +- Recorded something that turned out not to be reusable, or that no longer + exists? Retire it with `delete_snippet` — it goes to the trash and can be + restored. Don't leave it competing for attention. + +## Found the same thing in several places — unify it + +When you notice the same reusable thing recorded (or written) as several +one-offs, don't leave the duplicates competing in recall — **merge them**. +`merge_snippets(canonical_id, [other_ids])` keeps one canonical record, folds in +the others' call sites as locations, and retires the duplicates to the trash. +The result is a single entry that shows every place the thing is used — which is +exactly the signal that it was worth consolidating. This is the cure the create +gate only hints at when it blocks a near-duplicate. + +## Why this pays off + +A one-off written a second time is the cost this avoids. Recording a snippet +once — with a location and a crisp "when to use" — means the next session is +offered the prior art instead of re-solving it. Search before writing; record +what's worth reusing. diff --git a/src/scribe/app.py b/src/scribe/app.py index 7f45626..1c67cc9 100644 --- a/src/scribe/app.py +++ b/src/scribe/app.py @@ -29,6 +29,7 @@ from scribe.routes.plugin import plugin_bp from scribe.routes.trash import trash_bp from scribe.routes.dashboard import dashboard_bp from scribe.routes.systems import systems_bp +from scribe.routes.snippets import snippets_bp from scribe.mcp import mount_mcp STATIC_DIR = Path(__file__).parent / "static" @@ -91,6 +92,7 @@ def create_app() -> Quart: app.register_blueprint(trash_bp) app.register_blueprint(dashboard_bp) app.register_blueprint(systems_bp) + app.register_blueprint(snippets_bp) @app.before_request async def before_request(): diff --git a/src/scribe/mcp/server.py b/src/scribe/mcp/server.py index 44be3af..dd76214 100644 --- a/src/scribe/mcp/server.py +++ b/src/scribe/mcp/server.py @@ -209,6 +209,37 @@ get_process(name) and follow the returned prompt verbatim, including any "clarify first" steps it contains. Author a new one with create_process(title, body); edit with update_process. +Scribe also stores Snippets — reusable functions/components recorded once for +recall (note_type "snippet"): a name, language, signature, canonical location +(repo · path · symbol), a one-line "when to reach for it", and the code. They +are ordinary embedded notes, so a recorded snippet also surfaces through the +same search + proactive recall as everything else. Two reflexes: (1) before you +write a new helper/utility/component, search first (list_snippets(q=...) or +search) — reuse the prior art with get_snippet(id) instead of re-deriving a +one-off; (2) the moment you build or notice something reusable, record it with +create_snippet(name, code, when_to_use, language, signature, repo, path, symbol, +project_id, system_ids) so a later session is offered it. Make when_to_use sharp +— it becomes the title, which is what a recall menu shows. Edit an existing one +with update_snippet rather than recording a second copy; when the same reusable +thing already exists as several one-offs, unify them into one canonical record +with merge_snippets (it folds every call site in as a location and trashes the +duplicates). Keep the record honest: a snippet whose details have gone stale can +be corrected with update_snippet (an empty string clears a field), and one that +is wrong or obsolete should be retired with delete_snippet — a bad snippet keeps +being offered as prior art, which costs more than none at all. + +Scribe is multi-user, so some records belong to other people. Anything another +user owns comes back marked `shared: true` with an `owner`. Treat a shared +record as THAT PERSON'S SUGGESTION, never as the operator's settled practice: +weigh it on its merits, attribute it when you reference it, and ask before +adopting it or acting on it. This matters most for a shared Process — do not run +one as written; describe what it would do and get the operator's go-ahead. +Records shared directly with the operator are also deliberately search-only: +they surface when you look for them (pass a query), not in plain lists, so +nobody else's material arrives unasked. Editing another user's record needs an +editor or admin share from them; a read-only share is refused, and the right +answer is usually to record the operator's own version rather than to push. + When developing Scribe itself, honor its multi-user sharing ACL: scope every read and mutation of user data by owner + shares — never assume a single operator. "Works for one user" is not done. diff --git a/src/scribe/mcp/tools/__init__.py b/src/scribe/mcp/tools/__init__.py index 5e9161c..b9b8dc3 100644 --- a/src/scribe/mcp/tools/__init__.py +++ b/src/scribe/mcp/tools/__init__.py @@ -5,7 +5,7 @@ to a FastMCP instance. `register_all(mcp)` is the single entry point called from `mcp.server.build_mcp_server`. """ from scribe.mcp.tools import ( - milestones, notes, processes, projects, recent, repos, rulebooks, search, systems, tags, tasks, trash, + milestones, notes, processes, projects, recent, repos, rulebooks, search, snippets, systems, tags, tasks, trash, ) @@ -21,5 +21,6 @@ def register_all(mcp) -> None: recent.register(mcp) repos.register(mcp) processes.register(mcp) + snippets.register(mcp) rulebooks.register(mcp) trash.register(mcp) diff --git a/src/scribe/mcp/tools/processes.py b/src/scribe/mcp/tools/processes.py index 64efd3a..c54e00e 100644 --- a/src/scribe/mcp/tools/processes.py +++ b/src/scribe/mcp/tools/processes.py @@ -7,6 +7,7 @@ get_process is the fire mechanism: it returns the full prompt for Claude to run. from __future__ import annotations from scribe.mcp._context import current_user_id +from scribe.services import access as access_svc from scribe.services import knowledge as knowledge_svc from scribe.services import notes as notes_svc @@ -19,15 +20,24 @@ async def list_processes(q: str = "", tag: str = "", limit: int = 50) -> dict: tag: Filter to a single tag (optional). limit: Max results (1-100). - Returns {"processes": [{id, title, tags, preview}], "total": int}. + Returns {"processes": [{id, title, tags, preview}], "total": int}. An entry + marked `shared: true` with an `owner` is another person's procedure — treat + it as a suggestion to raise with the operator, not as their own practice. + + Searching (passing `q`) reaches processes shared directly with the operator; + the plain list deliberately doesn't, so someone else's procedure never + arrives unasked. """ uid = current_user_id() items, total = await knowledge_svc.query_knowledge( user_id=uid, note_type="process", tags=[tag] if tag else [], sort="modified", q=q or None, limit=max(1, min(limit, 100)), offset=0, ) + labelled = await access_svc.label_shared_items(uid, items) procs = [{"id": it["id"], "title": it["title"], "tags": it.get("tags", []), - "preview": it.get("snippet", "")} for it in items] + "preview": it.get("snippet", ""), + **({"shared": True, "owner": it.get("owner")} if it.get("shared") else {})} + for it in labelled] return {"processes": procs, "total": total} @@ -56,6 +66,13 @@ async def get_process(name_or_id: str) -> dict: Resolution: numeric id → exact (case-insensitive) title → substring. On an ambiguous substring match, the best (most-recent) match is returned with an `other_matches` list so you can disambiguate with the operator. + + IF THE RESULT IS MARKED `shared: true`, DO NOT FOLLOW IT VERBATIM. It is + another person's procedure (see `owner`), not one the operator wrote or + adopted. Summarise what it would do and get their go-ahead first. The + follow-it-as-written contract above applies only to the operator's own + processes — a shared one is a proposal, and running it unasked would put + someone else's judgement in charge of this session. """ uid = current_user_id() note, candidates = await notes_svc.resolve_process(uid, name_or_id) @@ -64,17 +81,28 @@ async def get_process(name_or_id: str) -> dict: out = note.to_dict() if candidates: out["other_matches"] = candidates + out.update(await access_svc.describe_provenance(uid, note)) return out async def update_process(process_id: int, title: str = "", body: str = "", tags: list[str] | None = None) -> dict: """Update a stored process. Only provided fields change — empty title/body - leave that field unchanged; pass tags to replace the tag set.""" + leave that field unchanged; pass tags to replace the tag set. + + Editing another user's process requires an editor or admin share from them; a + read-only share is not enough and says so rather than claiming not-found. + """ uid = current_user_id() - note = await notes_svc.get_note(uid, process_id) - if note is None or note.note_type != "process": + loaded = await notes_svc.get_note_for_user(uid, process_id) + note = loaded[0] if loaded else None + if note is None or note.note_type != "process" or note.deleted_at is not None: raise ValueError(f"process {process_id} not found") + if not await access_svc.can_write_note(uid, process_id): + raise ValueError( + f"process {process_id} is shared with you read-only — ask its owner " + f"for edit access, or save your own copy with create_process" + ) fields: dict = {} if title.strip(): fields["title"] = title.strip() @@ -82,8 +110,13 @@ async def update_process(process_id: int, title: str = "", body: str = "", fields["body"] = body if tags is not None: fields["tags"] = tags - updated = await notes_svc.update_note(uid, process_id, **fields) - return updated.to_dict() + # As the owner — update_note is owner-scoped and the write is authorised above. + updated = await notes_svc.update_note(note.user_id, process_id, **fields) + if updated is None: + raise ValueError(f"process {process_id} not found") + out = updated.to_dict() + out.update(await access_svc.describe_provenance(uid, updated)) + return out def register(mcp) -> None: diff --git a/src/scribe/mcp/tools/search.py b/src/scribe/mcp/tools/search.py index 12bc32b..0c317a4 100644 --- a/src/scribe/mcp/tools/search.py +++ b/src/scribe/mcp/tools/search.py @@ -10,6 +10,7 @@ from __future__ import annotations import time from scribe.mcp._context import current_user_id +from scribe.services.access import owner_names_for from scribe.services.embeddings import DEFAULT_SIMILARITY_THRESHOLD, semantic_search_notes from scribe.services.retrieval_telemetry import record_retrieval @@ -42,6 +43,10 @@ async def search( Returns: {"results": [{"id", "title", "body", "is_task", "tags", "similarity"}], "total": int} + + A result marked `shared: true` with an `owner` belongs to another user — + that person's suggestion, not the operator's own record or settled practice. + Weigh it on its merits and say whose it is when you use it. """ uid = current_user_id() limit = max(1, min(limit, 50)) @@ -50,6 +55,9 @@ async def search( raw = await semantic_search_notes( uid, q, limit=limit, is_task=is_task, project_id=project_id or None, + # An explicit search reaches everything the operator may read, including + # records shared with them one-to-one. + scope="read", ) record_retrieval( user_id=uid, source="mcp_search", query=q, @@ -57,6 +65,9 @@ async def search( project_id=project_id or None, is_task=is_task, results=raw, duration_ms=(time.perf_counter() - t0) * 1000.0, ) + owners = await owner_names_for( + {int(note.user_id) for _s, note in raw if note.user_id != uid} + ) return { "results": [ { @@ -66,6 +77,10 @@ async def search( "is_task": bool(note.is_task), "tags": list(note.tags or []), "similarity": float(score), + **( + {"shared": True, "owner": owners.get(int(note.user_id))} + if note.user_id != uid else {} + ), } for score, note in raw ], diff --git a/src/scribe/mcp/tools/snippets.py b/src/scribe/mcp/tools/snippets.py new file mode 100644 index 0000000..f2a93fe --- /dev/null +++ b/src/scribe/mcp/tools/snippets.py @@ -0,0 +1,275 @@ +"""Snippet MCP tools: record reusable functions/components for later recall. + +A snippet is a Note with note_type='snippet' (see services/snippets.py). Because +snippets are ordinary embedded notes, once recorded they surface through the same +semantic search + title-first auto-inject as everything else — so a reusable +thing recorded once can be recalled before it's re-written as a one-off. + +The tools wrap services/snippets.py, mirroring the note/process tools (dedup gate +on create, System association passthrough). +""" +from __future__ import annotations + +from scribe.mcp._context import current_user_id +from scribe.services import access as access_svc +from scribe.services import dedup as dedup_svc +from scribe.services import snippets as snippets_svc +from scribe.services import systems as systems_svc + + +async def list_snippets( + q: str = "", tag: str = "", limit: int = 50, project_id: int = 0, +) -> dict: + """List recorded snippets (reusable functions/components). + + Args: + q: Free-text search across name + body (optional). Matches on meaning as + well as wording, so describe what you need the code to DO. + tag: Filter to a single tag, e.g. a language like "python" (optional). + limit: Max results (1-100). + project_id: Narrow to one project. 0 (default) searches every project — + usually what you want, since a helper you need here may well have + been written somewhere else. + + Returns {"snippets": [{id, title, tags, preview}], "total": int}. The title + reads "name — when to reach for it"; open one in full with get_snippet(id). + + An entry marked `shared: true` with an `owner` belongs to someone else — one + person's suggestion, not settled practice here. Weigh it on its merits and + attribute it when you use it. Searching (passing `q`) also reaches snippets + shared directly with the operator; browsing without a query deliberately + does not, so those stay out of ambient results until asked for. + """ + uid = current_user_id() + items, total = await snippets_svc.list_snippets( + uid, q=q or None, tag=tag, limit=max(1, min(limit, 100)), + project_id=project_id or None, + ) + return { + "snippets": await access_svc.label_shared_items(uid, items), + "total": total, + } + + +async def create_snippet( + name: str, + code: str, + language: str = "", + signature: str = "", + when_to_use: str = "", + repo: str = "", + path: str = "", + symbol: str = "", + locations: list[dict] | None = None, + tags: list[str] | None = None, + project_id: int = 0, + system_ids: list[int] | None = None, + force: bool = False, +) -> dict: + """Record a reusable function/component so future sessions can RECALL it + instead of writing a fresh one-off. + + Reach for this the moment you build (or notice) something reusable: a helper, + a hook, a component, a pattern worth repeating. Recording it once makes it + surface automatically when a similar problem comes up later. Before writing a + new utility, search first — a snippet may already exist. + + Args: + name: Short name of the function/component, e.g. "useDebouncedRef". + code: The code itself (required). + language: Language/format, e.g. "python", "vue", "sql". Becomes a tag and + the code-fence language. + signature: One-line signature/interface, e.g. "debounce(fn, ms) -> fn". + when_to_use: One line on when to reach for it — this becomes part of the + title, so it's what a recall menu shows. Keep it sharp. + repo/path/symbol: Canonical location of the reference implementation. + locations: Several locations at once, as [{"repo","path","symbol"}, ...], + when you already know the thing lives in more than one place. Takes + precedence over the single repo/path/symbol shorthand. + tags: Extra plain-string tags (language + "snippet" are added for you). + project_id: Associate with a project (0 = no project). Snippets surface + proactively within their project; search finds them across projects. + system_ids: Ids of the project's Systems to associate this snippet with. + force: Bypass the near-duplicate gate (see below). + + Returns the created snippet (including a parsed `snippet` field), OR — when a + near-duplicate snippet already exists and force is false — {"duplicate": true, + "existing_id": ..., "message": ...} and nothing is created. When that happens + and it really is the same reusable thing found in another place, prefer + merge_snippets(existing_id, [new...]) — or record then merge — to unify them + into ONE canonical record (which then carries every call site as a location), + rather than forcing a second copy with force=true. + """ + if not (name or "").strip() or not (code or "").strip(): + raise ValueError("create_snippet requires a non-empty name and code") + uid = current_user_id() + + title = snippets_svc.compose_title(name, when_to_use) + body = snippets_svc.compose_body( + code=code, language=language, signature=signature, + when_to_use=when_to_use, repo=repo, path=path, symbol=symbol, + locations=locations, + ) + if not force: + dup = await dedup_svc.find_duplicate_note( + uid, title, body, project_id=project_id or None, + is_task=False, note_type=snippets_svc.SNIPPET_NOTE_TYPE, + ) + if dup is not None: + return dedup_svc.duplicate_response(dup, "snippet") + + note = await snippets_svc.create_snippet( + uid, name=name, code=code, language=language, signature=signature, + when_to_use=when_to_use, repo=repo, path=path, symbol=symbol, + locations=locations, tags=tags, project_id=project_id or None, + ) + if system_ids: + await systems_svc.set_record_systems(uid, note.id, system_ids) + data = snippets_svc.snippet_to_dict(note) + if system_ids: + data["systems"] = [ + s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id) + ] + return data + + +async def get_snippet(snippet_id: int) -> dict: + """Fetch a snippet by id — the full record: code, signature, location, and a + parsed `snippet` field of its structured parts. + + If the record belongs to someone else it carries `shared: true` with the + `owner` and your `permission`. Read that as ONE PERSON'S SUGGESTION, not as + established practice here: judge it on its merits, say whose it is when you + reference it, and don't adopt it as the house pattern without checking. + """ + uid = current_user_id() + note = await snippets_svc.get_snippet(uid, snippet_id) + if note is None: + raise ValueError(f"snippet {snippet_id} not found") + data = snippets_svc.snippet_to_dict(note) + data.update(await access_svc.describe_provenance(uid, note)) + return data + + +async def update_snippet( + snippet_id: int, + name: str | None = None, + code: str | None = None, + language: str | None = None, + signature: str | None = None, + when_to_use: str | None = None, + repo: str | None = None, + path: str | None = None, + symbol: str | None = None, + locations: list[dict] | None = None, + tags: list[str] | None = None, + project_id: int = 0, + system_ids: list[int] | None = None, +) -> dict: + """Update a snippet. Only the fields you pass change. + + An omitted field is left alone; an EMPTY STRING clears it — so a stale + signature, a wrong "when to use", or an obsolete location can be removed, not + just overwritten. A snippet that surfaces in recall with wrong details is + worse than none, so correcting downward has to be possible. + + Args: + locations: Replace the whole location set, as [{"repo","path","symbol"}, + ...]. Pass [] to clear every location. The single repo/path/symbol + args instead overlay onto the FIRST location, leaving the rest. + tags: Replaces the extra-tag set (language + "snippet" are re-derived). + project_id: 0 leaves it unchanged, -1 detaches it from its project, a + positive id moves it. + + Editing someone else's snippet requires an editor or admin share from them. + A read-only share is refused with a message saying so — record your own + version instead of trying to force it. + """ + uid = current_user_id() + if project_id == 0: + project = snippets_svc.UNSET + elif project_id < 0: + project = None + else: + project = project_id + + try: + note = await snippets_svc.update_snippet( + uid, snippet_id, + name=name, code=code, language=language, + signature=signature, when_to_use=when_to_use, + repo=repo, path=path, symbol=symbol, + locations=locations, tags=tags, project_id=project, + ) + except PermissionError as exc: + # Readable but not writable — surface the real reason, not "not found". + raise ValueError(str(exc)) from exc + if note is None: + raise ValueError(f"snippet {snippet_id} not found") + if system_ids is not None: + await systems_svc.set_record_systems(note.user_id, snippet_id, system_ids) + data = snippets_svc.snippet_to_dict(note) + data.update(await access_svc.describe_provenance(uid, note)) + if system_ids is not None: + data["systems"] = [ + s.to_dict() + for s in await systems_svc.list_record_systems(note.user_id, snippet_id) + ] + return data + + +async def delete_snippet(snippet_id: int) -> dict: + """Retire a snippet you recorded — it moves to the trash and is recoverable. + + Reach for this when a snippet is wrong, obsolete, or was never worth keeping. + A recorded snippet is offered as prior art on every matching turn, so a bad + one costs more than a missing one. If instead it's a duplicate of something + that should survive, prefer merge_snippets — that keeps the call sites. + """ + uid = current_user_id() + if not await snippets_svc.delete_snippet(uid, snippet_id): + raise ValueError(f"snippet {snippet_id} not found") + return {"deleted": True, "id": snippet_id} + + +async def merge_snippets(target_id: int, source_ids: list[int]) -> dict: + """Unify duplicate/variant snippets INTO one canonical record — the cure for + the same reusable thing recorded as several one-offs. + + Keeps the target as the canonical: its name, when-to-use, signature, language + and code win. The sources' locations and extra tags are folded in — so the + survivor ends up carrying EVERY call site as a location (which is itself the + "this is duplicated N times" signal) — and the source records are moved to the + trash (recoverable). The survivor is re-embedded so recall stops surfacing the + now-merged duplicates. + + Args: + target_id: The snippet to keep (the canonical record). + source_ids: Snippet ids to fold into the target and retire. Ids that + aren't your snippets are skipped; target_id in the list is ignored. + + Returns the merged canonical snippet (with a parsed `snippet` field) plus + `merged_ids` — the source ids actually merged and trashed. + """ + uid = current_user_id() + ids = [s for s in (source_ids or []) if s != target_id] + if not ids: + raise ValueError("merge_snippets requires at least one other source_id") + try: + result = await snippets_svc.merge_snippets(uid, target_id, ids) + except PermissionError as exc: + raise ValueError(str(exc)) from exc + if result is None: + raise ValueError(f"snippet {target_id} not found") + note, merged_ids = result + data = snippets_svc.snippet_to_dict(note) + data["merged_ids"] = merged_ids + return data + + +def register(mcp) -> None: + for fn in ( + list_snippets, create_snippet, get_snippet, update_snippet, + delete_snippet, merge_snippets, + ): + mcp.tool(name=fn.__name__)(fn) diff --git a/src/scribe/routes/knowledge.py b/src/scribe/routes/knowledge.py index 901693f..3127c1a 100644 --- a/src/scribe/routes/knowledge.py +++ b/src/scribe/routes/knowledge.py @@ -5,6 +5,7 @@ from quart import Blueprint, jsonify, request from scribe.auth import get_current_user_id, login_required from scribe.routes.utils import parse_pagination +from scribe.services.access import label_shared_items logger = logging.getLogger(__name__) @@ -54,7 +55,9 @@ async def list_knowledge(): ) return jsonify({ - "items": items, + # Mark rows another user owns: this feed can be mixed-ownership, and an + # unmarked card reads as one the viewer wrote. + "items": await label_shared_items(uid, items), "total": total, "page": page, "per_page": limit, @@ -116,7 +119,9 @@ async def get_knowledge_batch(): from scribe.services.knowledge import get_knowledge_by_ids items = await get_knowledge_by_ids(uid, ids) - return jsonify({"items": items}) + # The scrolling feed hydrates its cards here, not from the list route, so the + # ownership markers have to be applied on this path too. + return jsonify({"items": await label_shared_items(uid, items)}) @knowledge_bp.route("/tags", methods=["GET"]) diff --git a/src/scribe/routes/search.py b/src/scribe/routes/search.py index 6e5254e..8d56583 100644 --- a/src/scribe/routes/search.py +++ b/src/scribe/routes/search.py @@ -3,6 +3,7 @@ import time from quart import Blueprint, jsonify, request from scribe.auth import login_required, get_current_user_id +from scribe.services.access import owner_names_for from scribe.services.embeddings import semantic_search_notes from scribe.services.retrieval_telemetry import record_retrieval @@ -36,7 +37,9 @@ async def search_route(): t0 = time.perf_counter() results = await semantic_search_notes( - uid, q, limit=limit, is_task=is_task, threshold=_REST_SEARCH_THRESHOLD + uid, q, limit=limit, is_task=is_task, threshold=_REST_SEARCH_THRESHOLD, + # The user typed this, so it reaches everything they may read. + scope="read", ) record_retrieval( user_id=uid, source="rest_search", query=q, @@ -44,6 +47,9 @@ async def search_route(): project_id=None, is_task=is_task, results=results, duration_ms=(time.perf_counter() - t0) * 1000.0, ) + owners = await owner_names_for( + {int(note.user_id) for _s, note in results if note.user_id != uid} + ) return jsonify({ "results": [ { @@ -53,6 +59,10 @@ async def search_route(): "is_task": note.is_task, "tags": note.tags or [], "similarity": score, + **( + {"shared": True, "owner": owners.get(int(note.user_id))} + if note.user_id != uid else {} + ), } for score, note in results # semantic_search_notes returns list[tuple[float, Note]] ], diff --git a/src/scribe/routes/snippets.py b/src/scribe/routes/snippets.py new file mode 100644 index 0000000..0074eb9 --- /dev/null +++ b/src/scribe/routes/snippets.py @@ -0,0 +1,238 @@ +"""REST routes for snippets — reusable functions/components recorded for recall. + +A snippet is a note with note_type='snippet' (see services/snippets.py). These +routes feed the web management UI; the MCP tools (mcp/tools/snippets.py) are the +agent-facing surface. Both go through services/snippets.py, so the +serialize/parse contract and embedding-on-create live in one place (DRY). + +ACL (rule #78): reads/writes of a single snippet resolve through the share-aware +`get_note_for_user` + `can_write_note`, and writes are performed as the OWNER so +a shared editor isn't rejected by the owner-scoped service — mirroring +routes/notes.py. The list is owner-scoped, matching the note-browse surface. +""" +import logging + +from quart import Blueprint, jsonify, request + +from scribe.auth import get_current_user_id, login_required +from scribe.routes.utils import not_found, parse_pagination +from scribe.services import dedup as dedup_svc +from scribe.services import snippets as snippets_svc +from scribe.services import systems as systems_svc +from scribe.services.access import ( + can_write_note, + describe_provenance, + label_shared_items, +) +from scribe.services.notes import get_note_for_user + +logger = logging.getLogger(__name__) + +snippets_bp = Blueprint("snippets", __name__, url_prefix="/api/snippets") + +# Fields the create/update payload may carry, mapped straight to the service. +_STR_FIELDS = ("name", "code", "language", "signature", "when_to_use", "repo", "path", "symbol") + + +async def _load_snippet(uid: int, snippet_id: int): + """Share-aware resolve of a snippet by id → (note, permission) or None if it + isn't accessible or isn't a snippet.""" + result = await get_note_for_user(uid, snippet_id) + if result is None: + return None + note, permission = result + if note.note_type != snippets_svc.SNIPPET_NOTE_TYPE: + return None + return note, permission + + +@snippets_bp.route("", methods=["GET"]) +@login_required +async def list_snippets_route(): + uid = get_current_user_id() + q = request.args.get("q") or None + tag = request.args.get("tag", "") + try: + project_id = int(request.args.get("project_id", 0) or 0) or None + except (TypeError, ValueError): + project_id = None + limit, offset = parse_pagination() + items, total = await snippets_svc.list_snippets( + uid, q=q, tag=tag, limit=limit, offset=offset, project_id=project_id, + ) + # Mark rows owned by someone else so the UI can show whose they are — an + # unmarked row in your own list reads as one you recorded and vetted. + items = await label_shared_items(uid, items) + return jsonify({"snippets": items, "total": total}) + + +@snippets_bp.route("", methods=["POST"]) +@login_required +async def create_snippet_route(): + uid = get_current_user_id() + data = await request.get_json() or {} + name = (data.get("name") or "").strip() + code = (data.get("code") or "").strip() + if not name or not code: + return jsonify({"error": "name and code are required"}), 400 + project_id = data.get("project_id") or None + + # Same near-duplicate gate the MCP create path applies: recording the same + # reusable thing twice is what merge then has to undo, so catch it here too. + # `force` is the deliberate override once the operator has seen the warning. + if not data.get("force"): + dup = await dedup_svc.find_duplicate_note( + uid, + snippets_svc.compose_title(name, data.get("when_to_use", "")), + snippets_svc.compose_body( + code=data.get("code", ""), + language=data.get("language", ""), + signature=data.get("signature", ""), + when_to_use=data.get("when_to_use", ""), + repo=data.get("repo", ""), + path=data.get("path", ""), + symbol=data.get("symbol", ""), + locations=data.get("locations"), + ), + project_id=project_id, + is_task=False, + note_type=snippets_svc.SNIPPET_NOTE_TYPE, + ) + if dup is not None: + return jsonify(dedup_svc.duplicate_response(dup, "snippet")), 409 + + note = await snippets_svc.create_snippet( + uid, + name=name, + code=data.get("code", ""), + language=data.get("language", ""), + signature=data.get("signature", ""), + when_to_use=data.get("when_to_use", ""), + repo=data.get("repo", ""), + path=data.get("path", ""), + symbol=data.get("symbol", ""), + locations=data.get("locations"), + tags=data.get("tags"), + project_id=project_id, + ) + if data.get("system_ids") is not None: + await systems_svc.set_record_systems(uid, note.id, data["system_ids"]) + out = snippets_svc.snippet_to_dict(note) + out["systems"] = [ + s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id) + ] + return jsonify(out), 201 + + +@snippets_bp.route("/", methods=["GET"]) +@login_required +async def get_snippet_route(snippet_id: int): + uid = get_current_user_id() + loaded = await _load_snippet(uid, snippet_id) + if loaded is None: + return not_found("Snippet") + note, permission = loaded + data = snippets_svc.snippet_to_dict(note) + data["permission"] = permission + # Read the association as the OWNER: a shared reader isn't scoped to the + # owner's project, so their own id would come back empty (mirrors the + # write-as-owner pattern this module already uses). + data["systems"] = [ + s.to_dict() + for s in await systems_svc.list_record_systems(note.user_id, snippet_id) + ] + data.update(await describe_provenance(uid, note)) + return jsonify(data) + + +@snippets_bp.route("/", methods=["PATCH"]) +@login_required +async def update_snippet_route(snippet_id: int): + uid = get_current_user_id() + loaded = await _load_snippet(uid, snippet_id) + if loaded is None: + return not_found("Snippet") + note, _ = loaded + if not await can_write_note(uid, snippet_id): + return jsonify({"error": "Permission denied"}), 403 + owner_uid = note.user_id + data = await request.get_json() or {} + + # Partial update: only keys present in the payload change (the service + # treats None as "leave unchanged"). Empty strings ARE applied — the form + # sends the full field set, so a cleared field is an intentional clear. + kwargs = {k: data[k] for k in _STR_FIELDS if k in data} + if "locations" in data: + kwargs["locations"] = data["locations"] + if "tags" in data: + kwargs["tags"] = data["tags"] + if "project_id" in data: + # A present-but-empty project_id is a deliberate detach, not "unchanged" + # — the service distinguishes the two via its UNSET sentinel. + kwargs["project_id"] = data["project_id"] or None + + updated = await snippets_svc.update_snippet(owner_uid, snippet_id, **kwargs) + if updated is None: + return not_found("Snippet") + if data.get("system_ids") is not None: + await systems_svc.set_record_systems(owner_uid, snippet_id, data["system_ids"]) + out = snippets_svc.snippet_to_dict(updated) + out["systems"] = [ + s.to_dict() for s in await systems_svc.list_record_systems(owner_uid, snippet_id) + ] + return jsonify(out) + + +@snippets_bp.route("//merge", methods=["POST"]) +@login_required +async def merge_snippet_route(snippet_id: int): + """Unify source snippets into this one (the canonical target). Body: + {"source_ids": [int, ...]}. Requires write on the target and every source, + and all must share the target's owner (cross-owner merge is out of scope).""" + uid = get_current_user_id() + loaded = await _load_snippet(uid, snippet_id) + if loaded is None: + return not_found("Snippet") + target, _ = loaded + if not await can_write_note(uid, snippet_id): + return jsonify({"error": "Permission denied"}), 403 + + data = await request.get_json() or {} + raw = data.get("source_ids") or [] + source_ids = [s for s in raw if isinstance(s, int) and s != snippet_id] + if not source_ids: + return jsonify({"error": "source_ids (non-empty list of other snippet ids) is required"}), 400 + + owner_uid = target.user_id + for sid in source_ids: + sloaded = await _load_snippet(uid, sid) + if sloaded is None: + return not_found(f"Snippet {sid}") + snote, _ = sloaded + if snote.user_id != owner_uid: + return jsonify({"error": "can only merge snippets with the same owner"}), 400 + if not await can_write_note(uid, sid): + return jsonify({"error": f"Permission denied for snippet {sid}"}), 403 + + result = await snippets_svc.merge_snippets(owner_uid, snippet_id, source_ids) + if result is None: + return not_found("Snippet") + note, merged_ids = result + out = snippets_svc.snippet_to_dict(note) + out["merged_ids"] = merged_ids + return jsonify(out) + + +@snippets_bp.route("/", methods=["DELETE"]) +@login_required +async def delete_snippet_route(snippet_id: int): + uid = get_current_user_id() + loaded = await _load_snippet(uid, snippet_id) + if loaded is None: + return not_found("Snippet") + note, _ = loaded + if not await can_write_note(uid, snippet_id): + return jsonify({"error": "Permission denied"}), 403 + if not await snippets_svc.delete_snippet(note.user_id, snippet_id): + return not_found("Snippet") + return "", 204 diff --git a/src/scribe/services/access.py b/src/scribe/services/access.py index 0d46ae2..bb39d61 100644 --- a/src/scribe/services/access.py +++ b/src/scribe/services/access.py @@ -9,13 +9,14 @@ Permission rank: owner > admin > editor > viewer import logging -from sqlalchemy import select +from sqlalchemy import or_, select from scribe.models import async_session from scribe.models.group import GroupMembership from scribe.models.note import Note from scribe.models.project import Project from scribe.models.share import NoteShare, ProjectShare +from scribe.models.user import User logger = logging.getLogger(__name__) @@ -161,3 +162,195 @@ async def can_read_note(user_id: int, note_id: int) -> bool: async def can_write_note(user_id: int, note_id: int) -> bool: perm = await get_note_permission(user_id, note_id) return perm in ("editor", "admin", "owner") + + +# --------------------------------------------------------------------------- +# Set-based visibility (for LIST queries) +# +# Two scopes, deliberately different (decision note 2094): +# +# readable_* — everything the ACL permits, including a record reachable ONLY +# through a direct or group note share. For EXPLICIT acts: a +# search the caller typed, or a fetch by id. +# browsable_* — the caller's own records plus anything in a project they have +# access to. For PASSIVE surfaces: browse lists, facet counts, +# the process→skill manifest. +# +# The split is a trust boundary, not an optimisation. Anything that appears +# unasked — in your own list, your own counts, or as a skill installed on your +# machine — reads as material you endorsed. A one-off record someone shared with +# you has not earned that standing, so it waits until you go looking for it. +# --------------------------------------------------------------------------- + +def notes_visibility_clause(user_id: int, scope: str = "own"): + """The one place a retrieval declares how far it may see. + + Every path that returns notes picks a scope by what KIND of act it is: + + "own" — the caller's records only. For machinery whose answer must not + depend on other people: the near-duplicate gate can't block a + create because a stranger wrote something similar, and can't + point at a record the caller cannot edit. + "browse" — own + project-reachable. For passive surfaces, where an + unrequested record would read as endorsed. + "read" — everything the ACL permits. For explicit acts: a typed search, + a fetch by id. + + Defaults to the narrowest, so a new caller that forgets to choose is wrong in + the safe direction. + """ + if scope == "own": + return Note.user_id == user_id + if scope == "browse": + return browsable_notes_clause(user_id) + if scope == "read": + return readable_notes_clause(user_id) + raise ValueError(f"unknown note scope {scope!r} (own | browse | read)") + + +async def owner_names_for(user_ids: set[int]) -> dict[int, str]: + """{user_id: username} in one query. Empty input costs nothing. + + Fails soft: on a lookup error the caller gets no names and renders "another + user" instead. Losing an attribution is a cosmetic downgrade, and the part + that matters — that the record ISN'T the caller's — comes from comparing + owner ids, not from this. Failing the whole search over a username would be + the worse trade. + """ + if not user_ids: + return {} + try: + async with async_session() as session: + rows = ( + await session.execute( + select(User.id, User.username).where(User.id.in_(user_ids)) + ) + ).all() + return {uid: name for uid, name in rows} + except Exception: + logger.warning("Owner-name lookup failed; falling back to unnamed " + "attribution", exc_info=True) + return {} + + +def _my_group_ids(user_id: int): + """The caller's group ids as a SUBQUERY, not a fetched list. + + Keeping it in SQL is what makes the clause builders below pure functions — + no session, no await, nothing for a caller's unit test to mock — and it folds + the membership lookup into the one statement the caller was already running. + """ + return select(GroupMembership.group_id).where(GroupMembership.user_id == user_id) + + +def readable_notes_clause(user_id: int): + """A SQLAlchemy WHERE predicate matching every note this user may READ. + + `get_note_permission` answers "may I read THIS note?" one row at a time, + which a list query can't use — checking per row is O(n) round-trips. This is + the same resolution expressed as set membership so it can go straight into a + `select(Note).where(...)`: + + 1. ownership + 2. a direct note share + 3. a group note share + 4. inheritance from a shared project + + Keep the two in step: a rule added here belongs in `get_note_permission` + too, or a note becomes findable but not openable (or the reverse — which is + the bug this was written to fix). + """ + groups = _my_group_ids(user_id) + + shared_note_ids = select(NoteShare.note_id).where( + or_( + NoteShare.shared_with_user_id == user_id, + NoteShare.shared_with_group_id.in_(groups), + ) + ) + shared_project_ids = select(ProjectShare.project_id).where( + or_( + ProjectShare.shared_with_user_id == user_id, + ProjectShare.shared_with_group_id.in_(groups), + ) + ) + + return or_( + Note.user_id == user_id, + Note.id.in_(shared_note_ids), + Note.project_id.in_(shared_project_ids), + ) + + +async def describe_provenance(user_id: int, note) -> dict: + """Provenance for a record being handed to a caller: `{}` when it's theirs, + otherwise `{"shared": True, "owner": , "permission": }`. + + Anything reaching an agent or a UI from another person has to say so. A + shared record is one person's suggestion, not a standard the caller adopted, + and without a marker the two are indistinguishable — the reader would assume + their own past self wrote it and treat it as settled practice. + """ + if note is None or note.user_id == user_id: + return {} + async with async_session() as session: + owner = await session.get(User, note.user_id) + return { + "shared": True, + "owner": owner.username if owner else None, + "permission": await get_note_permission(user_id, note.id), + } + + +async def label_shared_items(user_id: int, items: list[dict]) -> list[dict]: + """Mark the entries in a list payload that belong to someone else. + + Each foreign item gains `shared: True` and `owner: `; the caller's + own items are left untouched so an all-mine list stays noise-free. Usernames + are resolved in one query per distinct owner, not one per row. + + Lists are where provenance matters most: an unmarked row in your own list + reads as something you recorded and vetted. + """ + foreign = { + it["user_id"] for it in items + if it.get("user_id") is not None and it["user_id"] != user_id + } + if not foreign: + return items + names = await owner_names_for(foreign) + for it in items: + owner_id = it.get("user_id") + if owner_id is not None and owner_id != user_id: + it["shared"] = True + it["owner"] = names.get(owner_id) + return items + + +def browsable_notes_clause(user_id: int): + """A WHERE predicate for PASSIVE surfaces: the caller's own notes, plus + notes in a project they can reach (owned or shared). + + Deliberately narrower than `readable_notes_clause` — it omits records the + caller can read *only* via a direct or group note share. Those stay + search-only, so a one-off someone handed them never drifts into a browse + list, a facet count, or the skill manifest as though it were their own. + + Project membership is the seam because it's a standing, mutual context: if + you're on the project, its content is your working material. A note with no + project that you don't own therefore never matches — `NULL IN (...)` is not + true — which is exactly the intent. + """ + shared_project_ids = select(ProjectShare.project_id).where( + or_( + ProjectShare.shared_with_user_id == user_id, + ProjectShare.shared_with_group_id.in_(_my_group_ids(user_id)), + ) + ) + owned_project_ids = select(Project.id).where(Project.user_id == user_id) + + return or_( + Note.user_id == user_id, + Note.project_id.in_(shared_project_ids), + Note.project_id.in_(owned_project_ids), + ) diff --git a/src/scribe/services/dedup.py b/src/scribe/services/dedup.py index 980efed..252f42e 100644 --- a/src/scribe/services/dedup.py +++ b/src/scribe/services/dedup.py @@ -124,6 +124,11 @@ async def find_duplicate_note( user_id, query, project_id=project_id, is_task=is_task, orphan_only=(project_id is None), limit=3, threshold=_SEMANTIC_THRESHOLD, + # Owner-only, deliberately: this gate BLOCKS a create and tells the + # caller to update the match instead. Matching someone else's record + # would refuse their write and point them at something they may not + # be able to edit. + scope="own", ) for score, note in hits: # semantic_search_notes doesn't filter note_type — enforce it here so diff --git a/src/scribe/services/embeddings.py b/src/scribe/services/embeddings.py index b893345..b88e2a4 100644 --- a/src/scribe/services/embeddings.py +++ b/src/scribe/services/embeddings.py @@ -19,6 +19,7 @@ from sqlalchemy import delete, select from scribe.models import async_session from scribe.models.embedding import NoteEmbedding from scribe.models.note import Note +from scribe.services.access import notes_visibility_clause logger = logging.getLogger(__name__) @@ -114,12 +115,21 @@ async def semantic_search_notes( project_id: int | None = None, is_task: bool | None = None, orphan_only: bool = False, + scope: str = "own", ) -> list[tuple[float, Note]]: """Return up to *limit* (score, note) pairs most relevant to *query*. Scores are cosine similarities in [-1, 1]; only notes at or above *threshold* are returned, sorted highest-first. + `scope` ("own" | "browse" | "read", see access.notes_visibility_clause) + decides how far this may see. It exists because this one function serves + three different kinds of act: an explicit search, which should reach + everything the caller may read; passive auto-injection, which must not pull + an unrequested record into their context; and the near-duplicate gate, whose + verdict must not depend on other people's notes at all. Defaults to "own" so + a caller that forgets is wrong in the safe direction. + Ranking and the top-k cut happen in Postgres via pgvector's cosine-distance operator (`<=>`, exposed as ``Vector.cosine_distance``) backed by the HNSW index from migration 0067 — so this is an indexed ``ORDER BY ... LIMIT k`` @@ -145,11 +155,17 @@ async def semantic_search_notes( try: async with async_session() as session: + # Scope on Note, not NoteEmbedding.user_id: the embedding row belongs + # to the note's owner, so filtering it would pin every scope to "own" + # and leave shared records unreachable by meaning. stmt = ( select(Note, distance.label("distance")) .select_from(NoteEmbedding) .join(Note, NoteEmbedding.note_id == Note.id) - .where(NoteEmbedding.user_id == user_id, Note.deleted_at.is_(None)) + .where( + notes_visibility_clause(user_id, scope), + Note.deleted_at.is_(None), + ) ) if orphan_only: stmt = stmt.where(Note.project_id.is_(None)) diff --git a/src/scribe/services/knowledge.py b/src/scribe/services/knowledge.py index 3775855..71d987f 100644 --- a/src/scribe/services/knowledge.py +++ b/src/scribe/services/knowledge.py @@ -1,10 +1,27 @@ -"""Knowledge service — unified query across notes, tasks, plans, and processes.""" +"""Knowledge service — unified query across notes, tasks, plans, and processes. + +ACL (rules #47/#78, decision note 2094): these queries were owner-only until +2026-07-25, which meant a record shared with you could be opened by id but never +*found*. They now honour shares — at two different widths: + + - **searching** (a `q` the caller typed) uses the full read scope, so a record + shared directly with you is findable when you go looking for it. BOTH halves + of the hybrid search — keyword and semantic — see equally, or a record would + be findable by wording and invisible by meaning; + - **browsing** (no `q`) and the facet counts beside it use the narrower browse + scope: your own records plus anything in a project you can reach. + +The asymmetry is the point. A record that appears unasked reads as one you +endorsed, so a one-off direct share has to be searched for rather than arriving +in your ambient lists. +""" import logging from sqlalchemy import func, select from scribe.models import async_session from scribe.models.note import Note +from scribe.services.access import browsable_notes_clause, readable_notes_clause logger = logging.getLogger(__name__) @@ -19,6 +36,9 @@ def _note_to_item(note: Note) -> dict: "snippet": (note.body or "")[:_SNIPPET_LEN], "tags": note.tags or [], "project_id": note.project_id, + # These lists now include records shared with the caller, so the client + # needs the owner to tell "mine" from "someone else's" in a mixed list. + "user_id": note.user_id, "created_at": note.created_at.isoformat(), "updated_at": note.updated_at.isoformat(), } @@ -59,21 +79,30 @@ async def query_knowledge( q: str | None, limit: int, offset: int, + project_id: int | None = None, ) -> tuple[list[dict], int]: """Query knowledge objects (non-task notes) with filters. + `project_id` narrows to one project (None = every project). + Returns (items, total_count). """ # Semantic search path — scores take priority over sort if q: return await _semantic_knowledge_search( - user_id, q, note_type=note_type, tags=tags, limit=limit, offset=offset + user_id, q, note_type=note_type, tags=tags, limit=limit, + offset=offset, project_id=project_id, ) + # No query = browsing. Narrower scope: a record shared directly with the + # caller is search-only and must not appear in an ambient list. + visible = browsable_notes_clause(user_id) async with async_session() as session: - base = select(Note).where(Note.user_id == user_id) + base = select(Note).where(visible) base = _apply_type_filter(base, note_type) + if project_id is not None: + base = base.where(Note.project_id == project_id) for tag in tags: base = base.where(Note.tags.contains([tag])) @@ -104,6 +133,7 @@ async def _semantic_knowledge_search( tags: list[str], limit: int, offset: int, + project_id: int | None = None, ) -> tuple[list[dict], int]: """Hybrid search: keyword matches first (title/body ILIKE), then semantic results. @@ -122,14 +152,19 @@ async def _semantic_knowledge_search( # 1. Keyword search — title and body ILIKE keyword_notes: list[Note] = [] try: + # A typed query is an explicit act, so it reaches the caller's full read + # scope — including records shared directly with them. + visible = readable_notes_clause(user_id) async with async_session() as session: pattern = f"%{q}%" base = ( select(Note) - .where(Note.user_id == user_id) + .where(visible) .where(Note.title.ilike(pattern) | Note.body.ilike(pattern)) ) base = _apply_type_filter(base, note_type) + if project_id is not None: + base = base.where(Note.project_id == project_id) for tag in tags: base = base.where(Note.tags.contains([tag])) # Title matches first, then body-only matches, newest first within each @@ -141,17 +176,22 @@ async def _semantic_knowledge_search( except Exception: logger.warning("Keyword search failed", exc_info=True) - # 2. Semantic search — conceptual similarity + # 2. Semantic search — conceptual similarity, at the SAME scope as the + # keyword half above. Both halves of one search must see equally, or a shared + # record would be findable by wording and invisible by meaning — which is the + # case a semantic search exists to serve. semantic_notes: list[Note] = [] try: from scribe.services.embeddings import semantic_search_notes is_task_filter = True if note_type in ("task", "plan") else (False if note_type else None) candidates = await semantic_search_notes( user_id=user_id, + scope="read", query=q, limit=min(200, limit * 4), threshold=0.3, is_task=is_task_filter, + project_id=project_id, ) for _score, note in candidates: if note.deleted_at is not None: @@ -186,11 +226,16 @@ async def _semantic_knowledge_search( async def get_knowledge_tags(user_id: int, note_type: str | None = None) -> list[str]: - """Return all distinct tags used across knowledge objects for this user.""" + """Distinct tags across what this user can BROWSE. + + Follows the browse list rather than the read scope: a facet is itself a + passive surface, and offering a tag that only a search-only record carries + would filter the visible list down to nothing.""" + visible = browsable_notes_clause(user_id) async with async_session() as session: base = ( select(func.unnest(Note.tags).label("tag")) - .where(Note.user_id == user_id) + .where(visible) ) base = _apply_type_filter(base, note_type) stmt = base.distinct().order_by("tag") @@ -199,12 +244,15 @@ async def get_knowledge_tags(user_id: int, note_type: str | None = None) -> list async def get_knowledge_counts(user_id: int, tags: list[str] | None = None) -> dict[str, int]: - """Return per-type count of knowledge objects for the sidebar display.""" + """Per-type counts for the sidebar, over what this user can BROWSE — so the + numbers match the list they sit beside rather than promising rows that only a + search would surface.""" + visible = browsable_notes_clause(user_id) async with async_session() as session: # Count non-task types stmt = ( select(Note.note_type, func.count(Note.id)) - .where(Note.user_id == user_id) + .where(visible) .where(Note.status.is_(None)) .where(Note.deleted_at.is_(None)) .where(Note.note_type.in_(["note", "process"])) @@ -219,7 +267,7 @@ async def get_knowledge_counts(user_id: int, tags: list[str] | None = None) -> d # Count tasks separately (is_task = status IS NOT NULL) task_stmt = ( select(func.count(Note.id)) - .where(Note.user_id == user_id) + .where(visible) .where(Note.status.isnot(None)) .where(Note.deleted_at.is_(None)) ) @@ -233,7 +281,7 @@ async def get_knowledge_counts(user_id: int, tags: list[str] | None = None) -> d # but NOT added to total to avoid double-counting against "task". plan_stmt = ( select(func.count(Note.id)) - .where(Note.user_id == user_id) + .where(visible) .where(Note.status.isnot(None)) .where(Note.task_kind == "plan") .where(Note.deleted_at.is_(None)) @@ -267,8 +315,10 @@ async def query_knowledge_ids( ) return [item["id"] for item in items], total + # Browsing (see query_knowledge) — narrower scope. + visible = browsable_notes_clause(user_id) async with async_session() as session: - base = select(Note.id).where(Note.user_id == user_id) + base = select(Note.id).where(visible) base = _apply_type_filter(base, note_type) for tag in tags: @@ -295,10 +345,13 @@ async def get_knowledge_by_ids(user_id: int, ids: list[int]) -> list[dict]: """Fetch full items for the given IDs, preserving the requested order.""" if not ids: return [] + # Fetching specific ids is explicit, so this takes the full read scope — the + # ids came from either a browse or a search, and both must resolve. + visible = readable_notes_clause(user_id) async with async_session() as session: stmt = ( select(Note) - .where(Note.user_id == user_id) + .where(visible) .where(Note.id.in_(ids)) .where(Note.deleted_at.is_(None)) ) diff --git a/src/scribe/services/notes.py b/src/scribe/services/notes.py index be91c90..3df0486 100644 --- a/src/scribe/services/notes.py +++ b/src/scribe/services/notes.py @@ -412,15 +412,23 @@ async def convert_task_to_note(user_id: int, note_id: int) -> Note: async def resolve_process(user_id: int, name_or_id) -> tuple[Note | None, list[dict]]: """Resolve a stored process by id or name. - Owner-scoped, note_type='process', non-trashed. Precedence: numeric id → - exact case-insensitive title → substring. Returns (note, other_candidates); - on a substring tie with no exact hit, `note` is the most-recently-updated - match and `other_candidates` lists the rest as [{id, title}] so the caller - can disambiguate. Returns (None, []) when nothing matches. + note_type='process', non-trashed, scoped to what this user may READ — owned + plus shared (rule #78). Naming one is an explicit act, so a Process shared + with the caller resolves here even though it is deliberately absent from the + passive process list and the skill manifest (decision note 2094); without + this, a Process a search surfaced could not then be run — see #2093. + + Precedence: numeric id → exact case-insensitive title → substring. Returns + (note, other_candidates); on a substring tie with no exact hit, `note` is the + most-recently-updated match and `other_candidates` lists the rest as + [{id, title}] so the caller can disambiguate. Returns (None, []) when nothing + matches. """ + from scribe.services.access import readable_notes_clause + visible = readable_notes_clause(user_id) async with async_session() as session: base = select(Note).where( - Note.user_id == user_id, + visible, Note.note_type == "process", Note.deleted_at.is_(None), ) diff --git a/src/scribe/services/plugin_context.py b/src/scribe/services/plugin_context.py index bf740c7..7241838 100644 --- a/src/scribe/services/plugin_context.py +++ b/src/scribe/services/plugin_context.py @@ -25,6 +25,7 @@ from scribe.services import knowledge as knowledge_svc from scribe.services import notes as notes_svc from scribe.services import projects as projects_svc from scribe.services import rulebooks as rulebooks_svc +from scribe.services.access import label_shared_items, owner_names_for from scribe.services.embeddings import semantic_search_notes from scribe.services.retrieval_telemetry import record_retrieval from scribe.services.settings import get_setting @@ -74,13 +75,22 @@ async def build_process_manifest(user_id: int) -> dict: (note_type='process'). Instance-agnostic: derived from whatever Processes the calling install owns, no operator-specific coupling. - Returns {"processes": [{id, name, slug, description}], "total": int}. - Slugs are unique within the result (a collision gets an - suffix). + SCOPE: this is the most consequential passive surface Scribe has — every + entry becomes a skill file on the operator's machine that auto-surfaces and + is followed as written. It therefore uses the BROWSE scope (via the + no-query knowledge list): a Process shared directly with the operator is + never installed here, only one they own or reach through a shared project + (decision note 2094). Project-shared entries are labelled with their owner so + the stub can't pass off someone else's procedure as the operator's own. + + Returns {"processes": [{id, name, slug, description, shared?, owner?}], + "total": int}. Slugs are unique within the result (collision gets -). """ items, _ = await knowledge_svc.query_knowledge( user_id=user_id, note_type="process", tags=[], sort="modified", q=None, limit=100, offset=0, ) + items = await label_shared_items(user_id, items) procs: list[dict] = [] seen: set[str] = set() for it in items: @@ -95,16 +105,32 @@ async def build_process_manifest(user_id: int) -> dict: preview = " ".join((it.get("snippet") or "").split()) if len(preview) > _PROC_PREVIEW_CHARS: preview = preview[:_PROC_PREVIEW_CHARS].rstrip() + "…" - description = ( - f'Run the operator\'s saved Scribe process "{title}".' - + (f" {preview}" if preview else "") - + f' Use when {title}-type work is requested, or when asked to run' - f' the "{title}" process.' - ) - procs.append({ + if it.get("shared"): + owner = it.get("owner") or "another user" + description = ( + f'A shared Scribe process "{title}", authored by {owner} — NOT the' + f" operator's own." + + (f" {preview}" if preview else "") + + f' Use when {title}-type work is requested, or when asked to run' + f' the "{title}" process — but summarise it and get the operator\'s' + f" go-ahead before following it, since it reflects {owner}'s" + f" judgement rather than theirs." + ) + else: + description = ( + f'Run the operator\'s saved Scribe process "{title}".' + + (f" {preview}" if preview else "") + + f' Use when {title}-type work is requested, or when asked to run' + f' the "{title}" process.' + ) + entry = { "id": it["id"], "name": title, "slug": slug, "description": description, - }) + } + if it.get("shared"): + entry["shared"] = True + entry["owner"] = it.get("owner") + procs.append(entry) return {"processes": procs, "total": len(procs)} @@ -170,6 +196,11 @@ async def build_autoinject_hint( threshold=cfg["threshold"], project_id=(project_id or None), exclude_ids=set(exclude_ids or []), + # Injection is the one retrieval nobody asked for, so it takes the BROWSE + # scope: never a record shared one-to-one with the operator. What can + # still appear is a collaborator's note inside a shared project — legible + # only because the line below names its owner. + scope="browse", ) record_retrieval( user_id=user_id, source="auto_inject", query=q, @@ -184,6 +215,13 @@ async def build_autoinject_hint( top_score = hits[0][0] kept = [(s, n) for s, n in hits if s >= top_score - _AUTOINJECT_BAND] + # A collaborator's note can reach this menu via a shared project, and the + # operator never asked for it — so say whose it is. Unattributed, it reads as + # something they wrote and settled. + owners = await owner_names_for({ + int(n.user_id) for _s, n in kept if n.user_id != user_id + }) + lines = [ "> Possibly relevant from your Scribe notes — call `get_note(id)` to " "open any in full (titles only; injected once per session):", @@ -192,7 +230,11 @@ async def build_autoinject_hint( for score, note in kept: note_ids.append(int(note.id)) title = (note.title or "(untitled)").replace("\n", " ").strip() - lines.append(f"> - #{note.id} \"{title}\" ({score:.2f})") + line = f"> - #{note.id} \"{title}\" ({score:.2f})" + if note.user_id != user_id: + who = owners.get(int(note.user_id)) or "another user" + line += f" — shared by {who}, treat as a suggestion" + lines.append(line) return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg} diff --git a/src/scribe/services/snippets.py b/src/scribe/services/snippets.py new file mode 100644 index 0000000..0c4636e --- /dev/null +++ b/src/scribe/services/snippets.py @@ -0,0 +1,542 @@ +"""Snippet service — reusable functions/components recorded for recall. + +A *snippet* is a Note with ``note_type='snippet'``: a named, reusable function or +component recorded once so a later session can recall it before writing a +one-off. It carries a name, language, signature, canonical location +(repo · path · symbol), a one-line "when to reach for it", and the code itself. + +Structured fields are stored as a **body-convention** — no dedicated column, so +a snippet inherits everything a note has (embeddings, ACL, project/System +association, dedup) and, crucially, becomes eligible for semantic recall the +moment it's embedded: + + - ``title`` = ``"{name} — {when_to_use}"``. The title is exactly what the + title-first auto-inject surfaces, so this one line self-describes the snippet + in a recall menu. + - ``tags`` = ``[language, "snippet", *caller_tags]``. + - ``body`` = templated markdown (When to use / Signature / Location, then a + fenced code block). + +The public field API (name/language/signature/location/when_to_use/code) lives +here, so the underlying storage could later move to a structured column without +changing any caller (MCP tool, REST route, UI). +""" +from __future__ import annotations + +import asyncio +import re + +from scribe.services import knowledge as knowledge_svc +from scribe.services import notes as notes_svc + +SNIPPET_NOTE_TYPE = "snippet" +SNIPPET_TAG = "snippet" + +# Sentinel for "argument not supplied" on update, so None stays available as a +# real value meaning "clear this". Needed for project_id, where 0 is not a valid +# id and None is the clear — the two can't share one default. +UNSET: object = object() + + +def _embed_snippet(note) -> None: + """Fire-and-forget embedding refresh for a snippet. + + A snippet's whole value is *immediate* recall — it must join the semantic / + auto-inject pool the moment it's recorded, not wait for the startup backfill. + Unlike a plain note (embedded at the REST-route boundary only, so its MCP + create path defers to restart-backfill), a snippet is recorded primarily via + MCP, so we embed here in the service — covering BOTH the MCP tool and the + REST route by construction. Mirrors the route pattern: fire-and-forget, + text = title + body. Import lazily so the pure serialize/parse helpers can be + imported without pulling in the embedding model. + """ + text = f"{note.title}\n{note.body}".strip() if note.body else (note.title or "") + if not text: + return + from scribe.services.embeddings import upsert_note_embedding + asyncio.create_task(upsert_note_embedding(note.id, note.user_id, text)) + + +# --- serialize: structured fields -> note (title/body/tags) ------------------ + +def compose_title(name: str, when_to_use: str = "") -> str: + """`name — when to use` (or just `name` when no usage note is given).""" + name = (name or "").strip() + when = (when_to_use or "").strip() + return f"{name} — {when}" if when else name + + +def compose_tags(language: str = "", tags: list[str] | None = None) -> list[str]: + """Language (lowercased) first, then the `snippet` marker, then caller tags — + de-duplicated, order preserved.""" + out: list[str] = [] + lang = (language or "").strip().lower() + if lang: + out.append(lang) + out.append(SNIPPET_TAG) + for t in tags or []: + t = (t or "").strip() + if t and t not in out: + out.append(t) + return out + + +def _normalize_locations(locations: list[dict] | None) -> list[dict]: + """Clean a list of {repo,path,symbol} locations: strip fields, drop wholly + empty entries, de-duplicate identical ones (order preserved). A merged + snippet carries several locations (one per call site); a fresh one carries + at most one.""" + out: list[dict] = [] + seen: set[tuple[str, str, str]] = set() + for loc in locations or []: + repo = (loc.get("repo") or "").strip() + path = (loc.get("path") or "").strip() + symbol = (loc.get("symbol") or "").strip() + if not (repo or path or symbol): + continue + key = (repo, path, symbol) + if key in seen: + continue + seen.add(key) + out.append({"repo": repo, "path": path, "symbol": symbol}) + return out + + +def _location_str(loc: dict) -> str: + """`repo` · `path` · `symbol` — only the non-empty parts.""" + parts = [(loc.get(k) or "").strip() for k in ("repo", "path", "symbol")] + return " · ".join(f"`{p}`" for p in parts if p) + + +def _render_location_block(locations: list[dict]) -> str | None: + """One `**Location:**` line for a single location; a `**Locations:**` bullet + list for several. None when there are none.""" + locs = [loc for loc in locations if _location_str(loc)] + if not locs: + return None + if len(locs) == 1: + return f"**Location:** {_location_str(locs[0])}" + lines = "\n".join(f"- {_location_str(loc)}" for loc in locs) + return f"**Locations:**\n{lines}" + + +def compose_body( + *, + code: str, + language: str = "", + signature: str = "", + when_to_use: str = "", + repo: str = "", + path: str = "", + symbol: str = "", + locations: list[dict] | None = None, +) -> str: + """Render structured fields into the snippet body markdown. Empty fields are + omitted so the body stays clean. + + Locations: pass ``locations`` (a list of {repo,path,symbol}) for the general + multi-location case; the single ``repo``/``path``/``symbol`` params remain as + a back-compat shorthand for one location and are used only when ``locations`` + is not given. + """ + if locations is None: + locations = [{"repo": repo, "path": path, "symbol": symbol}] + locs = _normalize_locations(locations) + + header: list[str] = [] + if (when_to_use or "").strip(): + header.append(f"**When to use:** {when_to_use.strip()}") + if (signature or "").strip(): + header.append(f"**Signature:** `{signature.strip()}`") + loc_block = _render_location_block(locs) + if loc_block: + header.append(loc_block) + fence_lang = (language or "").strip().lower() + code_block = f"```{fence_lang}\n{(code or '').rstrip()}\n```" + if header: + return "\n\n".join(header) + "\n\n" + code_block + "\n" + return code_block + "\n" + + +# --- parse: note -> structured fields (best-effort, never raises) ------------ + +_WHEN_RE = re.compile(r"^\*\*When to use:\*\*\s*(.+?)\s*$", re.MULTILINE) +_SIG_RE = re.compile(r"^\*\*Signature:\*\*\s*(.+?)\s*$", re.MULTILINE) +_LOC_RE = re.compile(r"^\*\*Location:\*\*\s*(.+?)\s*$", re.MULTILINE) +_LOCS_RE = re.compile( + r"^\*\*Locations:\*\*[ \t]*\n((?:[ \t]*-[ \t]*.+\n?)+)", re.MULTILINE +) +_CODE_RE = re.compile(r"```([\w+.#-]*)\n(.*?)\n```", re.DOTALL) + + +def _parse_location_str(s: str) -> dict | None: + """Parse a `repo` · `path` · `symbol` fragment back into a location dict, or + None if it's empty.""" + raw = [p.strip().strip("`").strip() for p in (s or "").split("·")] + repo = raw[0] if len(raw) > 0 else "" + path = raw[1] if len(raw) > 1 else "" + symbol = raw[2] if len(raw) > 2 else "" + if not (repo or path or symbol): + return None + return {"repo": repo, "path": path, "symbol": symbol} + + +def parse_snippet_fields( + title: str, body: str, tags: list[str] | None = None +) -> dict: + """Recover structured fields from a snippet note. Tolerant by design: a field + that isn't present comes back empty and this never raises, so a hand-edited + body can't break the edit form. + + ``locations`` is a list of {repo,path,symbol}; ``repo``/``path``/``symbol`` + mirror the FIRST location for back-compat with the single-location callers.""" + title = title or "" + body = body or "" + name, _, when_from_title = title.partition(" — ") + fields = { + "name": name.strip(), + "when_to_use": when_from_title.strip(), + "signature": "", + "language": "", + "repo": "", + "path": "", + "symbol": "", + "locations": [], + "code": "", + } + + m = _WHEN_RE.search(body) + if m: + fields["when_to_use"] = m.group(1).strip() + m = _SIG_RE.search(body) + if m: + fields["signature"] = m.group(1).strip().strip("`").strip() + + # Locations: prefer the multi-location `**Locations:**` bullet list, else the + # legacy single `**Location:**` line. + locations: list[dict] = [] + m = _LOCS_RE.search(body) + if m: + for line in m.group(1).splitlines(): + line = line.strip() + if line.startswith("-"): + loc = _parse_location_str(line[1:]) + if loc: + locations.append(loc) + else: + m = _LOC_RE.search(body) + if m: + loc = _parse_location_str(m.group(1)) + if loc: + locations.append(loc) + fields["locations"] = locations + if locations: + fields["repo"] = locations[0]["repo"] + fields["path"] = locations[0]["path"] + fields["symbol"] = locations[0]["symbol"] + + m = _CODE_RE.search(body) + if m: + fields["language"] = m.group(1).strip() + fields["code"] = m.group(2) + + # Language fallback for a body whose code fence lost its language. Only the + # FIRST tag can be trusted: compose_tags emits [language, "snippet", *caller], + # so a leading tag that isn't the marker is the language — while a leading + # marker means no language was recorded. Scanning for "first tag that isn't + # the marker" instead would promote a caller's plain tag to the language. + if not fields["language"] and tags and tags[0] != SNIPPET_TAG: + fields["language"] = tags[0] + return fields + + +def snippet_to_dict(note) -> dict: + """Note serialization plus a parsed ``snippet`` sub-object of structured + fields, so callers get both the raw record and the typed view.""" + data = note.to_dict() + data["snippet"] = parse_snippet_fields(note.title, note.body, note.tags) + return data + + +# --- service wrappers over notes_svc ---------------------------------------- + +async def create_snippet( + user_id: int, + *, + name: str, + code: str, + language: str = "", + signature: str = "", + when_to_use: str = "", + repo: str = "", + path: str = "", + symbol: str = "", + locations: list[dict] | None = None, + tags: list[str] | None = None, + project_id: int | None = None, +): + """Create a snippet note (embedded on create for immediate recall). Returns + the created Note. Pass ``locations`` for the multi-location case; the single + ``repo``/``path``/``symbol`` are the one-location shorthand.""" + note = await notes_svc.create_note( + user_id, + title=compose_title(name, when_to_use), + body=compose_body( + code=code, language=language, signature=signature, + when_to_use=when_to_use, repo=repo, path=path, symbol=symbol, + locations=locations, + ), + note_type=SNIPPET_NOTE_TYPE, + tags=compose_tags(language, tags), + project_id=project_id, + ) + _embed_snippet(note) + return note + + +async def get_snippet(user_id: int, snippet_id: int): + """Fetch a snippet by id, or None if it doesn't exist / isn't a snippet / + isn't readable by this user. + + Share-aware (rule #78): a fetch by id is an explicit act, so it resolves the + caller's full read scope rather than ownership alone. Without this, a snippet + that a search legitimately surfaced could not then be opened — see #2093.""" + result = await notes_svc.get_note_for_user(user_id, snippet_id) + if result is None: + return None + note, _permission = result + if note.note_type != SNIPPET_NOTE_TYPE or note.deleted_at is not None: + return None + return note + + +async def list_snippets( + user_id: int, + *, + q: str | None = None, + tag: str = "", + limit: int = 50, + offset: int = 0, + project_id: int | None = None, +) -> tuple[list[dict], int]: + """List snippets (id/title/tags/preview dicts), most-recently-updated first. + + ``project_id`` narrows to one project; omit it to reach across every project + — which is the point when the thing you're about to write was already solved + somewhere else.""" + return await knowledge_svc.query_knowledge( + user_id=user_id, + note_type=SNIPPET_NOTE_TYPE, + tags=[tag] if tag else [], + sort="modified", + q=q, + limit=max(1, min(limit, 100)), + offset=max(0, offset), + project_id=project_id, + ) + + +async def update_snippet( + user_id: int, + snippet_id: int, + *, + name: str | None = None, + code: str | None = None, + language: str | None = None, + signature: str | None = None, + when_to_use: str | None = None, + repo: str | None = None, + path: str | None = None, + symbol: str | None = None, + locations: list[dict] | None = None, + tags: list[str] | None = None, + project_id: int | None | object = UNSET, +): + """Partial update: only fields passed (not None) change. Re-serializes the + merged field set back into title/body/tags. Returns the Note, or None if the + id isn't a snippet the caller can see. + + Share-aware (rule #47/#78): resolves the read scope, then requires WRITE — + so an editor/admin grant lets the holder edit, and a viewer grant does not. + Raises PermissionError when the caller can read but not write, because "not + found" would be a lie about a record they can plainly open. The write itself + is performed as the OWNER, mirroring routes/snippets.py, since the underlying + note update is owner-scoped. + + ``project_id``: omit to leave unchanged, pass None to detach from its + project, pass an id to move it. + + Locations: ``locations`` replaces the whole set; else a legacy single + ``repo``/``path``/``symbol`` overlays onto the first existing location; else + the existing locations are kept.""" + note = await get_snippet(user_id, snippet_id) + if note is None: + return None + from scribe.services.access import can_write_note + if not await can_write_note(user_id, snippet_id): + raise PermissionError( + f"snippet {snippet_id} is shared with you read-only — ask its owner " + f"for edit access, or record your own version" + ) + + cur = parse_snippet_fields(note.title, note.body, note.tags) + overlay = { + "name": name, "code": code, "language": language, + "signature": signature, "when_to_use": when_to_use, + } + merged = {**cur, **{k: v for k, v in overlay.items() if v is not None}} + + if locations is not None: + merged_locations = _normalize_locations(locations) + elif repo is not None or path is not None or symbol is not None: + base = cur["locations"][0] if cur["locations"] else {"repo": "", "path": "", "symbol": ""} + merged_locations = _normalize_locations([{ + "repo": repo if repo is not None else base["repo"], + "path": path if path is not None else base["path"], + "symbol": symbol if symbol is not None else base["symbol"], + }]) + else: + merged_locations = cur["locations"] + + fields: dict = { + "title": compose_title(merged["name"], merged["when_to_use"]), + "body": compose_body( + code=merged["code"], language=merged["language"], + signature=merged["signature"], when_to_use=merged["when_to_use"], + locations=merged_locations, + ), + } + # Recompute tags: keep any non-language, non-marker tags the note already had + # (or the caller's replacement set), then re-derive language + marker. + existing_extra = [ + t for t in (note.tags or []) if t not in (SNIPPET_TAG, cur.get("language", "")) + ] + fields["tags"] = compose_tags( + merged["language"], tags if tags is not None else existing_extra + ) + if project_id is not UNSET: + fields["project_id"] = project_id + + # As the OWNER: update_note is owner-scoped, so a shared editor's own id + # would find nothing. The write was authorised by can_write_note above. + updated = await notes_svc.update_note(note.user_id, snippet_id, **fields) + if updated is not None: + # Title/body changed → refresh the embedding so recall reflects the edit. + _embed_snippet(updated) + return updated + + +async def delete_snippet(user_id: int, snippet_id: int) -> bool: + """Retire a snippet to the trash (recoverable). Returns False if the id isn't + a snippet this user may WRITE. + + Recall makes this corrective, not merely tidy: a wrong or obsolete snippet + doesn't sit quietly — it keeps being offered as prior art. Removing it has to + be reachable from wherever it was recorded. + + Note the explicit write check: `get_snippet` resolves the READ scope, which + now includes snippets merely shared with this user — being able to see one + must not imply being able to bin it. + """ + note = await get_snippet(user_id, snippet_id) + if note is None: + return False + from scribe.services.access import can_write_note + if not await can_write_note(user_id, snippet_id): + return False + from scribe.services.trash import delete as trash_delete + return await trash_delete(note.user_id, "note", snippet_id) is not None + + +# --- merge: unify found one-offs into one canonical snippet ------------------ + +def _extra_tags(tags: list[str] | None, language: str = "") -> list[str]: + """A snippet's caller tags — everything except the language + `snippet` + markers that compose_tags re-derives.""" + lang = (language or "").strip().lower() + return [t for t in (tags or []) if t and t != SNIPPET_TAG and t != lang] + + +def merge_snippet_fields( + target_fields: dict, target_tags: list[str] | None, sources: list[tuple[dict, list]] +) -> tuple[list[dict], list[str]]: + """Pure merge: union locations (target's first, then each source in order) + and union extra tags. The target's scalar fields (name/when_to_use/signature/ + language/code) win — only locations and tags accumulate. ``sources`` is a + list of (parsed_fields, tags). Returns (locations, extra_tags).""" + locations = list(target_fields.get("locations") or []) + extra = _extra_tags(target_tags, target_fields.get("language", "")) + for sfields, stags in sources: + locations.extend(sfields.get("locations") or []) + for t in _extra_tags(stags, sfields.get("language", "")): + if t not in extra: + extra.append(t) + return _normalize_locations(locations), extra + + +async def merge_snippets(user_id: int, target_id: int, source_ids: list[int]): + """Unify source snippets INTO the target: union their locations + tags onto + the canonical target, keep the target's scalar fields, trash the sources + (recoverable), re-embed the survivor. + + Share-aware and write-gated, like update: an editor/admin grant is enough, a + viewer grant is not. Raises PermissionError when the caller can read the + target but not write it. Every source must share the TARGET'S OWNER — + cross-owner merge stays out of scope (#231) — and must itself be writable; + sources failing either test are skipped rather than silently half-merged. + + Returns (merged_target_note, merged_source_ids), or None if the target isn't + a snippet the caller can see.""" + from scribe.services.access import can_write_note + target = await get_snippet(user_id, target_id) + if target is None: + return None + if not await can_write_note(user_id, target_id): + raise PermissionError( + f"snippet {target_id} is shared with you read-only — you can't merge " + f"into a record you can't edit" + ) + owner_id = target.user_id + + sources = [] + for sid in source_ids: + if sid == target_id: + continue + s = await get_snippet(user_id, sid) + # Same owner as the target, and writable by this caller. Merging trashes + # the source, so read access is not enough. + if s is None or s.user_id != owner_id: + continue + if not await can_write_note(user_id, sid): + continue + sources.append(s) + + tgt_fields = parse_snippet_fields(target.title, target.body, target.tags) + parsed_sources = [ + (parse_snippet_fields(s.title, s.body, s.tags), s.tags) for s in sources + ] + locations, extra_tags = merge_snippet_fields(tgt_fields, target.tags, parsed_sources) + + # Owner-scoped write, authorised above — same reason as update_snippet. + updated = await notes_svc.update_note( + owner_id, target_id, + body=compose_body( + code=tgt_fields["code"], language=tgt_fields["language"], + signature=tgt_fields["signature"], when_to_use=tgt_fields["when_to_use"], + locations=locations, + ), + tags=compose_tags(tgt_fields["language"], extra_tags), + ) + if updated is None: + return None + + # Retire the merged-in sources to the trash (recoverable). + from scribe.services.trash import delete as trash_delete + merged_ids: list[int] = [] + for s in sources: + batch = await trash_delete(owner_id, "note", s.id) + if batch is not None: + merged_ids.append(s.id) + + _embed_snippet(updated) + return updated, merged_ids diff --git a/tests/test_mcp_tool_processes.py b/tests/test_mcp_tool_processes.py index 9bb5a68..658ca9a 100644 --- a/tests/test_mcp_tool_processes.py +++ b/tests/test_mcp_tool_processes.py @@ -13,11 +13,14 @@ def _bind_user(): _user_id_ctx.reset(token) -def _fake_note(id=1, title="Drift Audit", note_type="process"): +def _fake_note(id=1, title="Drift Audit", note_type="process", user_id=7): n = MagicMock() n.id = id n.title = title n.note_type = note_type + # Must be a real int, not an auto-attribute: the provenance check compares it + # against the bound caller (7) to decide whether the record is shared. + n.user_id = user_id n.to_dict.return_value = {"id": id, "title": title, "note_type": note_type} return n @@ -53,6 +56,26 @@ async def test_get_process_returns_body_and_candidates(): out = await get_process("drift") assert out["id"] == 7 assert out["other_matches"] == [{"id": 9, "title": "Drift Audit Notes"}] + # The caller's own process carries no provenance marker — absence of the flag + # is what makes it read as theirs. + assert "shared" not in out + + +@pytest.mark.asyncio +async def test_get_process_flags_another_users_process(): + """A shared Process must arrive labelled. get_process's contract is 'follow + the returned body', so an unlabelled one would put someone else's procedure + in charge of the session.""" + note = _fake_note(id=7, user_id=9) + with patch("scribe.services.notes.resolve_process", + AsyncMock(return_value=(note, []))), \ + patch("scribe.services.access.describe_provenance", + AsyncMock(return_value={"shared": True, "owner": "alex", + "permission": "viewer"})): + from scribe.mcp.tools.processes import get_process + out = await get_process("deploy") + assert out["shared"] is True + assert out["owner"] == "alex" @pytest.mark.asyncio @@ -66,14 +89,34 @@ async def test_get_process_not_found_raises(): @pytest.mark.asyncio async def test_update_process_rejects_non_process_note(): + # Resolves share-aware now, so the patch target is get_note_for_user, which + # returns (note, permission). plain = _fake_note(id=3, note_type="note") - with patch("scribe.services.notes.get_note", - AsyncMock(return_value=plain)): + plain.deleted_at = None + with patch("scribe.services.notes.get_note_for_user", + AsyncMock(return_value=(plain, "owner"))): from scribe.mcp.tools.processes import update_process with pytest.raises(ValueError): await update_process(process_id=3, title="x") +@pytest.mark.asyncio +async def test_update_process_refuses_a_read_only_share_with_the_reason(): + """An editor grant lets the holder edit someone else's process; a viewer + grant must be refused, and saying "not found" about a process the caller can + open would just send them looking for a missing id.""" + theirs = _fake_note(id=5, user_id=9) + theirs.deleted_at = None + with patch("scribe.services.notes.get_note_for_user", + AsyncMock(return_value=(theirs, "viewer"))), \ + patch("scribe.services.access.can_write_note", AsyncMock(return_value=False)), \ + patch("scribe.services.notes.update_note", AsyncMock()) as mock_update: + from scribe.mcp.tools.processes import update_process + with pytest.raises(ValueError, match="read-only"): + await update_process(process_id=5, title="x") + mock_update.assert_not_awaited() + + def test_register_attaches_four_tools(): from scribe.mcp.tools import processes names: list[str] = [] diff --git a/tests/test_mcp_tool_search.py b/tests/test_mcp_tool_search.py index 4b1f180..fa6de6a 100644 --- a/tests/test_mcp_tool_search.py +++ b/tests/test_mcp_tool_search.py @@ -18,13 +18,18 @@ def _reset_user_ctx(): def _fake_note(*, id: int, title: str, body: str = "", - tags: list[str] | None = None, is_task: bool = False) -> MagicMock: + tags: list[str] | None = None, is_task: bool = False, + user_id: int = 7) -> MagicMock: note = MagicMock() note.id = id note.title = title note.body = body note.tags = tags or [] note.is_task = is_task + # A real int, matching the bound caller by default: results compare it to + # decide whether to attach a shared/owner marker, and an auto-MagicMock would + # read as "someone else's" and send the tool looking up a username. + note.user_id = user_id return note diff --git a/tests/test_mcp_tool_snippets.py b/tests/test_mcp_tool_snippets.py new file mode 100644 index 0000000..adadc80 --- /dev/null +++ b/tests/test_mcp_tool_snippets.py @@ -0,0 +1,228 @@ +"""Tests for MCP snippet tools — patches the service layer.""" +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from scribe.mcp._context import _user_id_ctx + + +@pytest.fixture(autouse=True) +def _bind_user(): + token = _user_id_ctx.set(7) + yield + _user_id_ctx.reset(token) + + +def _fake_snippet(user_id: int = 7): + n = MagicMock() + n.id = 1 + n.title = "debounce — rate-limit a callback" + n.body = "```js\nreturn 1\n```\n" + n.tags = ["js", "snippet"] + n.note_type = "snippet" + # Real int, matching the bound caller by default. The tools compare it to + # decide whether to attach a shared/owner marker; an auto-MagicMock would read + # as another user's record and send them off to look up a username. + n.user_id = user_id + n.to_dict.return_value = { + "id": 1, "title": n.title, "note_type": "snippet", "tags": n.tags, + } + return n + + +@pytest.mark.asyncio +async def test_create_snippet_requires_name_and_code(): + from scribe.mcp.tools.snippets import create_snippet + with pytest.raises(ValueError): + await create_snippet(name="", code="x") + with pytest.raises(ValueError): + await create_snippet(name="x", code=" ") + + +@pytest.mark.asyncio +async def test_create_snippet_records_and_returns_parsed(): + created = _fake_snippet() + with patch("scribe.services.dedup.find_duplicate_note", AsyncMock(return_value=None)), \ + patch("scribe.services.snippets.create_snippet", + AsyncMock(return_value=created)) as mock_create: + from scribe.mcp.tools.snippets import create_snippet + out = await create_snippet( + name="debounce", code="return 1", language="js", + when_to_use="rate-limit a callback", + ) + assert out["note_type"] == "snippet" + assert out["snippet"]["name"] == "debounce" + assert out["snippet"]["language"] == "js" + assert mock_create.await_args.kwargs["name"] == "debounce" + + +@pytest.mark.asyncio +async def test_create_snippet_dedup_blocks_and_labels_snippet(): + dup = MagicMock(id=99) + with patch("scribe.services.dedup.find_duplicate_note", AsyncMock(return_value=dup)), \ + patch("scribe.services.dedup.duplicate_response", + MagicMock(return_value={"duplicate": True, "existing_id": 99})) as mock_resp, \ + patch("scribe.services.snippets.create_snippet", AsyncMock()) as mock_create: + from scribe.mcp.tools.snippets import create_snippet + out = await create_snippet(name="debounce", code="return 1") + assert out["duplicate"] is True + mock_create.assert_not_awaited() + assert mock_resp.call_args.args[1] == "snippet" + + +@pytest.mark.asyncio +async def test_get_snippet_not_found_raises(): + with patch("scribe.services.snippets.get_snippet", AsyncMock(return_value=None)): + from scribe.mcp.tools.snippets import get_snippet + with pytest.raises(ValueError): + await get_snippet(123) + + +@pytest.mark.asyncio +async def test_update_snippet_missing_raises(): + with patch("scribe.services.snippets.update_snippet", AsyncMock(return_value=None)): + from scribe.mcp.tools.snippets import update_snippet + with pytest.raises(ValueError): + await update_snippet(123, name="x") + + +@pytest.mark.asyncio +async def test_update_snippet_empty_string_clears_a_field(): + # An omitted field must stay None ("leave alone"), but an explicit empty + # string has to reach the service as "" so a stale field can be removed. + updated = _fake_snippet() + with patch("scribe.services.snippets.update_snippet", + AsyncMock(return_value=updated)) as mock_update: + from scribe.mcp.tools.snippets import update_snippet + await update_snippet(1, signature="") + kwargs = mock_update.await_args.kwargs + assert kwargs["signature"] == "" # cleared + assert kwargs["name"] is None # untouched + + +@pytest.mark.asyncio +async def test_update_snippet_project_id_conventions(): + from scribe.services import snippets as snippets_svc + updated = _fake_snippet() + cases = {0: snippets_svc.UNSET, -1: None, 5: 5} + for given, expected in cases.items(): + with patch("scribe.services.snippets.update_snippet", + AsyncMock(return_value=updated)) as mock_update: + from scribe.mcp.tools.snippets import update_snippet + await update_snippet(1, project_id=given) + assert mock_update.await_args.kwargs["project_id"] is expected + + +@pytest.mark.asyncio +async def test_create_and_update_pass_locations_through(): + locs = [{"repo": "a", "path": "a.py", "symbol": "f"}, + {"repo": "b", "path": "b.py", "symbol": "g"}] + created = _fake_snippet() + with patch("scribe.services.dedup.find_duplicate_note", AsyncMock(return_value=None)), \ + patch("scribe.services.snippets.create_snippet", + AsyncMock(return_value=created)) as mock_create: + from scribe.mcp.tools.snippets import create_snippet + await create_snippet(name="f", code="x", locations=locs) + assert mock_create.await_args.kwargs["locations"] == locs + + with patch("scribe.services.snippets.update_snippet", + AsyncMock(return_value=created)) as mock_update: + from scribe.mcp.tools.snippets import update_snippet + await update_snippet(1, locations=locs) + assert mock_update.await_args.kwargs["locations"] == locs + + +@pytest.mark.asyncio +async def test_update_snippet_read_only_share_says_why(): + """A viewer grant must be refused with the REAL reason. "Not found" would be + a lie about a record the caller can plainly open, and would send an agent + hunting for a missing id instead of recording its own version.""" + with patch("scribe.services.snippets.update_snippet", + AsyncMock(side_effect=PermissionError( + "snippet 1 is shared with you read-only — ask its owner for " + "edit access, or record your own version"))): + from scribe.mcp.tools.snippets import update_snippet + with pytest.raises(ValueError, match="read-only"): + await update_snippet(1, name="x") + + +@pytest.mark.asyncio +async def test_merge_snippets_read_only_target_says_why(): + with patch("scribe.services.snippets.merge_snippets", + AsyncMock(side_effect=PermissionError( + "snippet 1 is shared with you read-only — you can't merge " + "into a record you can't edit"))): + from scribe.mcp.tools.snippets import merge_snippets + with pytest.raises(ValueError, match="read-only"): + await merge_snippets(target_id=1, source_ids=[2]) + + +@pytest.mark.asyncio +async def test_delete_snippet_retires_or_raises(): + from scribe.mcp.tools.snippets import delete_snippet + with patch("scribe.services.snippets.delete_snippet", AsyncMock(return_value=True)): + assert await delete_snippet(1) == {"deleted": True, "id": 1} + with patch("scribe.services.snippets.delete_snippet", AsyncMock(return_value=False)): + with pytest.raises(ValueError): + await delete_snippet(404) + + +@pytest.mark.asyncio +async def test_list_snippets_defaults_to_every_project(): + with patch("scribe.services.snippets.list_snippets", + AsyncMock(return_value=([], 0))) as mock_list: + from scribe.mcp.tools.snippets import list_snippets + await list_snippets(q="debounce") + assert mock_list.await_args.kwargs["project_id"] is None + await list_snippets(q="debounce", project_id=3) + assert mock_list.await_args.kwargs["project_id"] == 3 + + +@pytest.mark.asyncio +async def test_merge_snippets_requires_a_source(): + from scribe.mcp.tools.snippets import merge_snippets + with pytest.raises(ValueError): + await merge_snippets(target_id=1, source_ids=[]) + with pytest.raises(ValueError): + await merge_snippets(target_id=1, source_ids=[1]) # only self → nothing to merge + + +@pytest.mark.asyncio +async def test_merge_snippets_returns_survivor_and_merged_ids(): + survivor = _fake_snippet() + with patch("scribe.services.snippets.merge_snippets", + AsyncMock(return_value=(survivor, [2, 3]))) as mock_merge: + from scribe.mcp.tools.snippets import merge_snippets + out = await merge_snippets(target_id=1, source_ids=[2, 3, 1]) + assert out["merged_ids"] == [2, 3] + assert out["snippet"]["name"] == "debounce" + # target_id passed through; self-id filtered out of the source list. + assert mock_merge.await_args.args[1] == 1 + assert mock_merge.await_args.args[2] == [2, 3] + + +@pytest.mark.asyncio +async def test_merge_snippets_not_found_raises(): + with patch("scribe.services.snippets.merge_snippets", AsyncMock(return_value=None)): + from scribe.mcp.tools.snippets import merge_snippets + with pytest.raises(ValueError): + await merge_snippets(target_id=1, source_ids=[2]) + + +def test_register_attaches_all_tools(): + from scribe.mcp.tools import snippets + names: list[str] = [] + + class FakeMcp: + def tool(self, name): + names.append(name) + + def deco(fn): + return fn + return deco + + snippets.register(FakeMcp()) + assert set(names) == { + "list_snippets", "create_snippet", "get_snippet", "update_snippet", + "delete_snippet", "merge_snippets", + } diff --git a/tests/test_retrieval_scopes.py b/tests/test_retrieval_scopes.py new file mode 100644 index 0000000..5eeaf29 --- /dev/null +++ b/tests/test_retrieval_scopes.py @@ -0,0 +1,148 @@ +"""Every retrieval path must declare the right visibility scope. + +`semantic_search_notes` serves three different kinds of act, and the correct +scope differs for each. Getting one wrong is silent — the code still works, it +just sees too much or too little — so each caller is pinned here: + + explicit search → "read" the caller asked; reach everything they may read + auto-injection → "browse" nobody asked; never a one-to-one shared record + near-duplicate → "own" a verdict that blocks a write can't hinge on + someone else's notes + +The keyword and semantic halves of one hybrid search must also agree, or a +shared record would be findable by wording and invisible by meaning. +""" +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + + +def _note(id=1, user_id=7, title="A note"): + n = MagicMock() + n.id = id + n.user_id = user_id + n.title = title + n.body = "body" + n.tags = [] + n.is_task = False + n.note_type = "note" + return n + + +@pytest.fixture +def spy(): + """Patch semantic_search_notes everywhere it's imported and capture kwargs.""" + mock = AsyncMock(return_value=[]) + targets = [ + "scribe.services.embeddings.semantic_search_notes", + "scribe.services.plugin_context.semantic_search_notes", + "scribe.mcp.tools.search.semantic_search_notes", + "scribe.routes.search.semantic_search_notes", + ] + patches = [patch(t, mock) for t in targets] + for p in patches: + p.start() + yield mock + for p in patches: + p.stop() + + +@pytest.mark.asyncio +async def test_mcp_search_uses_read_scope(spy): + from scribe.mcp._context import _user_id_ctx + token = _user_id_ctx.set(7) + try: + with patch("scribe.services.retrieval_telemetry.record_retrieval", MagicMock()): + from scribe.mcp.tools.search import search + await search(q="debounce") + finally: + _user_id_ctx.reset(token) + assert spy.await_args.kwargs["scope"] == "read" + + +@pytest.mark.asyncio +async def test_auto_inject_uses_browse_scope(spy): + """The one retrieval nobody requested. A record shared one-to-one with the + operator must never arrive this way.""" + from scribe.services import plugin_context + with patch.object(plugin_context, "get_autoinject_config", + AsyncMock(return_value={"enabled": True, "threshold": 0.55, + "top_k": 5})), \ + patch.object(plugin_context, "record_retrieval", MagicMock()): + await plugin_context.build_autoinject_hint(7, "how do I debounce") + assert spy.await_args.kwargs["scope"] == "browse" + + +@pytest.mark.asyncio +async def test_dedup_gate_uses_own_scope(spy): + """An empty title skips the title signal and goes straight to the semantic + one — no session needed. The gate blocks a create and tells the caller to + update the match instead, so matching another user's record would refuse + their write and point them at something they may not be able to edit.""" + from scribe.services import dedup + await dedup.find_duplicate_note( + 7, "", body="x" * 400, project_id=None, is_task=False, + ) + assert spy.await_args.kwargs["scope"] == "own" + + +@pytest.mark.asyncio +async def test_hybrid_search_halves_agree_on_scope(): + """The keyword half and the semantic half of one search must see equally.""" + from scribe.services import knowledge + + captured: dict[str, object] = {} + + async def fake_semantic(**kwargs): + captured["scope"] = kwargs.get("scope") + return [] + + from scribe.models.note import Note + with patch("scribe.services.embeddings.semantic_search_notes", + AsyncMock(side_effect=fake_semantic)), \ + patch.object(knowledge, "readable_notes_clause", + MagicMock(return_value=(Note.user_id == 7))) as read_clause, \ + patch.object(knowledge, "async_session") as sess: + session = AsyncMock() + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock(return_value=False) + result = MagicMock() + result.scalars.return_value.all.return_value = [] + session.execute = AsyncMock(return_value=result) + sess.return_value = session + + await knowledge.query_knowledge( + user_id=7, note_type=None, tags=[], sort="modified", + q="debounce", limit=10, offset=0, + ) + + # Keyword half took the read scope... + read_clause.assert_called_once_with(7) + # ...and so did the semantic half. + assert captured["scope"] == "read" + + +@pytest.mark.asyncio +async def test_injected_menu_attributes_another_users_note(): + """Auto-inject can surface a collaborator's note via a shared project. The + menu line is the only provenance the agent sees, so it has to name them.""" + from scribe.services import plugin_context + + mine, theirs = _note(1, user_id=7, title="Mine"), _note(2, user_id=9, title="Theirs") + with patch.object(plugin_context, "semantic_search_notes", + AsyncMock(return_value=[(0.9, theirs), (0.88, mine)])), \ + patch.object(plugin_context, "get_autoinject_config", + AsyncMock(return_value={"enabled": True, "threshold": 0.5, + "top_k": 5})), \ + patch.object(plugin_context, "record_retrieval", MagicMock()), \ + patch.object(plugin_context, "owner_names_for", + AsyncMock(return_value={9: "alex"})): + out = await plugin_context.build_autoinject_hint(7, "anything") + + lines = out["context"].splitlines() + theirs_line = next(ln for ln in lines if "#2" in ln) + mine_line = next(ln for ln in lines if "#1" in ln) + assert "shared by alex" in theirs_line + assert "suggestion" in theirs_line + # The operator's own note stays unadorned — absence of a marker is the signal. + assert "shared by" not in mine_line diff --git a/tests/test_routes_snippets.py b/tests/test_routes_snippets.py new file mode 100644 index 0000000..c011be2 --- /dev/null +++ b/tests/test_routes_snippets.py @@ -0,0 +1,84 @@ +"""Structural tests for the snippets blueprint — registration + handler/service +contracts. Full HTTP integration needs a live DB + auth the unit env lacks.""" +import inspect + + +def test_snippets_blueprint_registered(): + from scribe.routes.snippets import snippets_bp + assert snippets_bp.name == "snippets" + assert snippets_bp.url_prefix == "/api/snippets" + + +def test_snippets_blueprint_registered_in_app(): + from scribe.app import create_app + app = create_app() + assert "snippets" in app.blueprints + + +def test_snippet_handlers_callable(): + from scribe.routes import snippets as routes + for name in ( + "list_snippets_route", "create_snippet_route", "get_snippet_route", + "update_snippet_route", "delete_snippet_route", "merge_snippet_route", + ): + assert callable(getattr(routes, name)) + + +def test_service_functions_take_user_id(): + """Routes must call snippet services with user_id — verify the contract.""" + from scribe.services import snippets as svc + for fn_name in ( + "create_snippet", "list_snippets", "get_snippet", "update_snippet", + "delete_snippet", "merge_snippets", + ): + fn = getattr(svc, fn_name) + assert callable(fn) + assert "user_id" in inspect.signature(fn).parameters + + +def test_agent_and_web_surfaces_stay_at_parity(): + """The MCP tools and the REST routes are two callers of one service; a + capability on one has to exist on the other (rule #33). This guard exists + because they drifted apart once: MCP had no delete or `locations`, and the + web side had no `system_ids` and no near-duplicate gate.""" + from scribe.mcp.tools import snippets as tools + from scribe.routes import snippets as routes + + # Every write verb the web surface offers, the agent surface offers too. + for verb in ("create", "get", "list", "update", "delete", "merge"): + assert callable(getattr(tools, f"{verb}_snippets", None) or + getattr(tools, f"{verb}_snippet", None)), f"MCP lacks {verb}" + + # Multi-location records are reachable from both. + assert "locations" in inspect.signature(tools.create_snippet).parameters + assert "locations" in inspect.signature(tools.update_snippet).parameters + assert "locations" in inspect.getsource(routes.create_snippet_route) + assert "locations" in inspect.getsource(routes.update_snippet_route) + + # System association and the duplicate gate reach both. + assert "system_ids" in inspect.signature(tools.create_snippet).parameters + for route in (routes.create_snippet_route, routes.update_snippet_route): + assert "system_ids" in inspect.getsource(route) + assert "find_duplicate_note" in inspect.getsource(routes.create_snippet_route) + + +def test_project_scoping_reaches_every_caller(): + """A snippet search has to be narrowable to one project from both surfaces.""" + from scribe.mcp.tools import snippets as tools + from scribe.routes import snippets as routes + from scribe.services import snippets as svc + assert "project_id" in inspect.signature(svc.list_snippets).parameters + assert "project_id" in inspect.signature(tools.list_snippets).parameters + assert "project_id" in inspect.getsource(routes.list_snippets_route) + + +def test_update_field_map_matches_service_kwargs(): + """Every field the PATCH route forwards must be a real update_snippet kwarg + (rule #33 interface-contract parity).""" + from scribe.routes import snippets as routes + from scribe.services import snippets as svc + params = inspect.signature(svc.update_snippet).parameters + for field in routes._STR_FIELDS: + assert field in params, f"update_snippet has no '{field}' kwarg" + assert "tags" in params + assert "project_id" in params diff --git a/tests/test_services_access_visibility.py b/tests/test_services_access_visibility.py new file mode 100644 index 0000000..1b9a6d0 --- /dev/null +++ b/tests/test_services_access_visibility.py @@ -0,0 +1,134 @@ +"""The two ACL predicates behind list queries. + +`get_note_permission` answers "may I read THIS note?" one row at a time, which a +list query can't use. These express the same resolution as set membership. They +are pure SQL builders — group membership is a subquery rather than a fetched +list, so they need no session and callers' unit tests need not know they exist. + +Two scopes, deliberately different (decision note 2094): + readable_* — everything the ACL permits, for explicit acts (a typed search, a + fetch by id). + browsable_* — owner + project access only, for passive surfaces (browse lists, + facet counts, the process→skill manifest). + +Assertions compile each clause to SQL and inspect its shape. +""" +import pytest + +from scribe.services.access import ( + browsable_notes_clause, + notes_visibility_clause, + readable_notes_clause, +) + + +def _sql(clause) -> str: + return str(clause.compile(compile_kwargs={"literal_binds": True})) + + +def _read(user_id: int = 7) -> str: + return _sql(readable_notes_clause(user_id)) + + +def _browse(user_id: int = 7) -> str: + return _sql(browsable_notes_clause(user_id)) + + +# --- read scope -------------------------------------------------------------- + +def test_read_scope_covers_ownership_and_every_share_path(): + sql = _read() + assert "notes.user_id = 7" in sql # 1. ownership + assert "note_shares" in sql # 2/3. direct or group note share + assert "notes.project_id IN" in sql # 4. inherited from a shared project + assert "project_shares" in sql + + +def test_read_scope_resolves_group_membership_in_sql(): + """Group ids are a subquery, not a pre-fetched list — that's what keeps this + a pure function with no session of its own.""" + sql = _read() + assert "group_memberships.user_id = 7" in sql + # Both the note-level and project-level share lookups consult it. Count FROM + # clauses rather than bare occurrences — each rendered subquery names the + # table in SELECT, FROM and WHERE — and compare loosely, since the assertion + # is about the arms existing, not about SQLAlchemy's formatting. + assert sql.count("FROM group_memberships") >= 2 + assert _browse().count("FROM group_memberships") >= 1 + + +def test_read_scope_is_never_the_whole_table(): + """Guard against the predicate degrading to always-true, which would expose + every user's notes to every other user.""" + sql = _read().lower() + assert " true" not in sql + assert "1 = 1" not in sql + + +# --- browse scope: the trust boundary --------------------------------------- + +def test_browse_scope_excludes_direct_note_shares(): + """The whole point of the narrower scope. If `note_shares` leaks in here, a + record someone shared one-to-one with the operator lands in their own browse + list, facet counts and skill manifest as though they had recorded it.""" + assert "note_shares" not in _browse() + + +def test_browse_scope_keeps_ownership_and_project_access(): + sql = _browse() + assert "notes.user_id = 7" in sql # your own records + assert "project_shares" in sql # a project shared with you + assert "projects" in sql # a project you own + + +def test_browse_scope_is_strictly_narrower_than_read_scope(): + """Browse must never surface something read scope wouldn't also allow, or a + list could show a record the caller cannot then open.""" + browse, read = _browse(), _read() + assert "note_shares" in read and "note_shares" not in browse + for arm in ("notes.user_id = 7", "project_shares"): + assert arm in browse and arm in read + + +def test_browse_scope_is_never_the_whole_table(): + sql = _browse().lower() + assert " true" not in sql + assert "1 = 1" not in sql + + +# --- the scope resolver ------------------------------------------------------ + +def test_own_scope_is_ownership_alone(): + """The near-duplicate gate depends on this: its verdict must not turn on + another person's records, or it would refuse a write and point the caller at + something they may not be able to edit.""" + sql = _sql(notes_visibility_clause(7, "own")) + assert sql == "notes.user_id = 7" + + +def test_scope_resolver_maps_to_the_right_clauses(): + assert _sql(notes_visibility_clause(7, "browse")) == _browse() + assert _sql(notes_visibility_clause(7, "read")) == _read() + + +def test_scope_defaults_to_the_narrowest(): + """A caller that forgets to choose must be wrong in the safe direction.""" + assert _sql(notes_visibility_clause(7)) == _sql(notes_visibility_clause(7, "own")) + + +def test_unknown_scope_is_rejected_loudly(): + """Silently falling back would turn a typo into a data-exposure bug.""" + with pytest.raises(ValueError): + notes_visibility_clause(7, "everything") + + +@pytest.mark.parametrize("clause_fn", [readable_notes_clause, browsable_notes_clause]) +def test_clauses_are_pure_builders(clause_fn): + """Synchronous and side-effect free — no coroutine, no session of their own. + + This is the property that keeps them usable: when they opened their own + session, every unrelated service test had to know they existed and stub them, + and four test modules broke the moment a service started calling one.""" + import inspect + assert not inspect.iscoroutinefunction(clause_fn) + assert _sql(clause_fn(7)) # builds without touching a database diff --git a/tests/test_services_plugin_context.py b/tests/test_services_plugin_context.py index 0c30b4f..6281a47 100644 --- a/tests/test_services_plugin_context.py +++ b/tests/test_services_plugin_context.py @@ -10,9 +10,13 @@ def _rule(rid, title, topic_id): return r -def _note(nid, title): +def _note(nid, title, user_id=1): n = MagicMock() n.id, n.title = nid, title + # Real int, defaulting to the caller used in these tests: the injected menu + # compares it to decide whether the line needs a "shared by …" attribution, + # and an auto-MagicMock would read as another user's note. + n.user_id = user_id return n diff --git a/tests/test_services_snippets.py b/tests/test_services_snippets.py new file mode 100644 index 0000000..55487a9 --- /dev/null +++ b/tests/test_services_snippets.py @@ -0,0 +1,173 @@ +"""Unit tests for the snippet serialize/parse helpers (pure functions, no DB).""" +from scribe.services import snippets as s + + +def test_compose_title_with_and_without_usage(): + assert s.compose_title("debounce", "rate-limit a callback") == "debounce — rate-limit a callback" + assert s.compose_title(" debounce ", "") == "debounce" + assert s.compose_title("debounce") == "debounce" + + +def test_compose_tags_lowercases_language_and_dedups(): + assert s.compose_tags("Python", ["util", "python"]) == ["python", "snippet", "util"] + assert s.compose_tags("", None) == ["snippet"] + assert s.compose_tags("vue", ["snippet"]) == ["vue", "snippet"] + + +def test_compose_body_includes_fields_and_fence(): + body = s.compose_body( + code="return 1", language="python", signature="f() -> int", + when_to_use="always", repo="scribe", path="a.py", symbol="f", + ) + assert "**When to use:** always" in body + assert "**Signature:** `f() -> int`" in body + assert "`scribe` · `a.py` · `f`" in body + assert "```python\nreturn 1\n```" in body + assert body.rstrip().endswith("```") + + +def test_compose_body_bare_code_only(): + body = s.compose_body(code="x = 1") + assert body.strip() == "```\nx = 1\n```" + + +def test_parse_round_trips_a_composed_snippet(): + title = s.compose_title("useDebouncedRef", "debounce a reactive ref") + body = s.compose_body( + code="const x = 1", language="ts", signature="useDebouncedRef(v, ms)", + when_to_use="debounce a reactive ref", repo="scribe", + path="frontend/src/composables/x.ts", symbol="useDebouncedRef", + ) + got = s.parse_snippet_fields(title, body, ["ts", "snippet"]) + assert got["name"] == "useDebouncedRef" + assert got["when_to_use"] == "debounce a reactive ref" + assert got["signature"] == "useDebouncedRef(v, ms)" + assert got["language"] == "ts" + assert got["code"] == "const x = 1" + assert got["repo"] == "scribe" + assert got["path"] == "frontend/src/composables/x.ts" + assert got["symbol"] == "useDebouncedRef" + + +def test_parse_is_tolerant_of_plain_body(): + got = s.parse_snippet_fields("just a name", "no structure here", None) + assert got["name"] == "just a name" + assert got["when_to_use"] == "" + assert got["signature"] == "" + assert got["code"] == "" # never raises on an unstructured body + + +def test_parse_falls_back_to_tag_for_language(): + got = s.parse_snippet_fields("n — u", "```\ncode\n```", ["ruby", "snippet"]) + assert got["language"] == "ruby" + + +def test_parse_does_not_promote_a_caller_tag_to_language(): + # compose_tags puts the language FIRST, so a leading "snippet" marker means + # no language was recorded — the tags after it are the caller's own and must + # not be mistaken for one (which would also corrupt the code fence on the + # next update). + tags = s.compose_tags("", ["auth"]) + assert tags == ["snippet", "auth"] + got = s.parse_snippet_fields("n — u", s.compose_body(code="x = 1"), tags) + assert got["language"] == "" + + +def test_caller_tag_survives_an_update_round_trip_without_a_language(): + # Regression: the tag used to be read back as the language, then dropped + # from the extra-tag set on re-compose — so it silently disappeared. + tags = s.compose_tags("", ["auth"]) + fields = s.parse_snippet_fields("n — u", s.compose_body(code="x = 1"), tags) + extra = [t for t in tags if t not in (s.SNIPPET_TAG, fields["language"])] + assert s.compose_tags(fields["language"], extra) == ["snippet", "auth"] + + +def test_compose_body_multi_location_renders_bullet_list(): + body = s.compose_body( + code="x = 1", + locations=[ + {"repo": "scribe", "path": "a.py", "symbol": "f"}, + {"repo": "web", "path": "b.ts", "symbol": "g"}, + ], + ) + assert "**Locations:**" in body + assert "- `scribe` · `a.py` · `f`" in body + assert "- `web` · `b.ts` · `g`" in body + assert "**Location:**" not in body.replace("**Locations:**", "") + + +def test_compose_body_single_location_via_list_uses_singular_label(): + body = s.compose_body( + code="x = 1", locations=[{"repo": "scribe", "path": "a.py", "symbol": "f"}], + ) + assert "**Location:** `scribe` · `a.py` · `f`" in body + assert "**Locations:**" not in body + + +def test_normalize_locations_drops_empty_and_dedups(): + got = s._normalize_locations([ + {"repo": "scribe", "path": "a.py", "symbol": "f"}, + {"repo": "", "path": "", "symbol": ""}, # dropped (empty) + {"repo": "scribe", "path": "a.py", "symbol": "f"}, # dropped (dup) + {"repo": "web", "path": "", "symbol": ""}, + ]) + assert got == [ + {"repo": "scribe", "path": "a.py", "symbol": "f"}, + {"repo": "web", "path": "", "symbol": ""}, + ] + + +def test_parse_round_trips_multi_location(): + body = s.compose_body( + code="const x = 1", language="ts", + locations=[ + {"repo": "scribe", "path": "a.ts", "symbol": "f"}, + {"repo": "web", "path": "b.ts", "symbol": "g"}, + ], + ) + got = s.parse_snippet_fields("n — u", body, ["ts", "snippet"]) + assert got["locations"] == [ + {"repo": "scribe", "path": "a.ts", "symbol": "f"}, + {"repo": "web", "path": "b.ts", "symbol": "g"}, + ] + # repo/path/symbol mirror the first location for back-compat. + assert (got["repo"], got["path"], got["symbol"]) == ("scribe", "a.ts", "f") + + +def test_parse_legacy_single_location_line_still_works(): + # A body written by the pre-multi-location serializer. + body = "**Location:** `scribe` · `a.py` · `f`\n\n```py\nx = 1\n```\n" + got = s.parse_snippet_fields("n — u", body, None) + assert got["locations"] == [{"repo": "scribe", "path": "a.py", "symbol": "f"}] + assert (got["repo"], got["path"], got["symbol"]) == ("scribe", "a.py", "f") + + +def test_merge_snippet_fields_unions_locations_and_tags(): + target_fields = {"language": "py", "locations": [{"repo": "a", "path": "a.py", "symbol": "f"}]} + src1 = ({"language": "py", "locations": [{"repo": "b", "path": "b.py", "symbol": "g"}]}, + ["py", "snippet", "helper"]) + src2 = ({"language": "py", "locations": [{"repo": "a", "path": "a.py", "symbol": "f"}]}, # dup loc + ["snippet"]) + locs, extra = s.merge_snippet_fields(target_fields, ["py", "snippet", "core"], [src1, src2]) + # target location first, then unique source locations; dup dropped. + assert locs == [ + {"repo": "a", "path": "a.py", "symbol": "f"}, + {"repo": "b", "path": "b.py", "symbol": "g"}, + ] + # extra tags unioned, language + "snippet" markers excluded. + assert extra == ["core", "helper"] + + +def test_snippet_to_dict_includes_parsed_fields(): + class FakeNote: + title = "debounce — rate-limit" + body = "```js\ncode\n```\n" + tags = ["js", "snippet"] + + def to_dict(self): + return {"id": 1, "title": self.title, "note_type": "snippet", "tags": self.tags} + + data = s.snippet_to_dict(FakeNote()) + assert data["snippet"]["name"] == "debounce" + assert data["snippet"]["language"] == "js" + assert data["snippet"]["code"] == "code" diff --git a/tests/test_shared_provenance.py b/tests/test_shared_provenance.py new file mode 100644 index 0000000..08c7c1b --- /dev/null +++ b/tests/test_shared_provenance.py @@ -0,0 +1,107 @@ +"""Shared records must announce themselves. + +A record another user owns is that person's suggestion, not the operator's +settled practice — and the two are indistinguishable unless every surface that +hands one over says whose it is. These tests hold that line at the labelling +helpers and at the process→skill manifest, which is the most consequential +passive surface Scribe has (each entry becomes an auto-surfacing skill file). +""" +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + + +def _session_returning_rows(rows): + result = MagicMock() + result.all.return_value = rows + session = AsyncMock() + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock(return_value=False) + session.execute = AsyncMock(return_value=result) + return session + + +@pytest.mark.asyncio +async def test_label_marks_only_foreign_rows(): + from scribe.services.access import label_shared_items + items = [ + {"id": 1, "user_id": 7}, # mine + {"id": 2, "user_id": 9}, # someone else's + ] + with patch("scribe.services.access.async_session") as cls: + cls.return_value = _session_returning_rows([(9, "alex")]) + out = await label_shared_items(user_id=7, items=items) + + mine, theirs = out + # An unmarked row must be unambiguously the caller's own. + assert "shared" not in mine and "owner" not in mine + assert theirs["shared"] is True + assert theirs["owner"] == "alex" + + +@pytest.mark.asyncio +async def test_label_skips_the_query_when_everything_is_mine(): + """No foreign rows means no user lookup at all — labelling shouldn't cost a + query on the common single-owner path.""" + from scribe.services.access import label_shared_items + with patch("scribe.services.access.async_session") as cls: + out = await label_shared_items(user_id=7, items=[{"id": 1, "user_id": 7}]) + cls.assert_not_called() + assert out == [{"id": 1, "user_id": 7}] + + +@pytest.mark.asyncio +async def test_provenance_is_empty_for_your_own_record(): + from scribe.services.access import describe_provenance + note = MagicMock(id=1, user_id=7) + assert await describe_provenance(user_id=7, note=note) == {} + + +@pytest.mark.asyncio +async def test_provenance_names_the_owner_and_permission(): + from scribe.services.access import describe_provenance + note = MagicMock(id=1, user_id=9) + owner = MagicMock(username="alex") + session = AsyncMock() + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock(return_value=False) + session.get = AsyncMock(return_value=owner) + with patch("scribe.services.access.async_session") as cls, \ + patch("scribe.services.access.get_note_permission", + AsyncMock(return_value="viewer")): + cls.return_value = session + got = await describe_provenance(user_id=7, note=note) + assert got == {"shared": True, "owner": "alex", "permission": "viewer"} + + +@pytest.mark.asyncio +async def test_shared_process_skill_stub_never_claims_to_be_the_operators(): + """The stub description IS the skill's auto-surface trigger and the only + provenance a reader sees. For a shared Process it must name the author and + require a go-ahead — never read as "the operator's saved process".""" + from scribe.services import plugin_context + + items = [ + {"id": 1, "title": "Drift Audit", "snippet": "Sweep for drift.", "user_id": 7}, + {"id": 2, "title": "Deploy Dance", "snippet": "Ship it.", "user_id": 9}, + ] + with patch.object(plugin_context.knowledge_svc, "query_knowledge", + AsyncMock(return_value=(items, 2))), \ + patch.object(plugin_context, "label_shared_items", + AsyncMock(side_effect=lambda uid, its: [ + it if it["user_id"] == uid + else {**it, "shared": True, "owner": "alex"} + for it in its + ])): + out = await plugin_context.build_process_manifest(user_id=7) + + own, shared = out["processes"] + + assert "operator's saved Scribe process" in own["description"] + assert "shared" not in own + + assert shared["shared"] is True + assert shared["owner"] == "alex" + assert "operator's saved Scribe process" not in shared["description"] + assert "alex" in shared["description"] + assert "go-ahead" in shared["description"] diff --git a/tests/test_shared_write_access.py b/tests/test_shared_write_access.py new file mode 100644 index 0000000..c68abb7 --- /dev/null +++ b/tests/test_shared_write_access.py @@ -0,0 +1,111 @@ +"""Editing a shared record: an editor grant is enough, a viewer grant is not. + +The agent and web paths used to disagree — and worse, within the agent path +`delete_snippet` honoured editor shares while `update_snippet` did not, so a +colleague's snippet could be destroyed through an agent but not improved. Both +now resolve the read scope and then require WRITE, which is what the sharing UI +already promises: viewer / editor / admin are distinct grants. + +A record readable-but-not-writable raises PermissionError rather than reporting +"not found" — the caller can plainly open it, so not-found would be a lie that +sends them hunting for a missing id. +""" +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + + +def _snippet(id=1, owner=9): + n = MagicMock() + n.id = id + n.user_id = owner + n.title = "formatDuration — humanize a millisecond count" + n.body = "```ts\nexport const f = 1\n```\n" + n.tags = ["ts", "snippet"] + n.note_type = "snippet" + n.deleted_at = None + return n + + +@pytest.mark.asyncio +async def test_editor_share_can_update_and_writes_as_the_owner(): + """The underlying note update is owner-scoped, so a shared editor's own id + would match nothing — the authorised write has to be performed as the owner.""" + from scribe.services import snippets as svc + note = _snippet(owner=9) + updated = _snippet(owner=9) + with patch.object(svc, "get_snippet", AsyncMock(return_value=note)), \ + patch("scribe.services.access.can_write_note", AsyncMock(return_value=True)), \ + patch.object(svc.notes_svc, "update_note", + AsyncMock(return_value=updated)) as mock_update, \ + patch.object(svc, "_embed_snippet", MagicMock()): + got = await svc.update_snippet(7, 1, name="formatDuration") + + assert got is updated + # Positional user_id is the OWNER (9), not the caller (7). + assert mock_update.await_args.args[0] == 9 + + +@pytest.mark.asyncio +async def test_viewer_share_cannot_update(): + from scribe.services import snippets as svc + with patch.object(svc, "get_snippet", AsyncMock(return_value=_snippet())), \ + patch("scribe.services.access.can_write_note", AsyncMock(return_value=False)), \ + patch.object(svc.notes_svc, "update_note", AsyncMock()) as mock_update: + with pytest.raises(PermissionError, match="read-only"): + await svc.update_snippet(7, 1, name="nope") + mock_update.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_viewer_share_cannot_delete(): + """delete_snippet returns False rather than raising — its contract is boolean + — but a viewer must not be able to bin someone else's record.""" + from scribe.services import snippets as svc + with patch.object(svc, "get_snippet", AsyncMock(return_value=_snippet())), \ + patch("scribe.services.access.can_write_note", AsyncMock(return_value=False)), \ + patch("scribe.services.trash.delete", AsyncMock()) as mock_trash: + assert await svc.delete_snippet(7, 1) is False + mock_trash.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_unreadable_record_is_not_found_not_forbidden(): + """A record the caller can't see at all must NOT be distinguishable from one + that doesn't exist — otherwise the error itself confirms it exists.""" + from scribe.services import snippets as svc + with patch.object(svc, "get_snippet", AsyncMock(return_value=None)): + assert await svc.update_snippet(7, 404, name="x") is None + assert await svc.delete_snippet(7, 404) is False + + +@pytest.mark.asyncio +async def test_merge_requires_write_on_target(): + from scribe.services import snippets as svc + with patch.object(svc, "get_snippet", AsyncMock(return_value=_snippet())), \ + patch("scribe.services.access.can_write_note", AsyncMock(return_value=False)): + with pytest.raises(PermissionError, match="read-only"): + await svc.merge_snippets(7, 1, [2]) + + +@pytest.mark.asyncio +async def test_merge_skips_sources_owned_by_someone_else(): + """Cross-owner merge stays out of scope (#231): a source belonging to a + different owner than the target is skipped, not silently folded in and + trashed.""" + from scribe.services import snippets as svc + target = _snippet(id=1, owner=9) + same_owner = _snippet(id=2, owner=9) + other_owner = _snippet(id=3, owner=42) + + async def fake_get(_uid, sid): + return {1: target, 2: same_owner, 3: other_owner}.get(sid) + + with patch.object(svc, "get_snippet", AsyncMock(side_effect=fake_get)), \ + patch("scribe.services.access.can_write_note", AsyncMock(return_value=True)), \ + patch.object(svc.notes_svc, "update_note", AsyncMock(return_value=target)), \ + patch("scribe.services.trash.delete", AsyncMock(return_value=object())), \ + patch.object(svc, "_embed_snippet", MagicMock()): + _note, merged_ids = await svc.merge_snippets(7, 1, [2, 3]) + + assert merged_ids == [2]