Milestone 288 — pattern library + forge integration #111

Merged
bvandeusen merged 10 commits from dev into main 2026-08-16 16:37:12 -04:00
30 changed files with 3475 additions and 77 deletions
+186
View File
@@ -418,11 +418,66 @@ async function loadNotes() {
}
}
/* ── Pattern-library coverage (#2692) ─────────────────────────── */
interface CoverageGap {
dir: string;
uncovered: number;
total: number;
}
interface Coverage {
total: number;
recorded: number;
estimate: boolean;
computed_at: string;
repos: { repo: string; ref: string; total: number; recorded: number }[];
largest_gaps: CoverageGap[];
}
const coverage = ref<Coverage | null>(null);
const coverageConfigured = ref(false);
const coverageRefreshing = ref(false);
const coverageError = ref<string | null>(null);
/** Swallows failure like loadDesignSystems: no forge is the ordinary state
* for most installs, and this card must never break the project page. */
async function loadCoverage() {
try {
const res = await apiGet<{ configured: boolean; coverage: Coverage | null }>(
`/api/projects/${projectId.value}/coverage`
);
coverageConfigured.value = res.configured;
coverage.value = res.coverage;
} catch {
coverageConfigured.value = false;
coverage.value = null;
}
}
async function refreshCoverage() {
if (coverageRefreshing.value) return;
coverageRefreshing.value = true;
coverageError.value = null;
try {
const res = await apiPost<{ coverage: Coverage }>(
`/api/projects/${projectId.value}/coverage/refresh`,
{}
);
coverage.value = res.coverage;
} catch (e) {
coverageError.value =
e instanceof Error ? e.message : "Coverage refresh failed";
} finally {
coverageRefreshing.value = false;
}
}
onMounted(async () => {
await loadProject();
loadTasks();
loadNotes();
loadDesignSystems();
loadCoverage();
});
/** Populate the design-system picker. Swallows failure on purpose: with no
@@ -440,6 +495,7 @@ watch(projectId, async () => {
await loadProject();
loadTasks();
loadNotes();
loadCoverage();
});
watch(
@@ -592,6 +648,53 @@ async function confirmDelete() {
</div>
</div>
<!-- Pattern-library coverage (#2692) only when a forge is configured;
forge-less installs never see this card at all. -->
<div v-if="coverageConfigured" class="coverage-card">
<div class="coverage-head">
<span class="coverage-title">Pattern coverage</span>
<span class="coverage-estimate" title="Keyword-based extraction over- and under-counts; watch the trend, not the digit">estimate</span>
<span v-if="coverage" class="coverage-when">computed {{ relativeTime(coverage.computed_at) }}</span>
<button
class="btn-ghost btn-compact coverage-refresh"
:disabled="coverageRefreshing"
@click="refreshCoverage"
>
{{ coverageRefreshing ? "Measuring…" : "Refresh" }}
</button>
</div>
<template v-if="coverage">
<div class="coverage-numbers">
<span class="coverage-count">{{ coverage.recorded }}/{{ coverage.total }}</span>
<span class="coverage-label">shapes recorded</span>
</div>
<div
class="coverage-bar"
role="progressbar"
:aria-valuenow="coverage.recorded"
:aria-valuemin="0"
:aria-valuemax="coverage.total"
aria-label="Shapes with a recorded snippet"
>
<div
class="coverage-bar-fill"
:style="{ width: (coverage.total ? (coverage.recorded / coverage.total) * 100 : 0) + '%' }"
></div>
</div>
<div v-if="coverage.largest_gaps?.length" class="coverage-gaps">
<span class="coverage-gaps-label">Largest gaps:</span>
<span v-for="gap in coverage.largest_gaps" :key="gap.dir" class="coverage-gap-chip">
{{ gap.dir }} <span class="coverage-gap-count">{{ gap.uncovered }}</span>
</span>
</div>
</template>
<p v-else class="coverage-empty">
Not measured yet Refresh compares the bound repo's definitions
against recorded snippets.
</p>
<p v-if="coverageError" class="coverage-error">{{ coverageError }}</p>
</div>
<div class="project-body">
<!-- Edit panel -->
@@ -1037,6 +1140,89 @@ async function confirmDelete() {
.stat-done { background: color-mix(in srgb, var(--fs-success) 10%, transparent); color: var(--fs-success); border-color: color-mix(in srgb, var(--fs-success) 28%, transparent); }
.stat-notes { background: color-mix(in srgb, var(--fs-accent) 8%, transparent); color: var(--fs-accent); border-color: color-mix(in srgb, var(--fs-accent) 22%, transparent); }
/* ── Pattern-library coverage card ───────────────────────────── */
.coverage-card {
background: var(--fs-surface-raised);
border: 1px solid var(--fs-border-color);
border-radius: var(--fs-radius-lg);
padding: 0.75rem 1rem;
margin-bottom: 1.25rem;
display: flex;
flex-direction: column;
gap: 0.5rem;
}
.coverage-head {
display: flex;
align-items: center;
gap: 0.5rem;
}
.coverage-title {
font-size: 0.72rem;
font-weight: 500;
text-transform: uppercase;
letter-spacing: 0.06em;
color: var(--fs-text-tertiary);
}
.coverage-estimate {
font-size: 0.68rem;
padding: 0.05rem 0.4rem;
border-radius: var(--fs-radius-lg);
border: 1px solid var(--fs-border-color);
color: var(--fs-text-tertiary);
cursor: help;
}
.coverage-when {
font-size: 0.72rem;
color: var(--fs-text-tertiary);
}
.coverage-refresh { margin-left: auto; }
.coverage-numbers {
display: flex;
align-items: baseline;
gap: 0.4rem;
}
.coverage-count { font-size: 1.15rem; font-weight: 500; }
.coverage-label { font-size: 0.82rem; color: var(--fs-text-secondary); }
.coverage-bar {
height: 6px;
border-radius: 3px;
background: color-mix(in srgb, var(--fs-text-tertiary) 14%, transparent);
overflow: hidden;
}
.coverage-bar-fill {
height: 100%;
border-radius: 3px;
background: var(--fs-accent);
transition: width 0.3s ease;
}
.coverage-gaps {
display: flex;
align-items: center;
gap: 0.4rem;
flex-wrap: wrap;
font-size: 0.78rem;
}
.coverage-gaps-label { color: var(--fs-text-tertiary); }
.coverage-gap-chip {
padding: 0.1rem 0.5rem;
border-radius: var(--fs-radius-lg);
border: 1px solid var(--fs-border-color);
color: var(--fs-text-secondary);
font-family: var(--fs-font-mono);
font-size: 0.74rem;
}
.coverage-gap-count { opacity: 0.65; }
.coverage-empty {
margin: 0;
font-size: 0.8rem;
color: var(--fs-text-tertiary);
}
.coverage-error {
margin: 0;
font-size: 0.8rem;
color: var(--fs-error);
}
/* ── Two-column body ─────────────────────────────────────────── */
/* `minmax(0, 1fr)`, not `1fr`. A bare `1fr` track has an AUTO minimum, so it
cannot shrink below its content — one wide descendant anywhere in the
+125
View File
@@ -420,6 +420,16 @@ const baseUrl = ref("");
const savingBaseUrl = ref(false);
const baseUrlSaved = ref(false);
// Git forge integration (admin only, #2689). The token round-trips masked;
// the server treats the mask as "unchanged".
const forge = ref({ kind: "", base_url: "", token: "", webhook_secret: "" });
const forgeKinds = ref<string[]>(["gitea"]);
const forgeConfigured = ref(false);
const savingForge = ref(false);
const forgeSaved = ref(false);
const testingForge = ref(false);
const forgeTestResult = ref<{ ok: boolean; message: string } | null>(null);
// Search test (SearXNG)
const searxngConfigured = ref(false);
@@ -565,10 +575,28 @@ onMounted(async () => {
} catch {
// base URL not configured yet
}
try {
await loadForgeSettings();
} catch {
// forge not configured yet
}
}
_loadTabContent(activeTab.value);
});
async function loadForgeSettings() {
const cfg = await apiGet<{
kind: string; base_url: string; token: string; webhook_secret: string;
configured: boolean; kinds: string[];
}>("/api/admin/forge");
forge.value = {
kind: cfg.kind, base_url: cfg.base_url, token: cfg.token,
webhook_secret: cfg.webhook_secret,
};
forgeConfigured.value = cfg.configured;
if (cfg.kinds?.length) forgeKinds.value = cfg.kinds;
}
async function changeEmail() {
changingEmail.value = true;
try {
@@ -734,6 +762,45 @@ async function sendTestEmail() {
}
}
async function saveForge() {
savingForge.value = true;
forgeSaved.value = false;
forgeTestResult.value = null;
try {
await apiPut("/api/admin/forge", forge.value);
await loadForgeSettings();
forgeSaved.value = true;
setTimeout(() => (forgeSaved.value = false), 2000);
} catch (e) {
const body = (e as { body?: { error?: string } }).body;
toastStore.show(body?.error || "Failed to save forge settings", "error");
} finally {
savingForge.value = false;
}
}
async function testForge() {
testingForge.value = true;
forgeTestResult.value = null;
try {
const res = await apiPost<{ version: string; username: string }>(
"/api/admin/forge/test", {},
);
forgeTestResult.value = {
ok: true,
message: `Connected — Gitea ${res.version}, authenticated as ${res.username}`,
};
} catch (e) {
const body = (e as { body?: { error?: string } }).body;
forgeTestResult.value = {
ok: false,
message: body?.error || "Connection test failed",
};
} finally {
testingForge.value = false;
}
}
async function saveBaseUrl() {
savingBaseUrl.value = true;
baseUrlSaved.value = false;
@@ -2090,6 +2157,64 @@ function formatUserDate(iso: string): string {
</div>
</section>
<section class="settings-section full-width">
<h2>Git Forge</h2>
<p class="section-desc">
Optional read-only connection to your git forge (Gitea or GitHub) so
snippet code can be fetched, drift-checked, and coverage-measured
server-side. A read-scope token is enough. Leave the kind unset to
keep the integration off.
</p>
<div class="smtp-grid">
<div class="field">
<label for="forge-kind">Forge</label>
<select id="forge-kind" v-model="forge.kind" class="input">
<option value="">Off</option>
<option v-for="k in forgeKinds" :key="k" :value="k">{{ k }}</option>
</select>
</div>
<div class="field">
<label for="forge-base-url">Base URL</label>
<input id="forge-base-url" v-model="forge.base_url" type="text" placeholder="https://git.example.com" class="input" />
</div>
<div class="field">
<label for="forge-token">API Token (read scope)</label>
<input id="forge-token" v-model="forge.token" type="password" class="input" />
<p class="field-hint">
Gitea: an access token with read scope on repositories. GitHub:
a fine-grained PAT with Contents: Read-only (or a classic token
with repo read). For GitHub, use
<code>https://github.com</code> as the base URL — or your
GitHub Enterprise instance's URL.
</p>
</div>
<div class="field">
<label for="forge-webhook-secret">Webhook Secret</label>
<input id="forge-webhook-secret" v-model="forge.webhook_secret" type="password" class="input" />
<p class="field-hint">
Optional: create a push webhook on the forge pointing at
<code>/api/webhooks/forge</code> with this secret, and snippets
whose recorded files change get flagged for re-verification.
</p>
</div>
</div>
<div class="actions" style="margin-bottom: 1.25rem;">
<button class="btn-primary" @click="saveForge" :disabled="savingForge">
{{ savingForge ? "Saving..." : "Save Forge Settings" }}
</button>
<button class="btn-ghost" @click="testForge" :disabled="testingForge || !forgeConfigured">
{{ testingForge ? "Testing..." : "Test Connection" }}
</button>
<span v-if="forgeSaved" class="saved-msg">Saved!</span>
</div>
<p
v-if="forgeTestResult"
:class="forgeTestResult.ok ? 'text-success' : 'text-error'"
>
{{ forgeTestResult.message }}
</p>
</section>
</div>
<!-- ── Users ── -->
+1 -1
View File
@@ -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, 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.29",
"version": "0.1.30",
"author": { "name": "Bryan Van Deusen" },
"mcpServers": {
"scribe": {
+43 -17
View File
@@ -79,34 +79,60 @@ fi
# Definition-shaped patterns only. Grepping for bare occurrences would match
# every CALL site and drown the real finding — and a hint that is mostly noise
# is one people learn to skip, which is worse than none.
#
# ALL code, not a language shortlist (#2682): the detector was born covering
# only the languages of the repo it was written in, which silently amputated
# this whole arm — and the record nudge gated on it — for every Go/Kotlin/Rust
# project. Definitions are announced by a small keyword family across
# languages (func/fun/fn/function/def/sub · class/struct/trait/interface/
# enum/object/protocol/type), so one modifier-strip + keyword match covers
# them all. Known out of scope: keyword-less declaration syntax (C/Java/Dart
# `ReturnType name(...)`) needs a real parser, and `impl` blocks are excluded
# because several per type is normal Rust, not duplication.
# ---------------------------------------------------------------------------
local_lines=""
if [ -n "$repo_root" ] && [ -n "$code" ]; then
# kind<TAB>name for each thing this payload DEFINES.
names=$(printf '%s' "$code" | awk '
match($0, /^[[:space:]]*\.[A-Za-z][A-Za-z0-9_-]*[[:space:]]*[,{]/) {
t = $0; sub(/^[[:space:]]*\./, "", t); sub(/[[:space:]]*[,{].*$/, "", t);
if (t != "") print "css\t" t; next }
match($0, /^[[:space:]]*(export[[:space:]]+)?(default[[:space:]]+)?(async[[:space:]]+)?function[[:space:]]+[A-Za-z_$][A-Za-z0-9_$]*/) {
t = $0; sub(/^.*function[[:space:]]+/, "", t); sub(/[^A-Za-z0-9_$].*$/, "", t);
if (t != "") print "sym\t" t; next }
match($0, /^[[:space:]]*(export[[:space:]]+)?class[[:space:]]+[A-Za-z_$][A-Za-z0-9_$]*/) {
t = $0; sub(/^.*class[[:space:]]+/, "", t); sub(/[^A-Za-z0-9_$].*$/, "", t);
if (t != "") print "sym\t" t; next }
match($0, /^[[:space:]]*(async[[:space:]]+)?def[[:space:]]+[A-Za-z_][A-Za-z0-9_]*/) {
t = $0; sub(/^.*def[[:space:]]+/, "", t); sub(/[^A-Za-z0-9_].*$/, "", t);
if (t != "") print "sym\t" t; next }
match($0, /^[[:space:]]*(export[[:space:]]+)?(const|let)[[:space:]]+[A-Za-z_$][A-Za-z0-9_$]*[[:space:]]*=[[:space:]]*(async[[:space:]]*)?[(<]/) {
t = $0; sub(/^[[:space:]]*(export[[:space:]]+)?(const|let)[[:space:]]+/, "", t);
sub(/[^A-Za-z0-9_$].*$/, "", t);
if (t != "") print "sym\t" t; next }
{
# CSS class definition: .name { or .name,
if (match($0, /^[[:space:]]*\.[A-Za-z][A-Za-z0-9_-]*[[:space:]]*[,{]/)) {
t = $0; sub(/^[[:space:]]*\./, "", t); sub(/[[:space:]]*[,{].*$/, "", t)
if (t != "") print "css\t" t; next
}
line = $0; sub(/^[[:space:]]+/, "", line)
# Strip leading declaration modifiers so the definition keyword is the
# first word regardless of language (export/pub/private/suspend/...).
sub(/^((pub(\([a-z]+\))?|export|default|private|internal|protected|public|static|suspend|async|open|sealed|data|abstract|final|inline|unsafe|extern|override)[[:space:]]+)*/, "", line)
# Go method with receiver: func (r *T) Name(
if (match(line, /^func[[:space:]]*\([^)]*\)[[:space:]]*[A-Za-z_]/)) {
t = line; sub(/^func[[:space:]]*\([^)]*\)[[:space:]]*/, "", t)
sub(/[^A-Za-z0-9_].*$/, "", t)
if (t != "") print "sym\t" t; next
}
# Keyword-announced definitions, functions and named types alike.
# Dunders are skipped: every class defines __init__, so "already defined
# in N other files" is guaranteed noise for them — and noise is what
# teaches sessions to skip the hint.
if (match(line, /^(function|def|class|func|fun|fn|sub|struct|trait|interface|enum|object|protocol|type)[[:space:]]+[A-Za-z_$]/)) {
t = line; sub(/^[a-z]+[[:space:]]+/, "", t)
sub(/[^A-Za-z0-9_$].*$/, "", t)
if (t != "" && t !~ /^__.*__$/) print "sym\t" t; next
}
# Arrow/expression assignment: const name = (…) / let name = async (
if (match(line, /^(const|let)[[:space:]]+[A-Za-z_$][A-Za-z0-9_$]*[[:space:]]*=[[:space:]]*(async[[:space:]]*)?[(<]/)) {
t = line; sub(/^(const|let)[[:space:]]+/, "", t)
sub(/[^A-Za-z0-9_$].*$/, "", t)
if (t != "") print "sym\t" t; next
}
}
' 2>/dev/null | sort -u | head -12) || names=""
while IFS=$'\t' read -r kind name; do
[ -n "${name:-}" ] || continue
case "$kind" in
css) pat="^[[:space:]]*\.${name}[[:space:]]*[,{]" ;;
*) pat="(function|class|def)[[:space:]]+${name}[^A-Za-z0-9_]|(const|let)[[:space:]]+${name}[[:space:]]*=" ;;
*) pat="(function|def|class|func|fun|fn|sub|struct|trait|interface|enum|object|protocol|type)[[:space:]]+${name}[^A-Za-z0-9_]|func[[:space:]]*\([^)]*\)[[:space:]]*${name}[[:space:]]*\(|(const|let)[[:space:]]+${name}[[:space:]]*=" ;;
esac
# -I skips binaries; :(exclude) drops the file being written, which would
# otherwise always match itself on an Edit.
+15 -10
View File
@@ -52,16 +52,21 @@ for the operator's work, and as your own working memory across sessions.
it. An untagged project record carries the `systems_hint` question instead,
on creates, updates, and work-logs alike — treat it as the tagging question
asked at the moment of work, not as noise to skip past.
- **Reuse before rebuilding — and record what you build** — 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.
The recording half has NAMED TRIGGERS, not a vibe: the moment you extract a
shared component, hoist a helper into a common module, or notice you are
writing the second copy of anything, record it with `create_snippet` (name,
code, when-to-reach-for-it, location) in the same breath as the commit.
Work that "refactors X into a shared Y" is not finished until Y is recorded
— an unrecorded shared component is invisible to every later session, which
is how a codebase grows four `.btn-primary` definitions.
- **The pattern library: start from recorded shapes, and record every shape
at first build** — recorded **snippets** are the project's pattern library,
not a dedup net. Before building ANY shape — a button, an input field, a
modal, a route handler, a service class, a test scaffold, up through complex
subsystem patterns — search snippets and START from the recorded shape; a
deliberate departure is recorded as its own named variant, never left as
silent drift. And the FIRST time a shape is built, record it with
`create_snippet` (name, when-to-reach-for-it, location, code) in the same
breath — do not judge whether it "might recur": the builder of the first
instance can never know, and a missed record is invisible until it
resurfaces as an uninformed duplicate. A mature project's snippet corpus
should read as a map of every shape in it. The backstop still holds:
noticing the second copy of anything, or consolidating copies into a shared
X, means X gets recorded before that work is finished — which is how a
codebase is kept from growing four `.btn-primary` definitions.
- 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
+26 -15
View File
@@ -1,20 +1,24 @@
---
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.
description: Use when you're about to build ANY shape — a component, control, route handler, service class, helper, test scaffold — search recorded snippets FIRST and start from the recorded shape instead of re-solving it. And the FIRST time a shape is built, record it as a snippet so every later instance starts from it. Triggers on "write a util/helper", "I need a function that…", "let me add a component/button/field/route", or having just built the first instance of anything.
---
# Reusing code — recall before you rebuild
# Reusing code — the pattern library
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 the project's **pattern library**, not a dedup net. Each records a
named shape — with its language, signature, canonical location (repo · path ·
symbol), a one-line *"when to reach for it,"* and the code — so every later
instance STARTS from the recorded shape: buttons start from the button shape,
fields from the field shape, and "special" is a deliberate, named exception
rather than drift. A mature project's snippet corpus reads as a map of every
shape in it, from the humblest control to the most complex subsystem pattern.
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
## Before you build any shape — search first
- About to write a utility, hook, formatter, adapter, or a reusable component?
- About to build a component, control, route handler, service class, utility,
hook, formatter, adapter, or test scaffold?
**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
@@ -38,10 +42,15 @@ through recall/auto-inject; this skill is the active reflex around that.
duplicate — reuse it and drop yours — or it isn't, and the record needs the new
location adding. Both are cheaper now than after the duplicate settles in.
## The moment you build something reusable — record it
## The first time a shape is built — record it
- Just wrote (or noticed) a helper, hook, pattern, or component worth repeating?
Record it with `create_snippet` while it's fresh:
- Just built the FIRST instance of anything with a shape — a component, a
field, a route, a service pattern, a scaffold? Record it with
`create_snippet` while it's fresh. Do **not** stop to judge whether it will
recur: the builder of the first instance can never know, and a missed record
is invisible until it resurfaces as an uninformed duplicate. Over-recording
is safe — dead weight shows up in the usage counters and can be pruned;
under-recording has no signal at all. The record is cheap — these fields:
- **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
@@ -92,7 +101,9 @@ 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.
A one-off written a second time is the cost this avoids — and at project
scale, the cost is an application whose buttons, fields, and services each
exist in four diverging shapes. Recording every shape once — with a location
and a crisp "when to use" — means every later session starts from the pattern
library instead of re-deriving it. Search before building; record every shape
at first build.
+2
View File
@@ -31,6 +31,7 @@ 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.routes.webhooks import webhooks_bp
from scribe.mcp import mount_mcp
STATIC_DIR = Path(__file__).parent / "static"
@@ -95,6 +96,7 @@ def create_app() -> Quart:
app.register_blueprint(dashboard_bp)
app.register_blueprint(systems_bp)
app.register_blueprint(snippets_bp)
app.register_blueprint(webhooks_bp)
@app.before_request
async def before_request():
+12
View File
@@ -60,6 +60,18 @@ class Config:
# the MCP layer doesn't proxy web search (Claude has its own).
SEARXNG_URL: str = os.environ.get("SEARXNG_URL", "")
# Git forge integration (#2689) — optional read access to the operator's
# forge so snippet bodies can be fetched/verified server-side. Normally
# configured in Settings → Config (stored as admin settings); these env
# fallbacks exist so a deployment can keep the token in a Docker secret
# instead of the database. DB value wins when both are set.
FORGE_KIND: str = os.environ.get("FORGE_KIND", "")
FORGE_BASE_URL: str = os.environ.get("FORGE_BASE_URL", "").rstrip("/")
FORGE_TOKEN: str = _read_secret("FORGE_TOKEN", "FORGE_TOKEN_FILE", "")
FORGE_WEBHOOK_SECRET: str = _read_secret(
"FORGE_WEBHOOK_SECRET", "FORGE_WEBHOOK_SECRET_FILE", ""
)
@classmethod
def oidc_enabled(cls) -> bool:
return bool(cls.OIDC_ISSUER and cls.OIDC_CLIENT_ID and cls.OIDC_CLIENT_SECRET)
+17 -1
View File
@@ -17,6 +17,7 @@ keeps working.
from __future__ import annotations
from scribe.mcp._context import current_user_id
from scribe.services import coverage as coverage_svc
from scribe.services import design_systems as design_systems_svc
from scribe.services import milestones as milestones_svc
from scribe.services import notes as notes_svc
@@ -56,7 +57,13 @@ async def enter_project(project_id: int) -> dict:
Returns a dict with keys: project, milestone_summary, applicable_rules,
project_rules, subscribed_rulebooks, applicable_rules_truncated,
open_tasks, recent_notes, design_system, systems.
open_tasks, recent_notes, design_system, systems, pattern_coverage.
`pattern_coverage` (usually null) is a one-line estimate of how much of
the bound repo's code has recorded snippets — e.g. "pattern-library
coverage: 34/210 shapes recorded (estimate); largest gaps: internal/api".
When present, treat the gaps as a standing invitation: as you touch code
in those areas, record the shapes you find with create_snippet.
`systems` is the project's vocabulary of named subsystems/areas. It is
returned here so you can TAG as you write: when creating or meaningfully
@@ -128,8 +135,17 @@ async def enter_project(project_id: int) -> dict:
uid, project.design_system_id,
)
# Cache read ONLY — computing coverage moves a repo tarball and never
# belongs in this request path. Null is the ordinary state (no forge, or
# never computed); the line appears exactly when there is evidence. Read
# on the OWNER's id: bindings and the cache live with the project owner.
coverage = await coverage_svc.cached_coverage(
project.user_id or uid, project_id
)
return {
"project": project.to_dict(),
"pattern_coverage": coverage_svc.coverage_line(coverage) if coverage else None,
# Trimmed to what tagging needs. The full charter is get_system's job —
# this list rides along on every session start, so it stays lean.
"systems": [
+46 -7
View File
@@ -23,7 +23,12 @@ async def list_snippets(
q: str = "", tag: str = "", limit: int = 50, project_id: int = 0,
repo: str = "", path: str = "", symbol: str = "", verification: str = "",
) -> dict:
"""List recorded snippets (reusable functions/components).
"""List recorded snippets — the project's pattern library.
Search here BEFORE building any shape (a component, control, route
handler, service class, helper, scaffold): a recorded shape is the
starting point for every later instance, and building without checking is
how the same button ends up defined four diverging ways.
Two ways to ask, usable together: by MEANING (`q` — "what do I need this code
to do?") and by PLACE (`repo`/`path`/`symbol` — "what canonical helpers
@@ -100,14 +105,20 @@ async def create_snippet(
project_id: int = 0,
system_ids: list[int] | None = None,
force: bool = False,
commit_sha: str = "",
) -> dict:
"""Record a reusable function/component so future sessions can RECALL it
instead of writing a fresh one-off.
"""Record a shape in the project's pattern library, so every later
instance starts from it instead of re-deriving it.
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.
Reach for this the FIRST time any shape is built — a component, a control,
a route handler, a service class, a helper, a test scaffold — not only
when something is judged "reusable": the builder of the first instance
can't know what will recur, and a missed record is invisible until it
resurfaces as an uninformed duplicate. Over-recording is safe (dead weight
shows in the usage counters and can be pruned); under-recording has no
signal. A deliberate departure from a recorded shape is recorded as its
own named variant, not left as drift. Before building, search first — the
shape may already be recorded.
Args:
name: Short name of the function/component, e.g. "useDebouncedRef".
@@ -126,6 +137,12 @@ async def create_snippet(
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).
commit_sha: The commit the code was read at (`git rev-parse HEAD` — you
have the repo, so it's free). The recorded location is the source
of truth for the code and the stored body is a cache of it; this
stamps what the cache is a cache OF, so staleness is judgeable
later. Optional, but pass it whenever you're recording from a
checkout.
Returns the created snippet (including a parsed `snippet` field), OR — when a
duplicate already exists and force is false — {"duplicate": true,
@@ -171,6 +188,7 @@ async def 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,
commit_sha=commit_sha,
)
if system_ids:
await systems_svc.set_record_systems(uid, note.id, system_ids)
@@ -183,6 +201,12 @@ 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.
On an instance with a forge configured, the response also carries
`body_source` + `body_freshness`: "current" means the code was just
confirmed against the recorded location; "diverged" or "missing" means
the source moved on — trust the location over the cached body and
consider verify_snippet after you look.
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
@@ -193,6 +217,9 @@ async def get_snippet(snippet_id: int) -> dict:
if note is None:
raise ValueError(f"snippet {snippet_id} not found")
data = snippets_svc.snippet_to_dict(note)
# Forge-checked freshness (#2690): attaches body_source/body_freshness
# when the instance has a forge; a no-forge instance sees no new fields.
await snippets_svc.attach_live_body(note, data)
data.update(await access_svc.describe_provenance(uid, note))
# A "pull" is an explicit open, so it's recorded HERE rather than in
# snippets_svc.get_snippet — the service is also reached by update/merge
@@ -285,6 +312,7 @@ async def find_duplicate_snippets(threshold: float = 0.0) -> dict:
async def verify_snippet(
snippet_id: int, status: str, detail: str = "", path: str = "",
commit_sha: str = "",
) -> dict:
"""Record whether a snippet's recorded location and code still match source.
@@ -319,6 +347,10 @@ async def verify_snippet(
path: The path you actually checked, if it differs from the recorded
one (e.g. you found the symbol at its new home). Defaults to the
recorded path.
commit_sha: The commit the working tree was at when you checked
(`git rev-parse HEAD`). An "ok" verdict with it also refreshes the
body's provenance — you just proved the cached code matches the
source at that commit.
Requires write access: a verdict changes how the record is presented, so
being able to read a snippet someone shared with you doesn't let you mark
@@ -327,6 +359,7 @@ async def verify_snippet(
uid = current_user_id()
note = await snippets_svc.record_verification(
uid, snippet_id, status=status, detail=detail, path=path,
commit_sha=commit_sha,
)
if note is None:
raise ValueError(
@@ -349,6 +382,7 @@ async def update_snippet(
tags: list[str] | None = None,
project_id: int = 0,
system_ids: list[int] | None = None,
commit_sha: str = "",
) -> dict:
"""Update a snippet. Only the fields you pass change.
@@ -364,6 +398,10 @@ async def update_snippet(
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.
commit_sha: When you're updating the code from a checkout, the commit
it was read at (`git rev-parse HEAD`). Restamps the body's
provenance; changing the code WITHOUT it drops the old stamp,
since the new body no longer comes from that commit.
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
@@ -384,6 +422,7 @@ async def update_snippet(
signature=signature, when_to_use=when_to_use,
repo=repo, path=path, symbol=symbol,
locations=locations, tags=tags, project_id=project,
commit_sha=commit_sha or None,
)
except PermissionError as exc:
# Readable but not writable — surface the real reason, not "not found".
+91
View File
@@ -19,6 +19,15 @@ from scribe.services.backup import (
restore_full_backup,
)
from scribe.services.email import SMTP_SETTING_KEYS, get_base_url, get_smtp_config, is_smtp_configured, send_test_email
from scribe.services.forge import (
FORGE_BASE_URL_KEY,
FORGE_KIND_KEY,
FORGE_KINDS,
FORGE_TOKEN_KEY,
ForgeError,
forge_config,
get_forge,
)
from scribe.services.logging import get_logs, get_log_stats, log_audit
from scribe.services.notifications import send_invitation_email
from scribe.services.settings import (
@@ -157,6 +166,88 @@ async def test_smtp():
return jsonify({"error": str(e)}), 500
_TOKEN_MASK = "********"
@admin_bp.route("/forge", methods=["GET"])
@admin_required
async def get_forge_settings():
from scribe.config import Config
from scribe.routes.webhooks import FORGE_WEBHOOK_SECRET_KEY
cfg = await forge_config()
webhook_secret = (
await get_admin_setting(FORGE_WEBHOOK_SECRET_KEY, "")
or Config.FORGE_WEBHOOK_SECRET
)
return jsonify({
"kind": cfg["kind"],
"base_url": cfg["base_url"],
# Secrets never leave the server — the smtp_password convention:
# masked when set, empty when not.
"token": _TOKEN_MASK if cfg["token"] else "",
"webhook_secret": _TOKEN_MASK if webhook_secret else "",
"configured": bool(await get_forge()),
"kinds": list(FORGE_KINDS),
})
@admin_bp.route("/forge", methods=["PUT"])
@admin_required
async def update_forge_settings():
data = await request.get_json() or {}
uid = get_current_user_id()
kind = str(data.get("kind", "")).strip().lower()
if kind and kind not in FORGE_KINDS:
return jsonify({"error": f"Unknown forge kind {kind!r}"}), 400
base_url = str(data.get("base_url", "")).strip().rstrip("/")
if base_url and not base_url.startswith(("http://", "https://")):
return jsonify({"error": "Forge base URL must use http or https"}), 400
await set_admin_setting(FORGE_KIND_KEY, kind)
await set_admin_setting(FORGE_BASE_URL_KEY, base_url)
token = data.get("token")
# The mask coming back means "unchanged" — the form round-trips what GET
# showed it, and storing the mask would silently break the integration.
if token is not None and token != _TOKEN_MASK:
await set_admin_setting(FORGE_TOKEN_KEY, str(token))
from scribe.routes.webhooks import FORGE_WEBHOOK_SECRET_KEY
webhook_secret = data.get("webhook_secret")
if webhook_secret is not None and webhook_secret != _TOKEN_MASK:
await set_admin_setting(FORGE_WEBHOOK_SECRET_KEY, str(webhook_secret))
# The token is deliberately absent from the audit detail.
await log_audit(
"forge_config", user_id=uid, username=g.user.username,
ip_address=request.remote_addr,
details={"kind": kind, "base_url": base_url},
)
return jsonify({"status": "ok"})
@admin_bp.route("/forge/test", methods=["POST"])
@admin_required
async def test_forge():
"""Probe the SAVED forge config: reachability and token acceptance in one
press, so a misconfiguration is visible now rather than as silent
fallbacks later (#2663's lesson, applied to integrations)."""
uid = get_current_user_id()
forge = await get_forge()
if forge is None:
return jsonify({"error": "Forge is not configured — save kind, base URL and token first"}), 400
try:
result = await forge.check()
except ForgeError as e:
return jsonify({"error": str(e)}), 502
await log_audit(
"forge_test", user_id=uid, username=g.user.username,
ip_address=request.remote_addr,
details={"ok": True, "username": result.get("username", "")},
)
return jsonify(result)
@admin_bp.route("/logs", methods=["GET"])
@admin_required
async def list_logs():
+55
View File
@@ -120,6 +120,61 @@ async def delete_project_route(project_id: int):
return "", 204
@projects_bp.route("/<int:project_id>/coverage", methods=["GET"])
@login_required
async def get_coverage_route(project_id: int):
"""The cached pattern-library coverage summary — never computes.
`configured` tells the card whether offering a Refresh button makes
sense; `coverage` is null until something has computed it (a webhook
push or an explicit refresh).
"""
from scribe.services.coverage import cached_coverage
from scribe.services.forge import get_forge
uid = get_current_user_id()
result = await get_project_for_user(uid, project_id)
if result is None:
return not_found("Project")
project, _ = result
owner_uid = project.user_id or uid
return jsonify({
"configured": await get_forge() is not None,
"coverage": await cached_coverage(owner_uid, project_id),
})
@projects_bp.route("/<int:project_id>/coverage/refresh", methods=["POST"])
@login_required
async def refresh_coverage_route(project_id: int):
"""Recompute coverage now (archive fetch — seconds, not milliseconds).
Synchronous on purpose: the caller is a person who just clicked Refresh
and wants the new number, and the forge timeout bounds the wait.
"""
from scribe.services.coverage import refresh_coverage
from scribe.services.forge import ForgeError, get_forge
uid = get_current_user_id()
result = await get_project_for_user(uid, project_id)
if result is None:
return not_found("Project")
project, _ = result
owner_uid = project.user_id or uid
if await get_forge() is None:
return jsonify({"error": "No git forge is configured (Settings → Config → Git Forge)"}), 400
try:
coverage = await refresh_coverage(owner_uid, project_id)
except ForgeError as exc:
return jsonify({"error": str(exc)}), 502
if coverage is None:
return jsonify({
"error": "No bound repo is served by the configured forge — "
"bind the project's repo (bind_repo) on a remote the forge hosts"
}), 400
return jsonify({"coverage": coverage})
@projects_bp.route("/<int:project_id>/notes", methods=["GET"])
@login_required
async def get_project_notes_route(project_id: int):
+20 -2
View File
@@ -16,13 +16,27 @@ logger = logging.getLogger(__name__)
settings_bp = Blueprint("settings", __name__, url_prefix="/api/settings")
# Keys whose values are credentials. The admin endpoints that own them mask on
# read and skip the mask on write; this generic KV surface has to apply the
# same treatment, or it silently un-masks what those endpoints masked — the
# rows live on the admin's own user_id, so the plain GET returned them raw.
_SECRET_KEYS = frozenset({"smtp_password", "forge_token", "forge_webhook_secret"})
_SECRET_MASK = "********"
def _masked(settings: dict) -> dict:
return {
k: (_SECRET_MASK if k in _SECRET_KEYS and v else v)
for k, v in settings.items()
}
@settings_bp.route("", methods=["GET"])
@login_required
async def get_settings_route():
uid = get_current_user_id()
settings = await get_all_settings(uid)
return jsonify(settings)
return jsonify(_masked(settings))
@settings_bp.route("", methods=["PUT"])
@@ -36,6 +50,10 @@ async def update_settings_route():
to_save = {}
for k, v in data.items():
str_v = str(v)
# A masked secret round-tripping through a client is "unchanged", not
# a request to store the mask over the real credential.
if k in _SECRET_KEYS and str_v == _SECRET_MASK:
continue
if not str_v:
await delete_setting(uid, k)
else:
@@ -45,7 +63,7 @@ async def update_settings_route():
await set_settings_batch(uid, to_save)
settings = await get_all_settings(uid)
return jsonify(settings)
return jsonify(_masked(settings))
@settings_bp.route("/search", methods=["GET"])
+2
View File
@@ -157,6 +157,8 @@ async def get_snippet_route(snippet_id: int):
return not_found("Snippet")
note, permission = loaded
data = snippets_svc.snippet_to_dict(note)
# Forge-checked freshness (#2690) — same decoration the MCP pull gets.
await snippets_svc.attach_live_body(note, data)
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
+132
View File
@@ -0,0 +1,132 @@
"""Forge push webhook — drift flagging at the moment the repo moves (#2691).
The forge POSTs here on every push; changed paths are matched against recorded
snippet locations (through repo bindings) and matched verdicts get an
``invalidated_by`` marker that surfaces in the ``verification="attention"``
listing. This is what makes verification scale past tens of records: sessions
recheck what pushes flagged instead of sweeping everything.
Registering the webhook on the forge is per-instance setup (Settings → Config
→ Git Forge shows the endpoint and holds the secret) — the server never
self-registers on the forge.
Contract with the forge's delivery loop:
- No secret configured → 404: the endpoint doesn't exist until an operator
creates it. Bad signature → 401: that's a caller problem worth signaling.
- A PROCESSING failure returns 200 with ``{"ok": false}`` and drops a
WARNING + AppLog row (the #2663 canary pattern): repeated 5xx responses
make forges mark deliveries failed and operators disable the hook, which
would silently turn the feature off — the exact failure mode this
milestone exists to prevent.
"""
from __future__ import annotations
import hashlib
import hmac
import logging
import traceback
from quart import Blueprint, jsonify, request
from scribe.config import Config
from scribe.services.background import spawn
from scribe.services.coverage import refresh_coverage
from scribe.services.repo_bindings import bindings_for_key, normalize_repo_key
from scribe.services.settings import get_admin_setting
from scribe.services.snippets import invalidate_for_push
logger = logging.getLogger(__name__)
webhooks_bp = Blueprint("webhooks", __name__, url_prefix="/api/webhooks")
FORGE_WEBHOOK_SECRET_KEY = "forge_webhook_secret"
def signature_ok(secret: str, body: bytes, signature: str) -> bool:
"""Validate a push signature: the hex HMAC-SHA256 of the raw body under
the webhook secret (Gitea's X-Gitea-Signature verbatim; GitHub's
X-Hub-Signature-256 minus its "sha256=" prefix). Constant-time compare."""
if not secret or not signature:
return False
expected = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, signature.strip().lower())
def delivered_signature(headers) -> str:
"""The HMAC hex a delivery carries, whichever forge sent it: Gitea's
X-Gitea-Signature verbatim, or GitHub's X-Hub-Signature-256 minus its
"sha256=" scheme prefix (#2693). Empty when neither header is present."""
return headers.get("X-Gitea-Signature", "") or headers.get(
"X-Hub-Signature-256", ""
).removeprefix("sha256=")
def push_facts(payload: dict) -> tuple[str, list[str], list[str], str]:
"""(repo identity, changed paths, removed paths, head commit) from a Gitea
push payload. Tolerant: absent fields read as empty, never raise."""
repo = payload.get("repository") or {}
raw_repo = repo.get("clone_url") or repo.get("html_url") or repo.get("full_name") or ""
changed: list[str] = []
removed: list[str] = []
for commit in payload.get("commits") or []:
changed.extend(commit.get("added") or [])
changed.extend(commit.get("modified") or [])
removed.extend(commit.get("removed") or [])
# De-dup while keeping order stable for logs.
changed = list(dict.fromkeys(changed))
removed = list(dict.fromkeys(removed))
return raw_repo, changed, removed, str(payload.get("after") or "")
@webhooks_bp.route("/forge", methods=["POST"])
async def forge_push():
secret = await get_admin_setting(FORGE_WEBHOOK_SECRET_KEY, "") or Config.FORGE_WEBHOOK_SECRET
if not secret:
# Not "forbidden" — the endpoint is not a thing on this instance.
return jsonify({"error": "Not found"}), 404
body = await request.get_data()
# The payload shape push_facts reads (repository.clone_url,
# commits[].added/modified/removed, after) is common to both forges, so
# the signature header is the whole GitHub mapping.
if not signature_ok(secret, body, delivered_signature(request.headers)):
return jsonify({"error": "Invalid signature"}), 401
try:
payload = await request.get_json(force=True) or {}
raw_repo, changed, removed, head = push_facts(payload)
repo_key = normalize_repo_key(raw_repo)
if not repo_key:
return jsonify({"ok": True, "flagged": 0, "reason": "no repository in payload"})
flagged = await invalidate_for_push(repo_key, changed, removed, head)
if flagged:
logger.info(
"forge push %s flagged %d snippet(s) for recheck", head[:12], flagged
)
# A push is exactly when the coverage number goes stale — recompute it
# off the delivery path (#2692). Fire-and-forget: the forge's delivery
# loop must not wait on an archive download, and a failure is a
# WARNING from spawn(), never a failed delivery. This also SEEDS the
# cache on a webhook-configured instance — no manual first refresh.
for binding in await bindings_for_key(repo_key):
spawn(
refresh_coverage(binding.user_id, binding.project_id),
site="webhooks.coverage_refresh",
)
return jsonify({"ok": True, "flagged": flagged})
except Exception:
logger.warning("forge webhook processing failed", exc_info=True)
try:
from scribe.services.logging import log_error
await log_error(
endpoint="webhooks/forge",
error_type="forge_webhook_failed",
error_message="push received but drift flagging failed — "
"snippets touched by this push were not marked for recheck",
traceback=traceback.format_exc(),
)
except Exception:
logger.debug("forge webhook canary write failed", exc_info=True)
# 200 on purpose — see the module docstring's delivery contract.
return jsonify({"ok": False})
+51
View File
@@ -0,0 +1,51 @@
"""Fire-and-forget background tasks that actually run.
The event loop holds only a WEAK reference to a task, so a bare
``create_task`` with no other holder can be garbage-collected mid-flight — a
write that never errors and never lands (the #2663 GC footgun). This module is
the one place that gets the pattern right: strong references in ``_pending``,
discarded on completion, with failures logged at WARNING instead of vanishing.
``note_usage`` and ``retrieval_telemetry`` predate this module and carry their
own copies with bespoke canary semantics; new fire-and-forget callers use this
instead of writing a fourth copy.
"""
from __future__ import annotations
import asyncio
import logging
from collections.abc import Coroutine
logger = logging.getLogger(__name__)
_pending: set[asyncio.Task] = set()
def spawn(coro: Coroutine, *, site: str) -> None:
"""Schedule ``coro`` fire-and-forget; ``site`` names it in failure logs.
No running loop (sync context outside the app) closes the coroutine and
skips — every app path runs on the loop, and blocking would be worse.
"""
try:
task = asyncio.get_running_loop().create_task(coro)
except RuntimeError:
coro.close()
logger.debug("background task %s skipped — no running event loop", site)
return
_pending.add(task)
def _done(t: asyncio.Task) -> None:
_pending.discard(t)
if not t.cancelled() and t.exception() is not None:
logger.warning(
"background task %s failed", site, exc_info=t.exception()
)
task.add_done_callback(_done)
async def drain() -> None:
"""Await everything in flight — for tests that need the writes landed."""
while _pending:
await asyncio.gather(*list(_pending), return_exceptions=True)
+336
View File
@@ -0,0 +1,336 @@
"""Pattern-library coverage — what fraction of a bound repo's shapes have a
recorded snippet (#2692, forge job 3 of decision #2686).
The all-shapes doctrine says every shape gets recorded at first build. This
module is the hoping→knowing move: it enumerates the definitions that exist in
a project's bound repos (via the forge, one archive download per repo) and
compares them against recorded snippet locations, so "record everything"
becomes a watched number instead of an aspiration.
The definition extractor MIRRORS the write-path hook's awk rules
(plugin/hooks/scribe_prior_art.sh, ARM 1) — one shared notion of "a
definition" between the hook and the server, so the metric and the backstop
agree on what counts. The two are pinned together by shared test vectors in
tests/test_pattern_coverage.py; change one, change both.
The number is an ESTIMATE and every surface must say so: keyword extraction
over-counts (private one-offs, generated code that slips the dir filter) and
under-counts (keyword-less declaration syntax — C/Java/Dart — needs a real
parser and is out of scope, exactly as it is for the hook). The trend carries
the meaning, like the usage counters; the raw number is not a grade.
Compute is on demand + cached with a freshness stamp — recomputed on webhook
push and explicit refresh, NEVER in the request path of enter_project, which
only ever reads the cache.
"""
from __future__ import annotations
import io
import json
import logging
import posixpath
import re
import tarfile
from datetime import datetime, timezone
from scribe.services.forge import ForgeAdapter, get_forge
from scribe.services.repo_bindings import keys_for_project
from scribe.services.settings import get_setting, set_setting
logger = logging.getLogger(__name__)
# Cache key in the settings KV, on the project OWNER's user_id — the same
# channel the scheduler's last-run summary uses for machine-written state.
_CACHE_KEY_PREFIX = "pattern_coverage_"
# Files whose content can't hold definitions — the hook's skip list, verbatim,
# plus sourcemaps (which are JSON in a trenchcoat).
_SKIP_SUFFIXES = (
".md", ".mdx", ".txt", ".rst", ".json", ".lock", ".log", ".csv", ".tsv",
".svg", ".png", ".jpg", ".jpeg", ".gif", ".ico", ".pdf", ".map",
)
# Vendored/generated trees would swamp the metric with shapes nobody should
# record — the dunder-skip lesson at directory scale: guaranteed noise teaches
# people to ignore the number.
_SKIP_DIRS = frozenset({
"node_modules", "vendor", "dist", "build", "target",
"__pycache__", ".git", ".venv", "venv",
})
# A single source file bigger than this is almost certainly generated or
# vendored (bundles, lockstep protos) — skipped, and part of why the number
# is labeled an estimate.
_MAX_FILE_BYTES = 1_000_000
# --- the definition extractor (mirror of scribe_prior_art.sh ARM 1) ----------
_CSS_RE = re.compile(r"^\s*\.([A-Za-z][A-Za-z0-9_-]*)\s*[,{]")
# Leading declaration modifiers, so the definition keyword is the first word
# regardless of language (export/pub/private/suspend/...).
_MODIFIERS_RE = re.compile(
r"^(?:(?:pub(?:\([a-z]+\))?|export|default|private|internal|protected"
r"|public|static|suspend|async|open|sealed|data|abstract|final|inline"
r"|unsafe|extern|override)\s+)*"
)
# Go method with receiver: func (r *T) Name(
_GO_METHOD_RE = re.compile(r"^func\s*\([^)]*\)\s*([A-Za-z_][A-Za-z0-9_]*)")
# Keyword-announced definitions, functions and named types alike. `impl` is
# excluded on purpose — several per type is normal Rust, not duplication.
_KEYWORD_RE = re.compile(
r"^(?:function|def|class|func|fun|fn|sub|struct|trait|interface|enum"
r"|object|protocol|type)\s+([A-Za-z_$][A-Za-z0-9_$]*)"
)
# Arrow/expression assignment: const name = (…) / let name = async (
_ARROW_RE = re.compile(
r"^(?:const|let)\s+([A-Za-z_$][A-Za-z0-9_$]*)\s*=\s*(?:async\s*)?[(<]"
)
def extract_shapes(text: str) -> list[tuple[str, str]]:
"""Every (kind, name) this text DEFINES — kind is "css" or "sym".
Rule-for-rule mirror of the hook's awk program: first match wins per
line, dunders are skipped (every class defines __init__ — guaranteed
noise), duplicates within one text count once.
"""
seen: set[tuple[str, str]] = set()
out: list[tuple[str, str]] = []
for raw in text.splitlines():
m = _CSS_RE.match(raw)
if m:
shape = ("css", m.group(1))
else:
line = _MODIFIERS_RE.sub("", raw.lstrip())
if m := _GO_METHOD_RE.match(line):
shape = ("sym", m.group(1))
elif m := _KEYWORD_RE.match(line):
name = m.group(1)
if name.startswith("__") and name.endswith("__"):
continue
shape = ("sym", name)
elif m := _ARROW_RE.match(line):
shape = ("sym", m.group(1))
else:
continue
if shape not in seen:
seen.add(shape)
out.append(shape)
return out
def scannable(path: str) -> bool:
"""Should this repo file be scanned for shapes at all?"""
parts = path.split("/")
if any(p in _SKIP_DIRS for p in parts[:-1]):
return False
return not path.lower().endswith(_SKIP_SUFFIXES)
def shapes_from_archive(blob: bytes) -> list[tuple[str, str, str]]:
"""(path, kind, name) for every definition in a repo tarball.
Forge archives wrap content in a single top-level directory (repo-ref/);
that component is stripped so paths match recorded snippet locations,
which are repo-relative. Non-UTF-8 files are binaries and skipped.
"""
shapes: list[tuple[str, str, str]] = []
with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as tar:
for member in tar:
if not member.isfile() or "/" not in member.name:
continue
path = member.name.split("/", 1)[1]
if not path or not scannable(path) or member.size > _MAX_FILE_BYTES:
continue
handle = tar.extractfile(member)
if handle is None:
continue
try:
text = handle.read().decode("utf-8")
except UnicodeDecodeError:
continue
shapes.extend((path, kind, name) for kind, name in extract_shapes(text))
return shapes
# --- matching shapes against recorded locations ------------------------------
def _norm_symbol(kind_or_symbol: str) -> str:
# CSS shapes and recorded CSS symbols may or may not carry the leading
# dot; compare without it so ".btn-primary" and "btn-primary" agree.
return kind_or_symbol.lstrip(".").strip()
def _location_covers(loc_path: str, loc_symbol: str, path: str, name: str) -> bool:
if _norm_symbol(loc_symbol) != _norm_symbol(name):
return False
if not loc_path:
# Symbol-only record: the symbol match is all the claim there is.
return True
# The drift check's location semantics, not a second copy of them: exact
# file, or the recorded path is a directory the file lives under.
from scribe.services.snippets import _path_touches
return _path_touches(loc_path, path)
def match_shapes(
shapes: list[tuple[str, str, str]],
recorded: list[tuple[str, str]],
) -> list[tuple[str, str, str, bool]]:
"""Each shape with whether some recorded (path, symbol) location covers it.
Symbol-less recorded locations never cover a shape — a whole-file record
makes no claim about any particular definition inside it. The recorded
repo NAME is deliberately not consulted: it is free-form ("Scribe") and
the project binding already did the scoping; on a project binding several
repos this can over-credit a same-named symbol, which the estimate label
owns.
"""
usable = [(p, s) for p, s in recorded if (s or "").strip()]
return [
(
path,
kind,
name,
any(_location_covers(lp, ls, path, name) for lp, ls in usable),
)
for path, kind, name in shapes
]
def largest_gaps(
matched: list[tuple[str, str, str, bool]], *, top: int = 3
) -> list[dict]:
"""The directories with the most uncovered shapes — where a backlog
session should start, named the way the repo names them."""
by_dir: dict[str, dict[str, int]] = {}
for path, _kind, _name, covered in matched:
d = posixpath.dirname(path) or "(root)"
row = by_dir.setdefault(d, {"total": 0, "uncovered": 0})
row["total"] += 1
if not covered:
row["uncovered"] += 1
ranked = sorted(
by_dir.items(), key=lambda kv: (-kv[1]["uncovered"], kv[0])
)
return [
{"dir": d, "uncovered": row["uncovered"], "total": row["total"]}
for d, row in ranked[:top]
if row["uncovered"]
]
# --- compute, cache, surface -------------------------------------------------
async def _recorded_locations(user_id: int, project_id: int) -> list[tuple[str, str]]:
"""(path, symbol) for every location of every live snippet in a project."""
from sqlalchemy import select
from scribe.models import async_session
from scribe.models.note import Note
from scribe.services.snippets import SNIPPET_NOTE_TYPE, snippet_fields
async with async_session() as session:
rows = await session.execute(
select(Note).where(
Note.user_id == user_id,
Note.project_id == project_id,
Note.note_type == SNIPPET_NOTE_TYPE,
Note.deleted_at.is_(None),
)
)
notes = list(rows.scalars().all())
out: list[tuple[str, str]] = []
for note in notes:
for loc in snippet_fields(note).get("locations") or []:
out.append((loc.get("path") or "", loc.get("symbol") or ""))
return out
async def compute_coverage(
user_id: int, project_id: int, *, forge: ForgeAdapter | None = None
) -> dict | None:
"""Measure a project's pattern-library coverage against its bound repos.
None means "nothing to measure" — no forge configured, or none of the
project's bound repos is served by it. That is the ordinary state for a
forge-less install and every caller treats it as silence, not failure.
Forge errors (unreachable, bad token) RAISE — the two callers are a
refresh button and a background task, and both want to know.
"""
forge = forge if forge is not None else await get_forge()
if forge is None:
return None
repos: list[dict] = []
matched_all: list[tuple[str, str, str, bool]] = []
recorded = await _recorded_locations(user_id, project_id)
for key in await keys_for_project(user_id, project_id):
api_repo = forge.resolve_repo(key)
if api_repo is None:
continue # bound to a host this forge doesn't serve
ref = await forge.default_branch(api_repo)
shapes = shapes_from_archive(await forge.archive(api_repo, ref))
matched = match_shapes(shapes, recorded)
matched_all.extend(matched)
repos.append({
"repo": key,
"ref": ref,
"total": len(matched),
"recorded": sum(1 for *_x, covered in matched if covered),
})
if not repos:
return None
return {
"total": len(matched_all),
"recorded": sum(1 for *_x, covered in matched_all if covered),
# Honesty flag, not decoration: every surface that shows the number
# is expected to carry it through.
"estimate": True,
"computed_at": datetime.now(timezone.utc).isoformat(),
"repos": repos,
"largest_gaps": largest_gaps(matched_all),
}
async def refresh_coverage(
user_id: int, project_id: int, *, forge: ForgeAdapter | None = None
) -> dict | None:
"""Compute and cache. The only writer of the cache key."""
coverage = await compute_coverage(user_id, project_id, forge=forge)
if coverage is not None:
await set_setting(
user_id, f"{_CACHE_KEY_PREFIX}{project_id}", json.dumps(coverage)
)
return coverage
async def cached_coverage(user_id: int, project_id: int) -> dict | None:
"""The last computed summary, or None — never computes."""
raw = await get_setting(user_id, f"{_CACHE_KEY_PREFIX}{project_id}", "")
if not raw:
return None
try:
parsed = json.loads(raw)
except ValueError:
return None
return parsed if isinstance(parsed, dict) else None
def coverage_line(coverage: dict) -> str:
"""The one-line evidence-carrying summary enter_project surfaces."""
day = (coverage.get("computed_at") or "")[:10]
line = (
f"pattern-library coverage: {coverage.get('recorded', 0)}"
f"/{coverage.get('total', 0)} shapes recorded"
f" (estimate{', computed ' + day if day else ''})"
)
gaps = [g["dir"] for g in coverage.get("largest_gaps") or []]
if gaps:
line += "; largest gaps: " + ", ".join(gaps)
return line
+397
View File
@@ -0,0 +1,397 @@
"""Forge adapter — optional server-side READ access to the operator's git forge.
Step 4 of milestone 288 (#2689, decision #2686). The recorded location of a
snippet is the source of truth for its code and the stored body is a cache;
this module is the seam that lets the SERVER read that source of truth, so the
cache can be refreshed at pull time (step 5), drift can be flagged from push
webhooks (step 6), and coverage can be measured (step 7).
Design constraints, in force everywhere below:
- OPTIONAL per instance (rule #115). `get_forge()` returns None when nothing
is configured, and every consumer must treat None as "keep today's
behavior". An install that never configures a forge is not degraded — it
is the baseline.
- READ-ONLY by construction. The adapter exposes reads; there is no write
method to misuse. The token an operator mints for it only ever needs read
scope, and the docs say so.
- The contract stays as small as its consumers (steps 5-7): read_file /
latest_commit / archive / default_branch / resolve_repo / check. Two
implementations (Gitea, GitHub — step 8) keep it honest; resist widening
it speculatively.
- Repo identity is the repo-binding key — `normalize_repo_key`'s
host/owner/repo — so the join between a snippet's recorded repo and the
forge needs no new identity scheme. The host segment selects whether THIS
forge can serve the repo; the remainder is the API path.
- Errors carry no token, ever, and failures are exceptions the caller
handles — a consumer decides whether to fall back (pull-time fetch) or
surface (settings test button); this module never silently swallows.
This is also the codebase's first outbound-HTTP client with a real timeout
convention (oauth.py predates it): short total timeout, no retries — every
consumer has a fallback, so a slow forge must cost bounded time.
"""
from __future__ import annotations
import base64
import binascii
import logging
from dataclasses import dataclass
from urllib.parse import quote, urlsplit
import httpx
from scribe.config import Config
from scribe.services.repo_bindings import normalize_repo_key
from scribe.services.settings import get_admin_setting
logger = logging.getLogger(__name__)
FORGE_KIND_KEY = "forge_kind"
FORGE_BASE_URL_KEY = "forge_base_url"
FORGE_TOKEN_KEY = "forge_token"
# Kinds an instance can configure. Matches _FORGE_CLASSES below.
FORGE_KINDS = ("gitea", "github")
# Total budget per forge call. Consumers either have a cache to fall back to
# (step 5) or a user watching a button (the test probe) — neither tolerates a
# hung socket, and there is no retry: the fallback IS the retry policy.
_TIMEOUT = httpx.Timeout(5.0)
# Archive downloads move a whole-repo tarball and only ever run off the
# request path (coverage recompute, step 7), so they get a bigger budget than
# the per-file reads — but still a bound, because a hung background task
# holds a connection slot as surely as a foreground one.
_ARCHIVE_TIMEOUT = httpx.Timeout(60.0)
class ForgeError(RuntimeError):
"""A forge call failed (network, auth, unexpected payload). Token-free."""
class ForgeNotFound(ForgeError):
"""The repo, path, or ref does not exist on the forge — the one failure
consumers treat differently, because for a recorded snippet location it is
itself a finding (the recorded path is gone)."""
@dataclass(frozen=True)
class ForgeFile:
"""One file read from the forge at a specific point in history."""
content: str
# The commit the content was served at — what provenance stores (#2688).
commit_sha: str
path: str
def _host_of(url: str) -> str:
return (urlsplit(url).hostname or "").lower()
class ForgeAdapter:
"""The shared plumbing of the forge contract; adapters supply the API
base, auth headers, and any endpoint that differs.
`transport` exists for tests: httpx.MockTransport makes the contract
testable without a live server or a new dependency. Production callers
never pass it.
"""
kind = ""
# One "newest commit for this path" page — the endpoint is shared but the
# page-size parameter is not, so each adapter names its own.
_commit_page_params: dict = {}
def __init__(self, base_url: str, token: str, *, transport=None) -> None:
self.base_url = (base_url or "").rstrip("/")
self._token = token or ""
self._transport = transport
@property
def host(self) -> str:
return _host_of(self.base_url)
def resolve_repo(self, repo_or_url: str) -> str | None:
"""The forge-API repo path for a recorded repo — or None if this forge
does not serve it.
Accepts anything `normalize_repo_key` accepts (a raw remote URL or an
already-normalized key). None is a NORMAL answer, not an error: a
snippet recorded against github.com on an instance whose forge is a
self-hosted Gitea is simply out of this forge's reach.
"""
key = normalize_repo_key(repo_or_url or "")
if not key or "/" not in key:
return None
host, _, rest = key.partition("/")
if host != self.host or "/" not in rest:
return None
return rest
def _api_base(self) -> str:
raise NotImplementedError
def _headers(self) -> dict:
raise NotImplementedError
def _client(self) -> httpx.AsyncClient:
kwargs: dict = {
"base_url": self._api_base(),
"headers": self._headers(),
"timeout": _TIMEOUT,
# GitHub serves tarballs via a 302 to codeload. httpx drops the
# Authorization header on the cross-host hop, and GitHub's
# redirect target carries its own short-lived token in the URL —
# so following is both necessary there and harmless on Gitea.
"follow_redirects": True,
}
if self._transport is not None:
kwargs["transport"] = self._transport
return httpx.AsyncClient(**kwargs)
async def _get(self, client: httpx.AsyncClient, url: str, **kw) -> httpx.Response:
try:
resp = await client.get(url, **kw)
except httpx.HTTPError as exc:
# str(exc) on transport errors names hosts and timeouts, never
# headers — safe, and the detail is what makes the test button useful.
raise ForgeError(f"forge unreachable: {exc}") from exc
if resp.status_code == 404:
raise ForgeNotFound(f"not found on forge: {url}")
if resp.status_code in (401, 403):
raise ForgeError("forge rejected the token (check its read scope)")
if resp.status_code >= 400:
raise ForgeError(f"forge returned HTTP {resp.status_code} for {url}")
return resp
def _decode_contents(self, payload, path: str) -> str:
"""Both forges speak the same contents-API dialect: a base64 file
object, a list for a directory."""
if isinstance(payload, list):
raise ForgeNotFound(f"{path} is a directory on the forge, not a file")
if payload.get("type") != "file":
raise ForgeNotFound(
f"{path} is a {payload.get('type', 'non-file')} on the forge"
)
if payload.get("encoding") != "base64" or payload.get("content") is None:
raise ForgeError(f"forge returned no readable content for {path}")
try:
return base64.b64decode(payload["content"]).decode("utf-8")
except (binascii.Error, UnicodeDecodeError) as exc:
raise ForgeError(f"forge content for {path} is not utf-8 text") from exc
async def _newest_commit(
self, client: httpx.AsyncClient, repo: str, path: str, ref: str
) -> str:
params: dict = {**self._commit_page_params, "path": path}
if ref:
params["sha"] = ref
resp = await self._get(client, f"/repos/{repo}/commits", params=params)
payload = resp.json()
# Tolerant parse on purpose: the caller uses this as an optimization
# and falls back to read_file, so a surprising payload must read as
# "don't know", never break a pull.
if isinstance(payload, list) and payload and isinstance(payload[0], dict):
return str(payload[0].get("sha") or "")
return ""
async def latest_commit(self, repo: str, path: str, ref: str = "") -> str:
"""The newest commit touching ``path`` — "" when it can't be told.
The cached-SHA short-circuit (#2693): when a snippet's provenance
already names a commit, this one small call can prove the file
hasn't moved since — no content transfer, which is what keeps
pull-time freshness inside GitHub's rate limits.
"""
async with self._client() as client:
return await self._newest_commit(client, repo, path, ref)
def _archive_url(self, repo: str, ref: str) -> str:
raise NotImplementedError
async def archive(self, repo: str, ref: str) -> bytes:
"""The repo's content at ``ref`` as a gzipped tarball, in one request.
Coverage measurement (step 7) needs every source file's text; per-file
reads would mean one API call per file, so the archive endpoint is the
only shape that scales past toy repos. Callers must never run this in
a request path — it moves the whole repo.
"""
async with self._client() as client:
resp = await self._get(
client, self._archive_url(repo, ref), timeout=_ARCHIVE_TIMEOUT
)
return resp.content
async def default_branch(self, repo: str) -> str:
async with self._client() as client:
resp = await self._get(client, f"/repos/{repo}")
branch = (resp.json() or {}).get("default_branch") or ""
if not branch:
raise ForgeError(f"forge reported no default branch for {repo}")
return branch
class GiteaForge(ForgeAdapter):
"""The Gitea implementation of the forge contract, over its REST API."""
kind = "gitea"
# stat/verification/files add per-commit work Gitea skips when told to.
_commit_page_params = {"limit": 1, "stat": "false"}
def _api_base(self) -> str:
return f"{self.base_url}/api/v1"
def _headers(self) -> dict:
return {"Authorization": f"token {self._token}"}
async def read_file(self, repo: str, path: str, ref: str = "") -> ForgeFile:
"""Read one file's current content, with the commit it was served at.
`repo` is the API path from resolve_repo ("owner/repo"); `ref` is a
branch, tag, or commit — empty means the default branch.
"""
params = {"ref": ref} if ref else None
async with self._client() as client:
resp = await self._get(
client,
f"/repos/{repo}/contents/{quote(path, safe='/')}",
params=params,
)
payload = resp.json()
content = self._decode_contents(payload, path)
return ForgeFile(
content=content,
# last_commit_sha is the commit that last touched the file — the
# honest provenance stamp. The blob sha is a content address, not
# a point in history, so it is deliberately not surfaced.
commit_sha=payload.get("last_commit_sha") or "",
path=payload.get("path") or path,
)
def _archive_url(self, repo: str, ref: str) -> str:
return f"/repos/{repo}/archive/{quote(ref, safe='')}.tar.gz"
async def check(self) -> dict:
"""Health probe for the settings test button: reach the forge AND
prove the token is accepted. Returns {"ok", "version", "username"}."""
async with self._client() as client:
version = (await self._get(client, "/version")).json() or {}
user = (await self._get(client, "/user")).json() or {}
return {
"ok": True,
"version": version.get("version") or "",
"username": user.get("login") or user.get("username") or "",
}
class GitHubForge(ForgeAdapter):
"""The GitHub implementation — the second one, which is the point (#2693):
it proves the seam is a contract rather than a Gitea-shaped hole. Works
against github.com and GitHub Enterprise; the token is a fine-grained PAT
with Contents: Read-only (or a classic token with `repo` read)."""
kind = "github"
_commit_page_params = {"per_page": 1}
_API_VERSION = "2022-11-28"
def _api_base(self) -> str:
# github.com's API lives on its own host; GitHub Enterprise serves
# the same API under the instance at /api/v3.
if self.host == "github.com":
return "https://api.github.com"
return f"{self.base_url}/api/v3"
def _headers(self) -> dict:
return {
"Authorization": f"Bearer {self._token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": self._API_VERSION,
}
async def read_file(self, repo: str, path: str, ref: str = "") -> ForgeFile:
"""Same contents-API dialect as Gitea, minus one field: GitHub's
payload carries only the blob sha — a content address, not a point in
history — so the provenance stamp costs one extra commits call. ""
when even that can't be told; consumers already treat an empty stamp
as "don't restamp"."""
params = {"ref": ref} if ref else None
async with self._client() as client:
resp = await self._get(
client,
f"/repos/{repo}/contents/{quote(path, safe='/')}",
params=params,
)
payload = resp.json()
content = self._decode_contents(payload, path)
try:
commit_sha = await self._newest_commit(client, repo, path, ref)
except ForgeError:
commit_sha = ""
return ForgeFile(
content=content,
commit_sha=commit_sha,
path=payload.get("path") or path,
)
def _archive_url(self, repo: str, ref: str) -> str:
return f"/repos/{repo}/tarball/{quote(ref, safe='')}"
async def check(self) -> dict:
"""GitHub has no /version endpoint; proving the token against /user
is the whole probe, and the pinned API version stands in as the
version string."""
async with self._client() as client:
user = (await self._get(client, "/user")).json() or {}
return {
"ok": True,
"version": f"GitHub API {self._API_VERSION}",
"username": user.get("login") or "",
}
_FORGE_CLASSES: dict[str, type[ForgeAdapter]] = {
"gitea": GiteaForge,
"github": GitHubForge,
}
async def forge_config() -> dict:
"""The instance's forge configuration, DB-first with env fallback.
The env channel exists so a deployment can keep the token out of the
database entirely (Docker secret via FORGE_TOKEN_FILE) — the DB value wins
when both are present because the admin UI writes there, and a UI edit
that silently loses to an env var would look exactly like a broken form.
"""
return {
"kind": (await get_admin_setting(FORGE_KIND_KEY, "") or Config.FORGE_KIND)
.strip()
.lower(),
"base_url": (
await get_admin_setting(FORGE_BASE_URL_KEY, "") or Config.FORGE_BASE_URL
).rstrip("/"),
"token": await get_admin_setting(FORGE_TOKEN_KEY, "") or Config.FORGE_TOKEN,
}
async def get_forge(*, transport=None) -> ForgeAdapter | None:
"""The configured forge adapter, or None — and None means "behave exactly
as if this module did not exist", which every consumer must honor."""
cfg = await forge_config()
cls = _FORGE_CLASSES.get(cfg["kind"])
if cls is None:
if cfg["kind"]:
# A kind we don't implement is a misconfiguration, not "off" —
# say so once per lookup rather than silently reading as absent.
logger.warning("unknown forge kind %r configured — forge disabled", cfg["kind"])
return None
if not cfg["base_url"] or not cfg["token"]:
return None
if not cfg["base_url"].startswith(("http://", "https://")):
logger.warning("forge base URL %r has no http(s) scheme — forge disabled", cfg["base_url"])
return None
return cls(cfg["base_url"], cfg["token"], transport=transport)
+17 -4
View File
@@ -142,20 +142,25 @@ def verification_matches(data: dict | None, value: str) -> bool:
if not status:
return want == "unverified"
expired = verdict.get("code_sha") != (data or {}).get("code_sha")
# A push touched the recorded location since the verdict (#2691): the repo
# moved under it, so it needs a look even though it hasn't failed.
invalidated = bool(verdict.get("invalidated_by"))
if want == "unverified":
return False
if want == "drifted":
return status != "ok"
if want == "attention":
return status != "ok" or expired
return status != "ok" or expired or invalidated
if want == "ok":
return status == "ok" and not expired
return status == "ok" and not expired and not invalidated
return status == want
_VERIFY_DRIFTED_JSONPATH = '$.verification ? (@.status != "ok")'
_VERIFY_EXPIRED_JSONPATH = "$ ? (@.verification.code_sha != @.code_sha)"
_VERIFY_ANY_JSONPATH = "$.verification"
# A push touched the recorded location since the verdict (#2691).
_VERIFY_INVALIDATED_JSONPATH = "$.verification.invalidated_by"
def _verification_clause(value: str):
@@ -171,19 +176,23 @@ def _verification_clause(value: str):
if want == "drifted":
return Note.data.path_exists(_VERIFY_DRIFTED_JSONPATH)
if want == "attention":
# Everything worth looking at: a failing verdict, OR an expired one.
# Everything worth looking at: a failing verdict, an expired one, OR
# one whose recorded location a push has since touched (#2691).
return or_(
Note.data.path_exists(_VERIFY_DRIFTED_JSONPATH),
and_(has_verdict, Note.data.path_exists(_VERIFY_EXPIRED_JSONPATH)),
Note.data.path_exists(_VERIFY_INVALIDATED_JSONPATH),
)
if want == "ok":
# A clean bill of health that still describes the current code. The
# `~expired` half matters: without it this would quietly include records
# whose blessing has lapsed, which is the exact failure the feature is
# meant to catch.
# meant to catch. Same for push-invalidation — "ok" must mean the repo
# hasn't moved under the verdict either.
return and_(
Note.data.path_exists('$.verification ? (@.status == "ok")'),
~Note.data.path_exists(_VERIFY_EXPIRED_JSONPATH),
~Note.data.path_exists(_VERIFY_INVALIDATED_JSONPATH),
)
# A specific status: 'missing' | 'moved' | 'changed'.
return Note.data.path_exists(
@@ -229,6 +238,10 @@ def _note_to_item(note: Note) -> dict:
"detail": verdict.get("detail"),
"path": verdict.get("path"),
}
# Present only when a push has touched the recorded location since the
# verdict (#2691) — the "recheck me" marker, cleared by re-verifying.
if verdict.get("invalidated_by"):
item["verification"]["invalidated_by"] = verdict["invalidated_by"]
# Task fields — override note_type and add status/priority/due_date
if note.is_task:
+35
View File
@@ -100,6 +100,41 @@ async def list_bindings(user_id: int) -> list[RepoBinding]:
return list(rows.scalars().all())
async def keys_for_project(user_id: int, project_id: int) -> list[str]:
"""Every repo key bound to a project — the snippet→forge join (#2691).
Recorded snippet locations carry free-form repo names ("Scribe"), which
can't address a forge API. The project's binding is the identity that can:
a snippet reaches its forge repo through the project it belongs to.
"""
async with async_session() as session:
rows = await session.execute(
select(RepoBinding.repo_key).where(
RepoBinding.user_id == user_id,
RepoBinding.project_id == project_id,
)
)
return [k for (k,) in rows.all()]
async def bindings_for_key(raw_repo: str) -> list[RepoBinding]:
"""All bindings (ANY user) for a repo key — the webhook's entry point.
A push webhook carries no Scribe caller, only the repository it happened
to; the flag it writes is about each record's truth, so every user who
bound the repo gets their project's snippets considered — each write still
lands as that record's owner.
"""
key = normalize_repo_key(raw_repo)
if not key:
return []
async with async_session() as session:
rows = await session.execute(
select(RepoBinding).where(RepoBinding.repo_key == key)
)
return list(rows.scalars().all())
async def delete_binding(user_id: int, raw_repo: str) -> bool:
"""Remove a repo's binding. Returns True if a row was deleted."""
key = normalize_repo_key(raw_repo)
+371 -10
View File
@@ -28,6 +28,7 @@ came from.
"""
from __future__ import annotations
import asyncio
import hashlib
import logging
import re
@@ -342,7 +343,7 @@ def parse_snippet_fields(
# and copying a blob into the column we index *around* would be pure weight.
_DATA_FIELDS = (
"name", "when_to_use", "signature", "language", "locations", "merged_from",
"verification",
"verification", "provenance",
)
# --- drift check (#2086) -----------------------------------------------------
@@ -379,16 +380,20 @@ VERIFY_STATUSES = (VERIFY_OK, VERIFY_MISSING, VERIFY_MOVED, VERIFY_CHANGED)
VERIFY_DRIFTED = (VERIFY_MISSING, VERIFY_MOVED, VERIFY_CHANGED)
def _normalized_code(code: str) -> str:
"""Whitespace normalization shared by the verdict hash and the pull-time
containment check, so 'unchanged' means the same thing in both places:
trailing whitespace per line and leading/trailing blank lines dropped."""
return "\n".join(line.rstrip() for line in (code or "").splitlines()).strip()
def code_sha(code: str) -> str:
"""Stable fingerprint of a snippet's code, for expiring stale verdicts.
Trailing whitespace per line and leading/trailing blank lines are stripped
before hashing: those change when a file is reformatted without the code
meaning anything different, and a verdict shouldn't expire over an editor's
trailing-newline habit.
Normalized first (see _normalized_code): a reformat that changes nothing
shouldn't expire a verdict over an editor's trailing-newline habit.
"""
normalized = "\n".join(line.rstrip() for line in (code or "").splitlines()).strip()
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:32]
return hashlib.sha256(_normalized_code(code).encode("utf-8")).hexdigest()[:32]
def compose_verification(
@@ -398,6 +403,7 @@ def compose_verification(
detail: str = "",
path: str = "",
checked_at: str = "",
commit_sha: str = "",
) -> dict:
"""Build the `data.verification` record. Unknown statuses are rejected here
rather than stored, so the filter never has to cope with a typo'd status."""
@@ -414,9 +420,36 @@ def compose_verification(
out["detail"] = detail.strip()
if (path or "").strip():
out["path"] = path.strip()
# The repo commit the working tree was at when the check ran (#2688). The
# code_sha above expires a verdict when the RECORD is edited; this makes
# "the REPO moved on since the check" computable too, once the forge
# integration can compare it against the current head.
if (commit_sha or "").strip():
out["commit_sha"] = commit_sha.strip()
return out
def compose_provenance(*, commit_sha: str, fetched_at: str = "") -> dict | None:
"""Build the `data.provenance` record: which commit the cached body was
read at, and when.
This is the pointer-model half of decision #2686 — the recorded location
is the source of truth for the code and the stored body is a CACHE of it.
Provenance says what that cache is a cache OF, so a reader (and later the
forge fetch, step 5 of milestone 288) can judge staleness instead of
guessing. Absent provenance is valid and means exactly what every snippet
meant before this existed: a body captured by hand at an unknown point.
"""
sha = (commit_sha or "").strip()
if not sha:
return None
return {
"commit_sha": sha,
"fetched_at": (fetched_at or "").strip()
or datetime.now(timezone.utc).isoformat(),
}
def verification_view(note, fields: dict) -> dict:
"""The verification readout for one snippet, including whether it's expired.
@@ -434,10 +467,20 @@ def verification_view(note, fields: dict) -> dict:
"checked_at": stored.get("checked_at"),
"detail": stored.get("detail"),
"path": stored.get("path"),
"commit_sha": stored.get("commit_sha"),
# A push touched the recorded location since this verdict (#2691) —
# the repo moved under it. Cleared by the next verdict, which builds
# a fresh dict.
"invalidated_by": stored.get("invalidated_by"),
# What the operator actually wants to know: is there something to fix?
# An expired verdict counts as "needs looking at" even if it said ok,
# since the code it blessed is not the code that's there now.
"needs_attention": (not current) or stored["status"] in VERIFY_DRIFTED,
# since the code it blessed is not the code that's there now — and so
# does a push-invalidated one, for the same reason from the repo side.
"needs_attention": (
(not current)
or stored["status"] in VERIFY_DRIFTED
or bool(stored.get("invalidated_by"))
),
}
@@ -451,6 +494,7 @@ def compose_data(
locations: list[dict] | None = None,
merged_from: list[int] | None = None,
verification: dict | None = None,
provenance: dict | None = None,
) -> dict:
"""Build the `notes.data` mirror of a snippet's structured fields.
@@ -479,6 +523,12 @@ def compose_data(
# either: the verdict's code_sha expires it on read if the code moved on.
if verification:
out["verification"] = verification
# Also carried: what commit the cached body was read at (#2688). The caller
# owns the live-or-die rule — update_snippet drops it when the code changes
# without a fresh SHA, because keeping it would claim the new body came
# from the old commit.
if provenance:
out["provenance"] = provenance
# The current code's fingerprint — NOT the code, which stays in the body
# (see _DATA_FIELDS). Its only job is to make "this verdict has expired"
# expressible in SQL: a jsonpath can compare `@.verification.code_sha` to
@@ -616,10 +666,15 @@ async def create_snippet(
locations: list[dict] | None = None,
tags: list[str] | None = None,
project_id: int | None = None,
commit_sha: str = "",
):
"""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."""
``repo``/``path``/``symbol`` are the one-location shorthand.
``commit_sha`` stamps the body's provenance — the commit the recording
session read the code at (#2688). Optional: absent means what it always
meant, a body captured at an unknown point."""
locations = resolve_locations(repo, path, symbol, locations)
note = await notes_svc.create_note(
user_id,
@@ -636,6 +691,7 @@ async def create_snippet(
data=compose_data(
name=name, when_to_use=when_to_use, signature=signature,
language=language, code=code, locations=locations,
provenance=compose_provenance(commit_sha=commit_sha),
),
)
return note
@@ -714,11 +770,17 @@ async def update_snippet(
locations: list[dict] | None = None,
tags: list[str] | None = None,
project_id: int | None | object = UNSET,
commit_sha: str | None = None,
):
"""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.
``commit_sha`` restamps the body's provenance (#2688). It lives or dies
with the code: passed → restamped at that commit; code changed without it →
dropped, because keeping it would claim the new body came from the old
commit; code untouched → carried.
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
@@ -761,6 +823,17 @@ async def update_snippet(
else:
merged_locations = cur["locations"]
# Provenance follows the code (#2688): a fresh SHA restamps it; a code
# change without one drops it; an edit that leaves the code alone carries
# it. Order matters — the explicit SHA wins even when the code changed,
# because that is precisely the caller saying where the new body came from.
if commit_sha is not None and commit_sha.strip():
provenance = compose_provenance(commit_sha=commit_sha)
elif code is not None and code != (cur.get("code") or ""):
provenance = None
else:
provenance = cur.get("provenance")
fields: dict = {
"title": compose_title(merged["name"], merged["when_to_use"]),
"body": compose_body(
@@ -782,6 +855,7 @@ async def update_snippet(
# the code, the verdict's code_sha stops matching and it reads as
# unverified from here on — no invalidation branch to get wrong.
verification=merged.get("verification"),
provenance=provenance,
),
}
# Recompute tags: keep any non-language, non-marker tags the note already had
@@ -808,6 +882,7 @@ async def record_verification(
status: str,
detail: str = "",
path: str = "",
commit_sha: str = "",
):
"""Record the result of a drift check against the snippet's source.
@@ -836,6 +911,7 @@ async def record_verification(
checked_code_sha=code_sha(fields.get("code") or ""),
detail=detail,
path=path or fields.get("path") or "",
commit_sha=commit_sha,
)
# Rebuilt from the CURRENT stored fields plus the new verdict, so recording a
# check can't quietly rewrite anything else about the record. Note the body
@@ -850,10 +926,295 @@ async def record_verification(
locations=fields.get("locations") or [],
merged_from=fields.get("merged_from") or [],
verification=verification,
# An "ok" verdict at a known commit IS a provenance claim — the checker
# just established that the cached body matches the source there — so
# it restamps. Any other verdict carries what was known: a verdict is
# about the code, not a change to it, and must not erase it (#2688).
provenance=(
compose_provenance(commit_sha=commit_sha)
if status == VERIFY_OK and (commit_sha or "").strip()
else fields.get("provenance")
),
)
return await notes_svc.update_note(note.user_id, snippet_id, data=data)
# --- pull-time freshness (#2690) ---------------------------------------------
# A pull is the moment freshness matters: the reader is about to trust the
# cached body. When the instance has a forge configured, the pull fetches the
# recorded file and answers the one mechanically-answerable question — does
# the cached code still appear in the source, verbatim after whitespace
# normalization? The body is a FRAGMENT of the file, so "serve the fetched
# file" would clobber the record; confirmation + provenance refresh is what
# fetching can honestly deliver, and divergence is reported, not overwritten.
#
# With no forge configured this function attaches NOTHING — the response is
# byte-identical to pre-forge behavior (rule #115's baseline).
# Total budget for the in-pull fetch. Tighter than the adapter's own timeout:
# the pull is the moment a session decides whether pulling is worth it
# (#2663's pull-through finding), so a slow forge must cost bounded time and
# then the cache serves.
PULL_FETCH_BUDGET_S = 2.5
async def _stamp_missing(note, host: str) -> None:
"""Record the mechanically-established 'missing' verdict from a pull-time
404 — the recorded path is gone at the forge's head. Runs in the
background; written as the owner, like every metadata write here."""
await record_verification(
note.user_id, note.id, status=VERIFY_MISSING,
detail=f"pull-time forge fetch: recorded path not found on {host}",
)
async def _refresh_provenance(note, commit_sha: str) -> None:
"""Restamp data.provenance after a pull confirmed the cache matches the
source at ``commit_sha``. Background write, rebuilt like record_verification
so nothing else about the record changes."""
fields = snippet_fields(note)
data = compose_data(
name=fields.get("name", ""),
when_to_use=fields.get("when_to_use", ""),
signature=fields.get("signature", ""),
language=fields.get("language", ""),
code=fields.get("code", ""),
locations=fields.get("locations") or [],
merged_from=fields.get("merged_from") or [],
verification=fields.get("verification"),
provenance=compose_provenance(commit_sha=commit_sha),
)
await notes_svc.update_note(note.user_id, note.id, data=data)
async def attach_live_body(note, data: dict) -> None:
"""Decorate a PULL response with forge-checked freshness (#2690).
Adds, when (and only when) a forge is configured:
- ``body_source``: "forge" (confirmed against the source just now) or
"cache" (the stored body, for whatever reason follows)
- ``body_freshness``: "current" | "diverged" | "missing" |
"unreachable" | "no-recorded-location" | "repo-not-on-this-forge"
Never raises, never blocks past PULL_FETCH_BUDGET_S, never rewrites the
body: a freshness probe must not be able to break or slow the pull it
decorates, and divergence is the READER's information, not license to
clobber a record mid-read. A confirmed-current pull refreshes provenance
in the background; a 404 stamps the 'missing' verdict into the same
attention state verify_snippet uses.
"""
from scribe.services.background import spawn
from scribe.services.forge import ForgeError, ForgeNotFound, get_forge
try:
forge = await get_forge()
except Exception:
logger.warning("forge lookup failed during pull", exc_info=True)
return
if forge is None:
return
fields = data.get("snippet") if isinstance(data.get("snippet"), dict) else None
if fields is None:
fields = snippet_fields(note)
loc = next(
(
entry
for entry in (fields.get("locations") or [])
if entry.get("repo") and entry.get("path")
),
None,
)
if loc is None:
data["body_source"] = "cache"
data["body_freshness"] = "no-recorded-location"
return
# Recorded location repos are free-form names ("Scribe"), which can't
# address a forge API — the project's repo BINDING is the identity that
# can (#2691). Try the location string first (it may be a real remote),
# then fall back to the bindings of the snippet's project.
repo = forge.resolve_repo(loc["repo"])
if repo is None and getattr(note, "project_id", None):
from scribe.services.repo_bindings import keys_for_project
for key in await keys_for_project(note.user_id, note.project_id):
repo = forge.resolve_repo(key)
if repo is not None:
break
if repo is None:
data["body_source"] = "cache"
data["body_freshness"] = "repo-not-on-this-forge"
return
stored_prov_sha = (fields.get("provenance") or {}).get("commit_sha") or ""
async def _probe():
# Cached-SHA short-circuit (#2693): provenance names the commit the
# cached code was last confirmed at, so one cheap "newest commit
# touching this path" call can prove the file hasn't moved since —
# no content transfer. That economy is what fits pull-time freshness
# inside GitHub's rate limits; it's merely nice on a self-hosted
# Gitea. Any surprise (error, empty, mismatch) falls through to the
# full fetch, which stays the authoritative path.
if stored_prov_sha:
try:
head = await forge.latest_commit(repo, loc["path"])
except ForgeError:
head = ""
if head and head == stored_prov_sha:
return None
return await forge.read_file(repo, loc["path"])
try:
fetched = await asyncio.wait_for(_probe(), timeout=PULL_FETCH_BUDGET_S)
except ForgeNotFound:
data["body_source"] = "cache"
data["body_freshness"] = "missing"
stored = fields.get("verification") or {}
# Don't re-stamp what's already stamped — a popular-but-broken record
# would otherwise be rewritten on every pull.
if stored.get("status") != VERIFY_MISSING:
spawn(_stamp_missing(note, forge.host), site="pull missing-verdict")
return
except (ForgeError, asyncio.TimeoutError):
data["body_source"] = "cache"
data["body_freshness"] = "unreachable"
return
if fetched is None:
# Unchanged since the provenance commit — confirmed against the
# source without moving the file. Same stamp, so nothing to persist
# (the same-sha rule); the body already reflects that commit.
data["body_source"] = "forge"
data["body_freshness"] = "current"
return
cached = _normalized_code(fields.get("code") or "")
if cached and cached in _normalized_code(fetched.content):
data["body_source"] = "forge"
data["body_freshness"] = "current"
if fetched.commit_sha:
# Read the stored stamp BEFORE writing the fresh one into the
# response: `fields` aliases data["snippet"], so the other order
# makes the staleness check compare the new stamp to itself and
# the persist never fires (caught by the unit test, run 3811).
stored_prov = fields.get("provenance") or {}
stale = stored_prov.get("commit_sha") != fetched.commit_sha
prov = compose_provenance(commit_sha=fetched.commit_sha)
# Reflected in THIS response as well as persisted — the reader
# shouldn't need a second pull to see the stamp they caused.
if isinstance(data.get("snippet"), dict):
data["snippet"]["provenance"] = prov
if stale:
spawn(
_refresh_provenance(note, fetched.commit_sha),
site="pull provenance-refresh",
)
else:
data["body_source"] = "cache"
data["body_freshness"] = "diverged"
# --- push-time drift flagging (#2691) ----------------------------------------
def _path_touches(recorded: str, changed: str) -> bool:
"""The location-path semantics, applied to a pushed file: the recorded path
is the changed file itself, or a directory above it."""
recorded = (recorded or "").strip("/")
changed = (changed or "").strip("/")
if not recorded or not changed:
return False
return changed == recorded or changed.startswith(recorded + "/")
async def invalidate_for_push(
repo_key: str,
changed: list[str],
removed: list[str],
commit_sha: str,
) -> int:
"""Flag snippets whose recorded location a push just touched (#2691).
Writes ``verification.invalidated_by = {commit_sha, at, path, removed}``
onto matched snippets that CARRY a verdict — the flag means "the repo
moved under this verdict, recheck it", and it clears itself the moment a
fresh verdict is recorded because compose_verification builds a new dict.
Unverified snippets are skipped: they are already in the unverified
bucket, and stacking a second unchecked-flavored flag on them adds noise,
not information.
Matching goes through repo BINDINGS (any user's — a webhook has no
caller): each binding names a project, and that project's snippets are
path-matched against the pushed files. O(bindings + snippets-in-project +
changed files); nothing else is scanned. Returns how many records were
newly flagged (an already-flagged record at the same commit is skipped,
so replayed deliveries don't churn).
"""
from scribe.services.repo_bindings import bindings_for_key
bindings = await bindings_for_key(repo_key)
if not bindings or not (changed or removed):
return 0
touched = [(p, False) for p in changed] + [(p, True) for p in removed]
flagged = 0
for binding in bindings:
async with async_session() as session:
rows = await session.execute(
select(Note).where(
Note.user_id == binding.user_id,
Note.project_id == binding.project_id,
Note.note_type == SNIPPET_NOTE_TYPE,
Note.deleted_at.is_(None),
Note.data.path_exists("$.verification"),
)
)
notes = list(rows.scalars().all())
for note in notes:
fields = snippet_fields(note)
verdict = fields.get("verification") or {}
if not verdict.get("status"):
continue
hit = next(
(
(path, was_removed)
for loc in (fields.get("locations") or [])
for path, was_removed in touched
if _path_touches(loc.get("path") or "", path)
),
None,
)
if hit is None:
continue
existing = verdict.get("invalidated_by") or {}
if existing.get("commit_sha") == commit_sha:
continue # replayed delivery — already says exactly this
verdict = dict(verdict)
verdict["invalidated_by"] = {
"commit_sha": commit_sha,
"at": datetime.now(timezone.utc).isoformat(),
"path": hit[0],
# A removed file is the strongest signal — the recorded
# location may simply be gone. Surfaced so the attention row
# says which kind of look it needs.
"removed": hit[1],
}
data = compose_data(
name=fields.get("name", ""),
when_to_use=fields.get("when_to_use", ""),
signature=fields.get("signature", ""),
language=fields.get("language", ""),
code=fields.get("code", ""),
locations=fields.get("locations") or [],
merged_from=fields.get("merged_from") or [],
verification=verdict,
provenance=fields.get("provenance"),
)
await notes_svc.update_note(note.user_id, note.id, data=data)
flagged += 1
return flagged
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.
+250
View File
@@ -0,0 +1,250 @@
"""Forge push webhook (#2691) — signature gate, payload parsing, and the
end-to-end drift flag on real Postgres.
The webhook is the seam that makes verification scale: a push names exactly
which files moved, so only the records that point at them get flagged. The
properties pinned here:
- No secret configured → the endpoint does not exist (404); a bad
signature → 401. Both BEFORE any payload parsing.
- Matching is O(bindings + snippets-in-project + changed files) and goes
through repo bindings — recorded location repos are free-form names and
cannot address a forge.
- A replayed delivery (same head commit) flags nothing new; re-verifying
clears the flag by construction.
"""
import hashlib
import hmac
import pytest
import pytest_asyncio
from scribe.routes.webhooks import delivered_signature, push_facts, signature_ok
from scribe.services.snippets import _path_touches
SECRET = "wh-secret"
HEAD = "e" * 40
# --- unit: the signature gate ------------------------------------------------
def _sign(body: bytes, secret: str = SECRET) -> str:
return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
def test_signature_gate():
body = b'{"x": 1}'
assert signature_ok(SECRET, body, _sign(body)) is True
assert signature_ok(SECRET, body, _sign(body).upper()) is True # hex case
assert signature_ok(SECRET, body, _sign(body, "wrong")) is False
assert signature_ok(SECRET, body, "") is False
assert signature_ok("", body, _sign(body)) is False
def test_delivered_signature_reads_both_forges_headers():
"""#2693: GitHub signs the same HMAC but ships it as
X-Hub-Signature-256: sha256=<hex> — the whole webhook payload mapping is
this header, so pin it."""
hexsig = _sign(b"{}")
assert delivered_signature({"X-Gitea-Signature": hexsig}) == hexsig
assert delivered_signature({"X-Hub-Signature-256": f"sha256={hexsig}"}) == hexsig
# Gitea's header wins when both appear; absence reads as empty (→ 401).
assert delivered_signature({}) == ""
# The stripped GitHub form still passes the gate end to end.
assert signature_ok(
SECRET, b'{"x": 1}',
delivered_signature({"X-Hub-Signature-256": "sha256=" + _sign(b'{"x": 1}')}),
)
def test_push_facts_reads_a_github_shaped_payload():
"""GitHub's push payload carries the same fields push_facts consumes —
asserted against a real-shaped sample so a rename on either side of the
mapping breaks a test instead of silently flagging nothing."""
payload = {
"ref": "refs/heads/main",
"after": HEAD,
"repository": {
"full_name": "alice/widget",
"clone_url": "https://github.com/alice/widget.git",
"html_url": "https://github.com/alice/widget",
},
"commits": [
{"id": "a" * 40, "added": [], "modified": ["src/x.py"], "removed": []},
],
"head_commit": {"id": HEAD},
}
raw, changed, removed, head = push_facts(payload)
assert raw == "https://github.com/alice/widget.git"
assert changed == ["src/x.py"]
assert removed == []
assert head == HEAD
# --- unit: payload parsing ---------------------------------------------------
def test_push_facts_collects_and_dedups_paths():
payload = {
"after": HEAD,
"repository": {"clone_url": "https://git.example.com/alice/widget.git"},
"commits": [
{"added": ["a.py"], "modified": ["b.py"], "removed": []},
{"added": [], "modified": ["b.py", "c/d.py"], "removed": ["gone.py"]},
],
}
raw, changed, removed, head = push_facts(payload)
assert raw.endswith("alice/widget.git")
assert changed == ["a.py", "b.py", "c/d.py"]
assert removed == ["gone.py"]
assert head == HEAD
def test_push_facts_tolerates_an_empty_payload():
assert push_facts({}) == ("", [], [], "")
def test_path_touches_uses_the_location_semantics():
assert _path_touches("src/x.py", "src/x.py")
assert _path_touches("src", "src/lib/x.py") # recorded dir, file below
assert not _path_touches("src/x.py", "src/x_test.py")
assert not _path_touches("src/lib", "src/library/x.py") # no prefix bleed
assert not _path_touches("", "src/x.py")
def test_route_is_registered_and_unauthenticated_by_design():
from scribe.app import create_app
from scribe.routes import webhooks as wh
assert callable(wh.forge_push)
rules = {r.rule for r in create_app().url_map.iter_rules()}
assert "/api/webhooks/forge" in rules
# --- integration: the flag lands and clears on real Postgres -----------------
@pytest_asyncio.fixture
async def _dispose_engine():
from scribe.models import engine
yield
await engine.dispose()
@pytest_asyncio.fixture
async def seeded(_dispose_engine):
"""User + project + binding + two verified snippets + one unverified."""
from sqlalchemy import select
from scribe.models import async_session
from scribe.models.project import Project
from scribe.models.user import User
from scribe.services import snippets as svc
from scribe.services.repo_bindings import set_binding
async with async_session() as s:
user = (
await s.execute(select(User).where(User.username == "webhook_itest"))
).scalar_one_or_none()
if user is None:
user = User(username="webhook_itest")
s.add(user)
await s.flush()
project = Project(user_id=user.id, title="Widget")
s.add(project)
await s.flush()
uid, pid = user.id, project.id
await s.commit()
await set_binding(uid, "https://git.example.com/alice/widget.git", pid)
async def _snippet(name, path, code, verify=True):
note = await svc.create_snippet(
uid, name=name, code=code, language="python",
repo="Widget", path=path, symbol=name, project_id=pid,
)
if verify:
await svc.record_verification(uid, note.id, status="ok")
return note.id
return {
"uid": uid,
"hit": await _snippet("wh_hit", "src/x.py", "def wh_hit():\n return 1\n"),
"miss": await _snippet("wh_miss", "src/other.py", "def wh_miss():\n return 2\n"),
"gone": await _snippet("wh_gone", "src/gone.py", "def wh_gone():\n return 3\n"),
"unchecked": await _snippet(
"wh_unchecked", "src/x.py", "def wh_unchecked():\n return 4\n",
verify=False,
),
}
@pytest.mark.integration
async def test_push_flags_matched_verdicts_and_replay_is_quiet(seeded):
from scribe.services import snippets as svc
flagged = await svc.invalidate_for_push(
"git.example.com/alice/widget",
changed=["src/x.py"], removed=["src/gone.py"], commit_sha=HEAD,
)
# wh_hit (modified) + wh_gone (removed). wh_miss untouched; wh_unchecked
# carries no verdict and is skipped by design.
assert flagged == 2
uid = seeded["uid"]
hit = svc.snippet_to_dict(await svc.get_snippet(uid, seeded["hit"]))
assert hit["verification"]["needs_attention"] is True
assert hit["verification"]["invalidated_by"]["commit_sha"] == HEAD
assert hit["verification"]["invalidated_by"]["removed"] is False
gone = svc.snippet_to_dict(await svc.get_snippet(uid, seeded["gone"]))
assert gone["verification"]["invalidated_by"]["removed"] is True
miss = svc.snippet_to_dict(await svc.get_snippet(uid, seeded["miss"]))
assert miss["verification"]["needs_attention"] is False
unchecked = svc.snippet_to_dict(await svc.get_snippet(uid, seeded["unchecked"]))
assert unchecked["verification"]["status"] == "unverified"
# The attention listing — the operator's single entry point — now shows
# exactly the flagged pair, through the SQL dialect.
items, total = await svc.list_snippets(uid, verification="attention")
ids = {i["id"] for i in items}
assert {seeded["hit"], seeded["gone"]} <= ids
assert seeded["miss"] not in ids
# Replayed delivery: same head commit flags nothing new.
again = await svc.invalidate_for_push(
"git.example.com/alice/widget",
changed=["src/x.py"], removed=["src/gone.py"], commit_sha=HEAD,
)
assert again == 0
@pytest.mark.integration
async def test_reverifying_clears_the_flag(seeded):
from scribe.services import snippets as svc
await svc.invalidate_for_push(
"git.example.com/alice/widget", changed=["src/x.py"], removed=[],
commit_sha=HEAD,
)
uid = seeded["uid"]
await svc.record_verification(
uid, seeded["hit"], status="ok", detail="rechecked after push",
commit_sha=HEAD,
)
hit = svc.snippet_to_dict(await svc.get_snippet(uid, seeded["hit"]))
assert hit["verification"]["needs_attention"] is False
assert hit["verification"]["invalidated_by"] is None
items, _ = await svc.list_snippets(uid, verification="attention")
assert seeded["hit"] not in {i["id"] for i in items}
@pytest.mark.integration
async def test_unbound_repo_flags_nothing(seeded):
from scribe.services import snippets as svc
flagged = await svc.invalidate_for_push(
"github.com/somebody/else", changed=["src/x.py"], removed=[],
commit_sha=HEAD,
)
assert flagged == 0
+12 -10
View File
@@ -129,21 +129,23 @@ def test_floor_states_the_systems_reflex():
def test_floor_names_the_snippet_recording_triggers():
"""The recording half of reuse needs NAMED trigger moments on the floor.
"""The floor must state the pattern-library recording model, by name.
#2664's behavioral finding: with recording guidance as a trailing clause of
the reuse bullet, zero snippets were ever recorded outside sessions already
thinking about snippets — extracting a shared component (Roundtable's
BaseModal) produced task prose and no record. The floor must name the
moments, not just the tool.
#2664's behavioral finding: recording guidance as a trailing clause of the
reuse bullet converted zero times outside snippet-minded sessions. The
2026-08-16 ruling (decision #2686) then replaced the reactive model
entirely: every shape is recorded at FIRST build — no "will it recur?"
judgment — and second-copy consolidation is only the backstop. The floor
is the delivery surface for that reflex, so all three elements must stay
stated: the tool, the first-build trigger, and the backstop.
"""
floor = (ROOT / "plugin" / "hooks" / "scribe_static_context.md").read_text()
for needle in ("create_snippet", "second copy"):
for needle in ("create_snippet", "first build", "second copy"):
assert needle in floor, (
f"plugin/hooks/scribe_static_context.md no longer states the "
f"snippet-recording trigger ({needle!r}) — the record-as-you-build "
f"reflex must be stated on the floor with its trigger moments "
f"(#2664)."
f"snippet-recording model ({needle!r}) — record-every-shape-at-"
f"first-build with second-copy consolidation as the backstop must "
f"be stated on the floor (#2664, decision #2686)."
)
+11
View File
@@ -29,6 +29,17 @@ def _no_systems():
yield
@pytest.fixture(autouse=True)
def _no_coverage():
"""enter_project also reads the pattern-coverage cache (#2692) — same
deal: no database here, stub the common case (nothing computed). The
populated line is asserted in tests/test_pattern_coverage.py.
"""
with patch("scribe.mcp.tools.projects.coverage_svc.cached_coverage",
AsyncMock(return_value=None)):
yield
def _fake_project(design_system_id=None, **overrides) -> MagicMock:
p = MagicMock()
base = {"id": 1, "title": "P", "description": "", "goal": "",
+317
View File
@@ -0,0 +1,317 @@
"""Pattern-library coverage (#2692) — the extractor that mirrors the hook,
the shape/record matcher, and the end-to-end measurement on real Postgres.
The extractor here and the hook's awk program (scribe_prior_art.sh ARM 1)
must agree on what counts as "a definition" — the metric and the write-path
backstop are two views of the same doctrine. The EXTRACTION_VECTORS below
deliberately reuse the definitions test_write_path_trigger stages for the
hook; extending one detector means extending both, and this comment is the
tripwire.
"""
import io
import json
import tarfile
import pytest
import pytest_asyncio
from scribe.services.coverage import (
coverage_line,
extract_shapes,
largest_gaps,
match_shapes,
scannable,
shapes_from_archive,
)
# --- unit: the definition extractor (shared vectors with the hook) -----------
EXTRACTION_VECTORS = [
# (id, source text, expected (kind, name) list)
("python", "def make_app():\n pass\nclass Config:\n pass\n",
[("sym", "make_app"), ("sym", "Config")]),
("python-dunder-skip", "class C:\n def __init__(self):\n pass\n",
[("sym", "C")]),
("go-func", "func Resolve(x int) error {\n\treturn nil\n}\n",
[("sym", "Resolve")]),
("go-method", "func (s *Scanner) Resolve(x int) error {\n\treturn nil\n}\n",
[("sym", "Resolve")]),
("kotlin-fun", "suspend fun refreshQueue(id: Long) {\n}\n",
[("sym", "refreshQueue")]),
("rust-fn", "pub async fn fetch_all() -> u32 {\n 0\n}\n",
[("sym", "fetch_all")]),
("go-type", "type ForgeAdapter struct {\n\tname string\n}\n",
[("sym", "ForgeAdapter")]),
("rust-pub-crate", "pub(crate) struct Widget {}\n",
[("sym", "Widget")]),
("js-export-default", "export default function App() {}\n",
[("sym", "App")]),
("js-arrow", "const useThing = (id) => id;\nlet fetcher = async () => 0;\n",
[("sym", "useThing"), ("sym", "fetcher")]),
# Every selector line in a group counts — .btn-ghost, and .btn-text { }
# both announce a class, exactly as the hook's awk sees them.
("css", ".btn-primary {\n color: red;\n}\n.btn-ghost,\n.btn-text { }\n",
[("css", "btn-primary"), ("css", "btn-ghost"), ("css", "btn-text")]),
# Call sites, imports, and impl blocks are NOT definitions — matching
# them would drown the metric exactly as it would drown the hook.
("non-definitions",
"make_app()\nimpl Widget {\nreturn fetch_all\nimport os\nx = 1\n",
[]),
("dedup-within-file", "def f():\n pass\ndef f():\n pass\n",
[("sym", "f")]),
]
@pytest.mark.parametrize(
("text", "expected"),
[(t, e) for _i, t, e in EXTRACTION_VECTORS],
ids=[i for i, _t, _e in EXTRACTION_VECTORS],
)
def test_extractor_agrees_with_the_hook_on_what_defines(text, expected):
assert extract_shapes(text) == expected
def test_scannable_gates_prose_vendored_and_sourcemaps():
assert scannable("src/app.py")
assert scannable("web/button.css")
assert scannable(".gitea/workflows/ci.yml") # config IS worth recording
assert not scannable("README.md")
assert not scannable("dist/bundle.js.map")
assert not scannable("node_modules/x/index.js")
assert not scannable("web/node_modules/y/util.ts")
# A FILE named like a skip-dir is not a directory hit.
assert scannable("src/vendor.py")
# --- unit: reading shapes out of a forge tarball -----------------------------
def _tarball(files: dict[str, bytes], top: str = "widget") -> bytes:
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
for path, data in files.items():
info = tarfile.TarInfo(f"{top}/{path}")
info.size = len(data)
tar.addfile(info, io.BytesIO(data))
return buf.getvalue()
TREE = {
"src/app.py": b"def make_app():\n pass\nclass Config:\n def __init__(self):\n pass\n",
"src/util.py": b"def helper():\n pass\n",
"web/button.css": b".btn {\n color: red;\n}\n",
"README.md": b"def not_code(): pass\n",
"node_modules/x/index.js": b"function vendored() {}\n",
"data.bin": b"\xff\xfe\x00\x01",
}
# What TREE holds once the gates run: 4 shapes, none from the skipped files.
TREE_SHAPES = [
("src/app.py", "sym", "make_app"),
("src/app.py", "sym", "Config"),
("src/util.py", "sym", "helper"),
("web/button.css", "css", "btn"),
]
def test_shapes_from_archive_strips_the_wrapper_and_gates_files():
assert shapes_from_archive(_tarball(TREE)) == TREE_SHAPES
# --- unit: matching shapes against recorded locations ------------------------
def test_match_covers_by_exact_path_dir_prefix_and_css_dot():
recorded = [
("src/app.py", "make_app"), # exact file
("web", ".btn"), # dir prefix + css dot normalization
]
matched = match_shapes(TREE_SHAPES, recorded)
covered = {name for _p, _k, name, ok in matched if ok}
assert covered == {"make_app", "btn"}
def test_a_symbol_less_record_covers_nothing():
"""A whole-file snippet makes no claim about any particular definition
inside it — crediting all of them would inflate the number for free."""
matched = match_shapes(TREE_SHAPES, [("src/app.py", "")])
assert not any(ok for *_x, ok in matched)
def test_no_prefix_bleed_between_sibling_directories():
matched = match_shapes(
[("src/library/x.py", "sym", "helper")], [("src/lib", "helper")]
)
assert not matched[0][3]
def test_largest_gaps_ranks_by_uncovered_and_drops_clean_dirs():
matched = match_shapes(TREE_SHAPES, [("src/app.py", "make_app"), ("web", ".btn")])
gaps = largest_gaps(matched)
assert gaps == [{"dir": "src", "uncovered": 2, "total": 3}]
def test_coverage_line_is_evidence_carrying_and_labeled_estimate():
line = coverage_line({
"total": 210, "recorded": 34, "estimate": True,
"computed_at": "2026-08-16T12:00:00+00:00",
"largest_gaps": [
{"dir": "internal/api", "uncovered": 40, "total": 60},
{"dir": "web/src/components", "uncovered": 25, "total": 30},
],
})
assert "34/210 shapes recorded" in line
assert "estimate" in line
assert "2026-08-16" in line
assert "internal/api, web/src/components" in line
def test_coverage_routes_are_registered():
from scribe.app import create_app
rules = {r.rule for r in create_app().url_map.iter_rules()}
assert "/api/projects/<int:project_id>/coverage" in rules
assert "/api/projects/<int:project_id>/coverage/refresh" in rules
# --- integration: the measurement end to end on real Postgres ----------------
def _forge(tar_bytes: bytes):
import httpx
from scribe.services.forge import GiteaForge
def handler(request: httpx.Request) -> httpx.Response:
path = request.url.path
if path == "/api/v1/repos/alice/widget":
return httpx.Response(200, json={"default_branch": "main"})
if path == "/api/v1/repos/alice/widget/archive/main.tar.gz":
return httpx.Response(200, content=tar_bytes)
return httpx.Response(404, json={"message": "not found"})
return GiteaForge(
"https://git.example.com", "tok", transport=httpx.MockTransport(handler)
)
@pytest_asyncio.fixture
async def _dispose_engine():
from scribe.models import engine
yield
await engine.dispose()
@pytest_asyncio.fixture
async def seeded(_dispose_engine):
"""User + project + binding + two snippets that cover 2 of TREE's 4 shapes."""
from sqlalchemy import select
from scribe.models import async_session
from scribe.models.project import Project
from scribe.models.user import User
from scribe.services import snippets as svc
from scribe.services.repo_bindings import set_binding
async with async_session() as s:
user = (
await s.execute(select(User).where(User.username == "coverage_itest"))
).scalar_one_or_none()
if user is None:
user = User(username="coverage_itest")
s.add(user)
await s.flush()
project = Project(user_id=user.id, title="Widget")
s.add(project)
await s.flush()
uid, pid = user.id, project.id
await s.commit()
await set_binding(uid, "https://git.example.com/alice/widget.git", pid)
await svc.create_snippet(
uid, name="cov_make_app", code="def make_app():\n pass\n",
language="python", repo="Widget", path="src/app.py",
symbol="make_app", project_id=pid,
)
await svc.create_snippet(
uid, name="cov_btn", code=".btn {\n color: red;\n}\n",
language="css", repo="Widget", path="web", symbol=".btn",
project_id=pid,
)
return {"uid": uid, "pid": pid}
@pytest.mark.integration
async def test_coverage_measures_the_tree_exactly_and_caches(seeded):
from scribe.services.coverage import (
cached_coverage,
compute_coverage,
refresh_coverage,
)
uid, pid = seeded["uid"], seeded["pid"]
forge = _forge(_tarball(TREE))
coverage = await compute_coverage(uid, pid, forge=forge)
assert coverage is not None
assert coverage["total"] == 4
assert coverage["recorded"] == 2
assert coverage["estimate"] is True
assert coverage["repos"] == [{
"repo": "git.example.com/alice/widget", "ref": "main",
"total": 4, "recorded": 2,
}]
assert coverage["largest_gaps"] == [{"dir": "src", "uncovered": 2, "total": 3}]
# Nothing computed → nothing cached; refresh writes; the cache reads back
# byte-equal, because enter_project will serve exactly this.
assert await cached_coverage(uid, pid) is None
stored = await refresh_coverage(uid, pid, forge=forge)
assert (await cached_coverage(uid, pid)) == json.loads(json.dumps(stored))
@pytest.mark.integration
async def test_enter_project_surfaces_the_line_only_once_computed(seeded):
from scribe.mcp._context import _user_id_ctx
from scribe.mcp.tools.projects import enter_project
from scribe.services.coverage import refresh_coverage
uid, pid = seeded["uid"], seeded["pid"]
token = _user_id_ctx.set(uid)
try:
# Forge-less / never-computed instance: the key is present, null, and
# nothing else about the response changes.
before = await enter_project(project_id=pid)
assert before["pattern_coverage"] is None
await refresh_coverage(uid, pid, forge=_forge(_tarball(TREE)))
after = await enter_project(project_id=pid)
line = after["pattern_coverage"]
assert line.startswith(
"pattern-library coverage: 2/4 shapes recorded (estimate, computed "
)
assert line.endswith("; largest gaps: src")
finally:
_user_id_ctx.reset(token)
@pytest.mark.integration
async def test_unservable_binding_measures_nothing(seeded):
"""A project bound only to a host the forge doesn't serve returns None —
the same silence as no forge at all, never an error."""
from scribe.services.coverage import compute_coverage
from scribe.services.repo_bindings import set_binding
from scribe.models import async_session
from scribe.models.project import Project
uid = seeded["uid"]
async with async_session() as s:
other = Project(user_id=uid, title="Elsewhere")
s.add(other)
await s.flush()
other_pid = other.id
await s.commit()
await set_binding(uid, "https://github.com/somebody/else.git", other_pid)
assert await compute_coverage(uid, other_pid, forge=_forge(_tarball(TREE))) is None
+371
View File
@@ -0,0 +1,371 @@
"""Forge adapter contract tests (#2689) — the Gitea implementation against a
mocked transport, plus the configuration gate.
httpx.MockTransport is the fake forge: the adapter takes an injectable
transport precisely so the CONTRACT (URLs hit, auth header shape, payload
decoding, error taxonomy) is testable with no live server and no new
dependency. These are the reference behaviors step 8's GitHub adapter must
reproduce.
The most load-bearing tests are the OFF ones: an unconfigured instance must
get None from get_forge(), because every consumer treats None as "behave as if
the module didn't exist" (rule #115 — the baseline install has no forge).
"""
import base64
from unittest.mock import AsyncMock, patch
import httpx
import pytest
from scribe.services.forge import (
ForgeError,
ForgeNotFound,
GiteaForge,
get_forge,
)
BASE = "https://git.example.com"
def _forge(handler) -> GiteaForge:
return GiteaForge(BASE, "tok-123", transport=httpx.MockTransport(handler))
def _json(status: int, payload) -> httpx.Response:
return httpx.Response(status, json=payload)
# --- resolve_repo: the join between recorded repos and this forge ------------
@pytest.mark.parametrize(
("recorded", "expected"),
[
("https://git.example.com/alice/Widget.git", "alice/widget"),
("git@git.example.com:alice/widget.git", "alice/widget"),
("git.example.com/alice/widget", "alice/widget"),
# Nested (GitLab-style) groups survive as the API path remainder.
("https://git.example.com/team/sub/widget", "team/sub/widget"),
# Another host is a NORMAL miss, not an error.
("https://github.com/alice/widget", None),
("", None),
("not a url", None),
# Host alone, no owner/repo remainder.
("git.example.com", None),
],
)
def test_resolve_repo_matches_by_host_and_yields_the_api_path(recorded, expected):
forge = GiteaForge(BASE, "tok")
assert forge.resolve_repo(recorded) == expected
# --- read_file ---------------------------------------------------------------
async def test_read_file_decodes_content_and_carries_the_commit():
seen = {}
def handler(request: httpx.Request) -> httpx.Response:
seen["url"] = str(request.url)
seen["auth"] = request.headers.get("Authorization")
return _json(200, {
"type": "file",
"encoding": "base64",
"content": base64.b64encode("def x():\n return 1\n".encode()).decode(),
"sha": "blob" * 10,
"last_commit_sha": "c" * 40,
"path": "src/x.py",
})
got = await _forge(handler).read_file("alice/widget", "src/x.py", ref="dev")
assert got.content == "def x():\n return 1\n"
assert got.commit_sha == "c" * 40
assert got.path == "src/x.py"
assert "/api/v1/repos/alice/widget/contents/src/x.py" in seen["url"]
assert "ref=dev" in seen["url"]
assert seen["auth"] == "token tok-123"
async def test_read_file_404_is_not_found_and_a_directory_is_too():
with pytest.raises(ForgeNotFound):
await _forge(lambda r: _json(404, {"message": "no"})).read_file(
"alice/widget", "gone.py"
)
# The contents API returns a LIST for a directory — that's "no such file",
# not a decoding error.
with pytest.raises(ForgeNotFound):
await _forge(lambda r: _json(200, [{"type": "file"}])).read_file(
"alice/widget", "src"
)
async def test_read_file_auth_failure_names_the_scope_never_the_token():
with pytest.raises(ForgeError) as err:
await _forge(lambda r: _json(401, {})).read_file("alice/widget", "x.py")
assert "tok-123" not in str(err.value)
assert "scope" in str(err.value)
async def test_read_file_binary_content_is_a_forge_error():
def handler(request):
return _json(200, {
"type": "file", "encoding": "base64",
"content": base64.b64encode(b"\xff\xfe\x00\x01").decode(),
})
with pytest.raises(ForgeError):
await _forge(handler).read_file("alice/widget", "img.bin")
async def test_unreachable_forge_is_a_forge_error_not_a_crash():
def handler(request):
raise httpx.ConnectError("boom", request=request)
with pytest.raises(ForgeError):
await _forge(handler).read_file("alice/widget", "x.py")
# --- default_branch / check --------------------------------------------------
async def test_default_branch_reads_the_repo_record():
forge = _forge(lambda r: _json(200, {"default_branch": "dev"}))
assert await forge.default_branch("alice/widget") == "dev"
async def test_check_reports_version_and_identity():
def handler(request):
if request.url.path.endswith("/version"):
return _json(200, {"version": "1.23.1"})
return _json(200, {"login": "scribe-bot"})
result = await _forge(handler).check()
assert result == {"ok": True, "version": "1.23.1", "username": "scribe-bot"}
# --- the configuration gate --------------------------------------------------
def _settings(values: dict):
async def fake(key, default=""):
return values.get(key, default)
return patch("scribe.services.forge.get_admin_setting", AsyncMock(side_effect=fake))
async def test_unconfigured_instance_gets_none():
with _settings({}), patch("scribe.services.forge.Config") as cfg:
cfg.FORGE_KIND = cfg.FORGE_BASE_URL = cfg.FORGE_TOKEN = ""
assert await get_forge() is None
async def test_partial_config_is_still_off():
# A base URL with no token (or vice versa) must not half-enable anything.
for values in (
{"forge_kind": "gitea", "forge_base_url": BASE},
{"forge_kind": "gitea", "forge_token": "tok"},
{"forge_base_url": BASE, "forge_token": "tok"}, # no kind selected
):
with _settings(values), patch("scribe.services.forge.Config") as cfg:
cfg.FORGE_KIND = cfg.FORGE_BASE_URL = cfg.FORGE_TOKEN = ""
assert await get_forge() is None
async def test_unknown_kind_disables_with_a_warning_not_a_crash():
with _settings({
"forge_kind": "sourcehut", "forge_base_url": BASE, "forge_token": "tok",
}), patch("scribe.services.forge.Config") as cfg:
cfg.FORGE_KIND = cfg.FORGE_BASE_URL = cfg.FORGE_TOKEN = ""
assert await get_forge() is None
async def test_full_config_builds_a_gitea_adapter():
with _settings({
"forge_kind": "gitea",
"forge_base_url": BASE + "/", # trailing slash normalized away
"forge_token": "tok",
}), patch("scribe.services.forge.Config") as cfg:
cfg.FORGE_KIND = cfg.FORGE_BASE_URL = cfg.FORGE_TOKEN = ""
forge = await get_forge()
assert isinstance(forge, GiteaForge)
assert forge.base_url == BASE
assert forge.host == "git.example.com"
async def test_env_channel_fills_gaps_but_db_wins():
# Docker-secret deployments set FORGE_* env; an admin-UI value overrides.
with _settings({"forge_base_url": "https://db.example.com"}), \
patch("scribe.services.forge.Config") as cfg:
cfg.FORGE_KIND = "gitea"
cfg.FORGE_BASE_URL = "https://env.example.com"
cfg.FORGE_TOKEN = "env-tok"
forge = await get_forge()
assert isinstance(forge, GiteaForge)
assert forge.host == "db.example.com"
def test_forge_error_taxonomy_is_catchable_as_one_family():
assert issubclass(ForgeNotFound, ForgeError)
assert issubclass(ForgeError, RuntimeError)
def test_adapter_contract_surface():
"""Both adapters implement exactly this surface — the second
implementation is what proves it's a contract (#2693)."""
from scribe.services.forge import FORGE_KINDS, GitHubForge
for cls in (GiteaForge, GitHubForge):
for method in (
"read_file", "latest_commit", "archive",
"default_branch", "resolve_repo", "check",
):
assert callable(getattr(cls, method))
assert GiteaForge.kind == "gitea"
assert GitHubForge.kind == "github"
assert set(FORGE_KINDS) == {"gitea", "github"}
def test_admin_routes_registered():
from scribe.app import create_app
from scribe.routes import admin as admin_routes
for name in ("get_forge_settings", "update_forge_settings", "test_forge"):
assert callable(getattr(admin_routes, name))
rules = {r.rule for r in create_app().url_map.iter_rules()}
assert "/api/admin/forge" in rules
assert "/api/admin/forge/test" in rules
def test_settings_kv_surface_masks_the_forge_token():
"""The generic /api/settings dump masked nothing — the admin endpoints'
masking was bypassable by reading the raw KV rows (found while wiring the
forge token; smtp_password had the same exposure)."""
from scribe.routes.settings import _SECRET_KEYS, _masked
out = _masked({"forge_token": "tok-123", "smtp_password": "pw", "theme": "dark"})
assert out["forge_token"] == "********"
assert out["smtp_password"] == "********"
assert out["theme"] == "dark"
assert {"forge_token", "smtp_password"} <= set(_SECRET_KEYS)
# An unset secret stays empty rather than reading as a set-but-masked one.
assert _masked({"forge_token": ""})["forge_token"] == ""
def test_config_has_the_docker_secret_channel():
from scribe.config import Config
for attr in ("FORGE_KIND", "FORGE_BASE_URL", "FORGE_TOKEN"):
assert hasattr(Config, attr)
# --- the GitHub adapter (#2693) ----------------------------------------------
# Same contract, second implementation. Where behavior below differs from the
# Gitea tests above, that difference IS the adapter's job: API host mapping,
# Bearer auth, the missing last_commit_sha, the codeload redirect.
def _github(handler, base: str = "https://github.com"):
from scribe.services.forge import GitHubForge
return GitHubForge(base, "gh-tok", transport=httpx.MockTransport(handler))
def test_github_resolve_repo_is_the_same_host_join():
from scribe.services.forge import GitHubForge
forge = GitHubForge("https://github.com", "t")
assert forge.resolve_repo("git@github.com:alice/Widget.git") == "alice/widget"
# A Gitea-hosted repo is a NORMAL miss for a GitHub forge, and vice versa.
assert forge.resolve_repo("https://git.example.com/alice/widget") is None
async def test_github_api_base_maps_dot_com_and_enterprise():
seen = []
def handler(request):
seen.append(str(request.url))
return _json(200, {"default_branch": "main"})
await _github(handler).default_branch("alice/widget")
await _github(handler, base="https://ghe.example.com").default_branch("alice/widget")
assert seen[0] == "https://api.github.com/repos/alice/widget"
assert seen[1] == "https://ghe.example.com/api/v3/repos/alice/widget"
async def test_github_read_file_decodes_and_stamps_from_the_commits_call():
content = "def canonical():\n return 1\n"
def handler(request):
assert request.headers["Authorization"] == "Bearer gh-tok"
assert request.headers["X-GitHub-Api-Version"]
if request.url.path.endswith("/commits"):
assert request.url.params["path"] == "src/x.py"
assert request.url.params["per_page"] == "1"
return _json(200, [{"sha": "c" * 40}])
return _json(200, {
"type": "file", "encoding": "base64",
"content": base64.b64encode(content.encode()).decode(),
"path": "src/x.py", "sha": "blob-sha-not-a-point-in-history",
})
f = await _github(handler).read_file("alice/widget", "src/x.py")
assert f.content == content
# From /commits — GitHub's contents payload only carries the blob sha,
# which is a content address, not the provenance stamp.
assert f.commit_sha == "c" * 40
async def test_github_read_file_serves_content_even_when_the_stamp_fails():
content = "x = 1\n"
def handler(request):
if request.url.path.endswith("/commits"):
return httpx.Response(500)
return _json(200, {"type": "file", "encoding": "base64",
"content": base64.b64encode(content.encode()).decode()})
f = await _github(handler).read_file("alice/widget", "x.py")
assert f.content == content
assert f.commit_sha == "" # unknown stamp, not a failed read
async def test_github_archive_follows_the_codeload_redirect():
def handler(request):
if request.url.host == "api.github.com":
return httpx.Response(302, headers={
"Location": "https://codeload.github.com/alice/widget/tar.gz/main",
})
assert request.url.host == "codeload.github.com"
# httpx drops Authorization on the cross-host hop — codeload's URL
# carries its own grant, and leaking the PAT there would be a bug.
assert "Authorization" not in request.headers
return httpx.Response(200, content=b"tarball-bytes")
assert await _github(handler).archive("alice/widget", "main") == b"tarball-bytes"
async def test_github_check_probes_the_token_with_user():
result = await _github(lambda r: _json(200, {"login": "octo"})).check()
assert result["ok"] is True
assert result["username"] == "octo"
async def test_latest_commit_parses_tolerantly_on_both_adapters():
"""The one caller treats latest_commit as an optimization with a fallback,
so a surprising payload must read as "don't know", never raise."""
assert await _github(
lambda r: _json(200, [{"sha": "d" * 40}])
).latest_commit("a/w", "x.py") == "d" * 40
assert await _github(
lambda r: _json(200, {"weird": True})
).latest_commit("a/w", "x.py") == ""
assert await _forge(
lambda r: _json(200, [{"sha": "e" * 40}])
).latest_commit("a/w", "x.py") == "e" * 40
assert await _forge(lambda r: _json(200, [])).latest_commit("a/w", "x.py") == ""
async def test_full_config_builds_a_github_adapter():
from scribe.services.forge import GitHubForge
with _settings({
"forge_kind": "github",
"forge_base_url": "https://github.com",
"forge_token": "tok",
}), patch("scribe.services.forge.Config") as cfg:
cfg.FORGE_KIND = cfg.FORGE_BASE_URL = cfg.FORGE_TOKEN = ""
forge = await get_forge()
assert isinstance(forge, GitHubForge)
+32
View File
@@ -170,6 +170,38 @@ def test_python_dialect_on_a_row_with_no_data_at_all():
assert knowledge_svc.verification_matches(None, "ok") is False
@pytest.mark.parametrize(
"value, expected",
[("ok", False), ("attention", True), ("drifted", False), ("unverified", False)],
)
def test_python_dialect_on_a_push_invalidated_ok_verdict(value, expected):
"""#2691: a push touched the recorded location since the verdict. Like the
expired case, it is neither drifted (nothing found wrong) nor unverified (a
check happened) — but the repo moved under the blessing, so `attention`
must include it and `ok` must not."""
data = _data("ok", "aaa", "aaa")
data["verification"]["invalidated_by"] = {"commit_sha": "d" * 40, "at": "t"}
assert knowledge_svc.verification_matches(data, value) is expected
def test_push_invalidation_reads_as_attention_and_clears_on_reverify():
"""The flag rides the verdict dict, so recording ANY fresh verdict clears
it by construction — compose_verification builds a new dict. No clearing
branch exists to forget."""
sha = code_sha("def f(): pass")
verdict = compose_verification(status="ok", checked_code_sha=sha)
verdict["invalidated_by"] = {"commit_sha": "d" * 40, "at": "t"}
fields = {"code": "def f(): pass", "verification": verdict}
view = verification_view(_note(), fields)
assert view["needs_attention"] is True
assert view["invalidated_by"]["commit_sha"] == "d" * 40
fresh = compose_verification(status="ok", checked_code_sha=sha)
assert "invalidated_by" not in fresh
view2 = verification_view(_note(), {"code": "def f(): pass", "verification": fresh})
assert view2["needs_attention"] is False
# --- the filter, SQL dialect ----------------------------------------------
+267
View File
@@ -0,0 +1,267 @@
"""Pull-time freshness (#2690) — attach_live_body against a mocked forge.
The properties that must hold, each with its own way of rotting:
- A no-forge instance's pull response is BYTE-IDENTICAL to today's (rule
#115 — the baseline, not a degraded mode).
- The probe never rewrites the body: "current" refreshes provenance,
"diverged" reports, and neither clobbers the record mid-read.
- A 404 stamps the mechanically-true 'missing' verdict — once, not on
every pull of an already-flagged record.
- The pull is never slower than the budget: a hung forge costs bounded
time and then the cache serves.
"""
import asyncio
import base64
import time
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import httpx
from scribe.services import background
from scribe.services import snippets as svc
from scribe.services.forge import GiteaForge
BASE = "https://git.example.com"
CODE = "def helper(x):\n return x + 1\n"
SHA = "f" * 40
def _note():
return SimpleNamespace(id=7, user_id=3, title="", body="", tags=[], data=None)
def _data(*, repo=f"{BASE}/alice/widget", code=CODE, verification=None, provenance=None):
snippet = {
"code": code,
"locations": [{"repo": repo, "path": "src/helper.py", "symbol": "helper"}],
}
if verification:
snippet["verification"] = verification
if provenance:
snippet["provenance"] = provenance
return {"snippet": snippet}
def _forge_with(handler) -> GiteaForge:
return GiteaForge(BASE, "tok", transport=httpx.MockTransport(handler))
def _file_response(content: str, commit_sha: str = SHA) -> httpx.Response:
return httpx.Response(200, json={
"type": "file", "encoding": "base64",
"content": base64.b64encode(content.encode()).decode(),
"last_commit_sha": commit_sha, "path": "src/helper.py",
})
def _patched(forge):
return patch("scribe.services.forge.get_forge", AsyncMock(return_value=forge))
async def test_no_forge_attaches_nothing():
data = _data()
before = repr(data)
with _patched(None):
await svc.attach_live_body(_note(), data)
assert repr(data) == before
assert "body_source" not in data
async def test_current_code_confirms_and_refreshes_provenance():
# The file wraps the cached code with extra context and trailing spaces —
# containment is judged after the same normalization the verdict hash uses.
file_content = "import os\n\n" + CODE.replace(" + 1", " + 1 ").rstrip() + "\n\n# eof\n"
forge = _forge_with(lambda r: _file_response(file_content))
saved = {}
async def fake_update(uid, nid, **fields):
saved.update(fields)
data = _data()
with _patched(forge), patch.object(svc.notes_svc, "update_note", fake_update):
await svc.attach_live_body(_note(), data)
await background.drain()
assert data["body_source"] == "forge"
assert data["body_freshness"] == "current"
# Reflected in the response...
assert data["snippet"]["provenance"]["commit_sha"] == SHA
# ...and persisted, without touching the body.
assert saved["data"]["provenance"]["commit_sha"] == SHA
assert "body" not in saved
async def test_current_with_same_stored_sha_skips_the_write():
forge = _forge_with(lambda r: _file_response("prefix\n" + CODE))
update = AsyncMock()
data = _data(provenance={"commit_sha": SHA, "fetched_at": "t"})
with _patched(forge), patch.object(svc.notes_svc, "update_note", update):
await svc.attach_live_body(_note(), data)
await background.drain()
assert data["body_freshness"] == "current"
update.assert_not_called()
async def test_stored_sha_short_circuit_skips_the_content_fetch():
"""#2693: when provenance already names a commit and the forge reports no
newer commit touching the path, the pull is confirmed current WITHOUT a
content transfer — the economy that fits pull-time freshness inside
GitHub's rate limits."""
calls = []
def handler(request):
calls.append(request.url.path)
if request.url.path.endswith("/commits"):
return httpx.Response(200, json=[{"sha": SHA}])
raise AssertionError("the content fetch should have been skipped")
update = AsyncMock()
data = _data(provenance={"commit_sha": SHA, "fetched_at": "t"})
with _patched(_forge_with(handler)), patch.object(svc.notes_svc, "update_note", update):
await svc.attach_live_body(_note(), data)
await background.drain()
assert data["body_source"] == "forge"
assert data["body_freshness"] == "current"
assert len(calls) == 1
update.assert_not_called() # same stamp — nothing to persist
async def test_moved_file_falls_through_to_the_full_fetch():
new_sha = "0" * 40
def handler(request):
if request.url.path.endswith("/commits"):
return httpx.Response(200, json=[{"sha": new_sha}])
return _file_response("prefix\n" + CODE, commit_sha=new_sha)
saved = {}
async def fake_update(uid, nid, **fields):
saved.update(fields)
data = _data(provenance={"commit_sha": SHA, "fetched_at": "t"})
with _patched(_forge_with(handler)), patch.object(
svc.notes_svc, "update_note", fake_update
):
await svc.attach_live_body(_note(), data)
await background.drain()
# The file moved but still contains the code — current, with the stamp
# advanced by the authoritative full fetch.
assert data["body_freshness"] == "current"
assert saved["data"]["provenance"]["commit_sha"] == new_sha
async def test_short_circuit_failure_degrades_to_the_full_fetch():
"""A forge whose commits endpoint errors must cost nothing: the full
fetch stays the authoritative path and the pull behaves as before."""
def handler(request):
if request.url.path.endswith("/commits"):
return httpx.Response(500)
return _file_response("prefix\n" + CODE)
update = AsyncMock()
data = _data(provenance={"commit_sha": SHA, "fetched_at": "t"})
with _patched(_forge_with(handler)), patch.object(svc.notes_svc, "update_note", update):
await svc.attach_live_body(_note(), data)
await background.drain()
assert data["body_freshness"] == "current"
update.assert_not_called() # same sha via the full fetch → same-sha skip
async def test_diverged_reports_without_clobbering():
forge = _forge_with(lambda r: _file_response("def helper(x):\n return x - 1\n"))
update = AsyncMock()
data = _data()
with _patched(forge), patch.object(svc.notes_svc, "update_note", update):
await svc.attach_live_body(_note(), data)
await background.drain()
assert data["body_source"] == "cache"
assert data["body_freshness"] == "diverged"
assert data["snippet"]["code"] == CODE
update.assert_not_called()
async def test_missing_stamps_the_verdict_once():
forge = _forge_with(lambda r: httpx.Response(404, json={}))
data = _data()
with _patched(forge), patch.object(
svc, "record_verification", AsyncMock()
) as verdict:
await svc.attach_live_body(_note(), data)
await background.drain()
assert data["body_freshness"] == "missing"
verdict.assert_awaited_once()
assert verdict.await_args.kwargs["status"] == svc.VERIFY_MISSING
# Already stamped missing → no re-stamp on the next pull.
data2 = _data(verification={"status": svc.VERIFY_MISSING, "code_sha": "x"})
with _patched(forge), patch.object(
svc, "record_verification", AsyncMock()
) as verdict2:
await svc.attach_live_body(_note(), data2)
await background.drain()
assert data2["body_freshness"] == "missing"
verdict2.assert_not_awaited()
async def test_unreachable_falls_back_to_cache():
def handler(request):
raise httpx.ConnectError("down", request=request)
data = _data()
with _patched(_forge_with(handler)):
await svc.attach_live_body(_note(), data)
assert data["body_source"] == "cache"
assert data["body_freshness"] == "unreachable"
async def test_hung_forge_costs_bounded_time(monkeypatch):
async def slow_handler(request):
await asyncio.sleep(30)
return _file_response(CODE)
monkeypatch.setattr(svc, "PULL_FETCH_BUDGET_S", 0.2)
data = _data()
start = time.monotonic()
with _patched(_forge_with(slow_handler)):
await svc.attach_live_body(_note(), data)
assert time.monotonic() - start < 2.0
assert data["body_freshness"] == "unreachable"
async def test_foreign_repo_and_placeless_records_read_as_cache():
forge = GiteaForge(BASE, "tok")
data = _data(repo="https://github.com/alice/widget")
with _patched(forge):
await svc.attach_live_body(_note(), data)
assert data["body_freshness"] == "repo-not-on-this-forge"
placeless = {"snippet": {"code": CODE, "locations": []}}
with _patched(forge):
await svc.attach_live_body(_note(), placeless)
assert placeless["body_freshness"] == "no-recorded-location"
def test_both_pull_surfaces_attach_freshness():
"""Source-inspection guard (the CI convention for wiring assertions): the
MCP pull and the REST detail route both decorate — a surface that forgets
is a surface whose readers silently lose freshness."""
import pathlib
root = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe"
mcp_src = (root / "mcp" / "tools" / "snippets.py").read_text()
rest_src = (root / "routes" / "snippets.py").read_text()
assert "attach_live_body" in mcp_src
assert "attach_live_body" in rest_src
async def test_forge_failure_inside_lookup_never_breaks_the_pull():
with patch(
"scribe.services.forge.get_forge", AsyncMock(side_effect=RuntimeError("cfg"))
):
data = _data()
await svc.attach_live_body(_note(), data)
assert "body_source" not in data
+186
View File
@@ -0,0 +1,186 @@
"""Body provenance — the cache-with-provenance half of the pointer model (#2688).
Decision #2686: the recorded location is the source of truth for a snippet's
code and the stored body is a CACHE of it. `data.provenance` records what that
cache is a cache OF — the commit the body was read at, and when — so staleness
becomes judgeable instead of guessed, and so the forge fetch (milestone 288
step 5) has something to refresh.
The rules under test, because each has a way to rot silently:
- Provenance follows the CODE. An edit that changes the code without a fresh
SHA must DROP the stamp — carrying it would claim the new body came from
the old commit, which is worse than not knowing.
- Writes that are ABOUT the code rather than changes TO it (a verification
verdict, a metadata edit) must CARRY it — record_verification rebuilds
`data` from scratch, so forgetting the field there erases it invisibly.
- An "ok" verdict at a known commit RESTAMPS it: the checker just proved the
cached body matches the source there.
Unit tests cover the compose/carry logic; the integration section runs the
same rules through the real service paths on real Postgres (the #2663 lesson —
a DB-touching path with only mocked coverage is a path with no coverage).
"""
import pytest
import pytest_asyncio
from scribe.services.snippets import (
VERIFY_CHANGED,
VERIFY_OK,
compose_data,
compose_provenance,
compose_verification,
snippet_fields,
verification_view,
)
SHA_A = "a" * 40
SHA_B = "b" * 40
# --- unit: composing ---------------------------------------------------------
def test_compose_provenance_stamps_sha_and_time():
prov = compose_provenance(commit_sha=f" {SHA_A} ")
assert prov["commit_sha"] == SHA_A
assert prov["fetched_at"] # ISO stamp, defaulted
def test_compose_provenance_without_a_sha_is_none_not_an_empty_record():
# Absent provenance must stay ABSENT (the pre-#2688 semantics), never an
# empty dict that readers would have to distinguish from a real one.
assert compose_provenance(commit_sha="") is None
assert compose_provenance(commit_sha=" ") is None
def test_compose_data_carries_provenance_only_when_present():
with_it = compose_data(name="x", provenance={"commit_sha": SHA_A, "fetched_at": "t"})
without = compose_data(name="x", provenance=None)
assert with_it["provenance"]["commit_sha"] == SHA_A
assert "provenance" not in without
def test_verification_records_and_reads_back_the_checked_commit():
verdict = compose_verification(
status=VERIFY_OK, checked_code_sha="c" * 32, commit_sha=SHA_A,
)
assert verdict["commit_sha"] == SHA_A
# And an empty one is omitted, not stored as "".
bare = compose_verification(status=VERIFY_OK, checked_code_sha="c" * 32)
assert "commit_sha" not in bare
class _N: # minimal note stand-in for the read-time view
data = None
fields = {"code": "", "verification": verdict}
view = verification_view(_N(), fields)
assert view["commit_sha"] == SHA_A
# --- integration: the rules through the real service paths -------------------
@pytest_asyncio.fixture
async def _dispose_engine():
from scribe.models import engine
yield
await engine.dispose()
@pytest_asyncio.fixture
async def user_id(_dispose_engine):
# Get-or-create: the lane's database persists across tests, so a second
# test re-creating the same username dies on the unique constraint.
from sqlalchemy import select
from scribe.models import async_session
from scribe.models.user import User
async with async_session() as s:
existing = (
await s.execute(
select(User).where(User.username == "snippet_prov_itest")
)
).scalar_one_or_none()
if existing is not None:
return existing.id
user = User(username="snippet_prov_itest")
s.add(user)
await s.flush()
uid = user.id
await s.commit()
return uid
async def _fresh(uid, note_id):
from scribe.services import snippets as svc
note = await svc.get_snippet(uid, note_id)
return snippet_fields(note), svc.snippet_to_dict(note)
@pytest.mark.integration
async def test_provenance_lives_and_dies_with_the_code_end_to_end(user_id):
from scribe.services import snippets as svc
note = await svc.create_snippet(
user_id, name="prov_helper", code="def prov_helper():\n return 1\n",
language="python", repo="Scribe", path="src/x.py", symbol="prov_helper",
commit_sha=SHA_A,
)
fields, view = await _fresh(user_id, note.id)
assert fields["provenance"]["commit_sha"] == SHA_A
assert view["snippet"]["provenance"]["commit_sha"] == SHA_A
# A metadata edit leaves the code alone → carried.
await svc.update_snippet(user_id, note.id, when_to_use="when proving")
fields, _ = await _fresh(user_id, note.id)
assert fields["provenance"]["commit_sha"] == SHA_A
# A code edit with a fresh SHA → restamped.
await svc.update_snippet(
user_id, note.id, code="def prov_helper():\n return 2\n",
commit_sha=SHA_B,
)
fields, _ = await _fresh(user_id, note.id)
assert fields["provenance"]["commit_sha"] == SHA_B
# A code edit WITHOUT one → dropped, not carried: the new body does not
# come from SHA_B and the record must not claim it does.
await svc.update_snippet(
user_id, note.id, code="def prov_helper():\n return 3\n",
)
fields, view = await _fresh(user_id, note.id)
assert "provenance" not in fields
assert "provenance" not in view["snippet"]
@pytest.mark.integration
async def test_verification_stamps_the_commit_and_ok_refreshes_provenance(user_id):
from scribe.services import snippets as svc
note = await svc.create_snippet(
user_id, name="prov_verify", code="def prov_verify():\n return 1\n",
language="python", repo="Scribe", path="src/y.py", symbol="prov_verify",
commit_sha=SHA_A,
)
# A non-ok verdict at a newer commit records where the check ran but must
# CARRY provenance — the check didn't change what the cached body is.
await svc.record_verification(
user_id, note.id, status=VERIFY_CHANGED, detail="diverged", commit_sha=SHA_B,
)
fields, view = await _fresh(user_id, note.id)
assert view["verification"]["commit_sha"] == SHA_B
assert fields["provenance"]["commit_sha"] == SHA_A
# An OK verdict at that commit proves the cache matches the source there —
# provenance refreshes without an edit.
await svc.record_verification(
user_id, note.id, status=VERIFY_OK, detail="matches", commit_sha=SHA_B,
)
fields, view = await _fresh(user_id, note.id)
assert view["verification"]["commit_sha"] == SHA_B
assert fields["provenance"]["commit_sha"] == SHA_B
# And the verdict itself survives untouched by the restamp.
assert view["verification"]["status"] == VERIFY_OK
assert view["verification"]["current"] is True
+49
View File
@@ -861,3 +861,52 @@ def test_hook_stays_quiet_about_recording_when_nothing_is_duplicated(tmp_path):
)
assert out.returncode == 0
assert "create_snippet" not in out.stdout
@pytest.mark.parametrize(
("fname", "definition"),
[
("scanner.go", "func Resolve(x int) error {\n\treturn nil\n}\n"),
("scanner_m.go",
"func (s *Scanner) Resolve(x int) error {\n\treturn nil\n}\n"),
("queue.kt", "suspend fun refreshQueue(id: Long) {\n}\n"),
("fetch.rs", "pub async fn fetch_all() -> u32 {\n 0\n}\n"),
("adapter.go", "type ForgeAdapter struct {\n\tname string\n}\n"),
],
ids=["go-func", "go-method", "kotlin-fun", "rust-fn", "go-type"],
)
def test_local_arm_finds_duplicates_in_every_language_family(
tmp_path, fname, definition
):
"""#2682: the definition detector must cover ALL code, not the languages of
the repo it was born in. Its original CSS/JS/Python-only patterns silently
amputated the local arm — and the #2664 recording nudge gated on it — for
every Go/Kotlin/Rust project, which is exactly where the operator observed
recording never happening. Each case stages an existing copy and writes the
same definition to a second file; the hook must prove the duplication and
ask for the record."""
env = _hook_runtime_env()
repo = tmp_path / "repo"
repo.mkdir()
subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
(repo / fname).write_text(definition)
subprocess.run(["git", "add", "."], cwd=repo, check=True, env=env)
ext = fname.rsplit(".", 1)[1]
out = subprocess.run(
["bash", str(HOOK)],
input=json.dumps({
"session_id": f"s-lang-{ext}", "cwd": str(repo),
"tool_name": "Write",
"tool_input": {"file_path": str(repo / f"copy.{ext}"),
"content": definition},
}),
capture_output=True, text=True, env=env,
)
assert out.returncode == 0
assert out.stdout.strip(), (
f"hook produced no output for {fname} — the local arm should have "
f"found the staged duplicate definition"
)
ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
assert "already defined" in ctx
assert "create_snippet" in ctx