petal-power/second-brain
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: petal-power:22f4a0915f000d0ee9e6f3626686fa3994997d53
Branches Tags
petal-power:main
petal-power:claude/config-local-overlay
petal-power:claude/source-edit-and-retry
petal-power:claude/laughing-shirley-52f4a6
..
compare: petal-power:93f71ebb14b90723fea9c0cacc66db805d601150
Branches Tags
petal-power:main
petal-power:claude/config-local-overlay
petal-power:claude/source-edit-and-retry
petal-power:claude/laughing-shirley-52f4a6

No commits in common. "22f4a0915f000d0ee9e6f3626686fa3994997d53" and "93f71ebb14b90723fea9c0cacc66db805d601150" have entirely different histories.
22f4a0915f ... 93f71ebb14

6 changed files with 30 additions and 654 deletions
Show Stats Expand all files Collapse all files

37
reviews/a-review-2026-05-25.md
View File

@ -1,37 +0,0 @@
# a-review — 2026-05-25
**Model:** gemini
**Diff base:** `HEAD~1` (5 files changed, 615 insertions(+), 30 deletions(-))
**Session reference:** none
**Context:** CLAUDE.md + session notes + expanded diff
---
Excellent work. The changes are well-structured, follow existing project conventions, and include a comprehensive test suite. This is a high-quality contribution. I have only one minor finding and one informational note.
### [WARNING] Logic Error
* **File:** `src/second_brain/web/templates/source_detail.html`
* **Location:** Line 151
* **Issue:** The "Re-extract only" button is conditionally rendered with `{% if source.has_transcript and source.source_type == 'video' %}`. This is too restrictive. The backend service function `reset_source_for_retry` correctly allows an "extract" retry for any source that has content in `transcript_text`, which includes processed articles. Limiting this feature to only videos in the UI prevents a valid and useful action for non-video sources.
* **Suggestion:** Change the conditional to `{% if source.has_transcript %}`. This will correctly show the button for any source (article or video) that has been pulled and has content ready for extraction, which matches the backend's capability.
```html
<!-- Change this -->
{% if source.has_transcript and source.source_type == 'video' %}
<!-- To this -->
{% if source.has_transcript %}
```
### [INFO] Security
* **Files:** `src/second_brain/web/app.py`, `src/second_brain/web/templates/source_detail.html`
* **Observation:** The changes introduce new endpoints (`/edit`, `/retry`) that modify database state. According to `CLAUDE.md`, the application has "No auth" and relies on being deployed in a trusted, LAN-only environment. These new endpoints increase the number of ways an unauthorized user on the network could alter data.
* **Context:** This is not a new vulnerability, but an expansion of the existing trust model. As long as the application remains on a trusted network, this is acceptable per the project's documented security posture. It's worth keeping in mind that no CSRF protection is being used for these HTMX `POST` actions.
### Summary
The overall code quality is very high. The changes are thoughtfully implemented, with a clean separation between the CLI, the web layer, and a shared service layer. The refactoring in `web/app.py` to create `_render_source_detail` is a good move for maintainability. The addition of a thorough, live-DB test suite for the new service functions is commendable and demonstrates a commitment to quality.
Once the minor template logic issue is resolved, this change will be in perfect shape.

42
src/second_brain/main.py
View File

@ -4,7 +4,6 @@ second-brain CLI entry point.
Commands:
add — queue a source URL (video or article)
process — run the full pipeline on pending/pulled/transcribed sources
retry — reset a source so the pipeline re-processes it
serve — launch the FastAPI web UI
compile — run the wiki compiler on accepted sources
schedule — start the overnight scheduler
@ -308,47 +307,6 @@ def serve(host: str, port: int, reload: bool) -> None:
)
# ---------------------------------------------------------------------------
# retry
# ---------------------------------------------------------------------------
@cli.command()
@click.argument("source_id", type=int)
@click.option(
"--mode",
type=click.Choice(["full", "extract"]),
default="full",
show_default=True,
help="`full` resets to PENDING (re-pull + re-process); "
"`extract` resets to TRANSCRIBED (requires existing transcript).",
)
def retry(source_id: int, mode: str) -> None:
"""Reset a source so the pipeline re-processes it.
Clears the claim pair and error_message and rewinds status to the
appropriate stage. Parity with the dashboard re-run buttons.
"""
from second_brain.sources_service import reset_source_for_retry
db = get_database()
try:
with db.session() as sess:
source = reset_source_for_retry(
sess, source_id=source_id, mode=mode
)
new_status = source.status.value
url = source.url
except LookupError as exc:
click.echo(f"[retry] {exc}", err=True)
sys.exit(2)
except ValueError as exc:
click.echo(f"[retry] {exc}", err=True)
sys.exit(2)
click.echo(f"[retry] id={source_id} status → {new_status} ({url})")
# ---------------------------------------------------------------------------
# compile
# ---------------------------------------------------------------------------

112
src/second_brain/sources_service.py
View File

@ -275,126 +275,14 @@ def add_playlist(
return result
# ---------------------------------------------------------------------------
# Edit + retry helpers (shared by web routes and CLI)
# ---------------------------------------------------------------------------
# Fields the edit-mode form is allowed to mutate. URL is intentionally
# excluded — it's the UNIQUE dedupe key and editing it risks collisions
# plus breaks the embeddings 4-tuple identity (source_id is stable but
# semantic identity drifts).
EDITABLE_METADATA_FIELDS = ("domain", "title", "focus")
def update_source_metadata(
sess: Session,
*,
source_id: int,
domain: Optional[str] = None,
title: Optional[str] = None,
focus: Optional[str] = None,
) -> Source:
"""Update editable metadata on a source row.
Empty strings collapse to NULL for title/focus (matches `add_source`
semantics). Domain is required to stay in the configured set —
bumping the constraint is a separate decision.
Raises ValueError on validation failures and LookupError if the row
is missing. Caller owns the transaction boundary.
"""
source = sess.query(Source).filter(Source.id == source_id).first()
if source is None:
raise LookupError(f"source id={source_id} not found")
if domain is not None:
if domain not in DOMAINS:
raise ValueError(
f"unknown domain {domain!r}; valid: {', '.join(DOMAINS)}"
)
source.domain = domain
if title is not None:
source.title = title.strip() or None
if focus is not None:
source.focus = focus.strip() or None
source.updated_at = utcnow()
sess.flush()
return source
# Retry modes — keep the surface small. Anything else is a separate ask.
RETRY_FULL = "full"
RETRY_EXTRACT = "extract"
RETRY_MODES = (RETRY_FULL, RETRY_EXTRACT)
def reset_source_for_retry(
sess: Session,
*,
source_id: int,
mode: str = RETRY_FULL,
) -> Source:
"""Reset a source so the pipeline re-processes it.
Modes:
- "full": status → PENDING. Article pull or video download +
transcribe will happen again on the appropriate worker.
- "extract": status → TRANSCRIBED. Requires the row to already
carry a transcript (transcript_text). The dev scheduler will
re-run extraction without re-pulling the source. Cheap.
Always clears `claimed_by` / `claimed_at` / `error_message` so the
queue dance can pick it up cleanly and stale errors don't bleed
into the UI.
Raises ValueError on bad mode / missing transcript for extract mode,
and LookupError if the row is missing.
"""
if mode not in RETRY_MODES:
raise ValueError(
f"unknown retry mode {mode!r}; valid: {', '.join(RETRY_MODES)}"
)
source = sess.query(Source).filter(Source.id == source_id).first()
if source is None:
raise LookupError(f"source id={source_id} not found")
if mode == RETRY_EXTRACT:
if not (source.transcript_text and source.transcript_text.strip()):
raise ValueError(
"extract-only retry requires an existing transcript; "
"use full retry instead"
)
source.status = SourceStatus.TRANSCRIBED
else:
source.status = SourceStatus.PENDING
source.claimed_by = None
source.claimed_at = None
source.error_message = None
source.updated_at = utcnow()
sess.flush()
return source
__all__ = [
"AddResult",
"DEFAULT_PLAYLIST_MAX_ITEMS",
"EDITABLE_METADATA_FIELDS",
"PlaylistAddResult",
"RETRY_EXTRACT",
"RETRY_FULL",
"RETRY_MODES",
"add_playlist",
"add_source",
"expand_youtube_playlist",
"is_article_url",
"is_youtube_playlist_url",
"reset_source_for_retry",
"update_source_metadata",
"validate_url",
]

135
src/second_brain/web/app.py
View File

@ -31,14 +31,9 @@ from second_brain.settings_store import (
)
from second_brain.sources_service import (
DEFAULT_PLAYLIST_MAX_ITEMS,
RETRY_EXTRACT,
RETRY_FULL,
RETRY_MODES,
add_playlist,
add_source,
is_youtube_playlist_url,
reset_source_for_retry,
update_source_metadata,
)
# ---------------------------------------------------------------------------
@ -125,33 +120,16 @@ async def index(
)
def _render_source_detail(
request: Request,
source_id: int,
*,
edit: bool = False,
flash: Optional[str] = None,
error: Optional[str] = None,
form_overrides: Optional[dict] = None,
) -> HTMLResponse:
"""Single render path for /sources/{id} — used by GET + every HTMX
mutation endpoint so each action gets a consistent re-render.
`form_overrides` lets a failed edit-save bounce show the user's
in-progress values instead of the row's stored values.
"""
@app.get("/sources/{source_id}", response_class=HTMLResponse)
async def source_detail(request: Request, source_id: int):
"""Source detail page with extraction results and accept/reject actions."""
db = get_database()
with db.session() as sess:
source = sess.query(Source).filter(Source.id == source_id).first()
if not source:
raise HTTPException(status_code=404, detail="Source not found")
source_data = _source_to_dict(source)
extraction_data = (
_extraction_to_dict(source.extraction) if source.extraction else None
)
if form_overrides:
source_data = {**source_data, **form_overrides}
extraction_data = _extraction_to_dict(source.extraction) if source.extraction else None
return templates.TemplateResponse(
request,
@ -159,25 +137,10 @@ def _render_source_detail(
{
"source": source_data,
"extraction": extraction_data,
"edit_mode": edit,
"flash": flash,
"error": error,
"domains": DOMAINS,
},
)
@app.get("/sources/{source_id}", response_class=HTMLResponse)
async def source_detail(request: Request, source_id: int, edit: bool = False):
"""Source detail page with extraction results and accept/reject actions.
`?edit=1` opens the metadata editor inline. Editing exposes domain,
title, and focus — URL stays read-only since it's the UNIQUE dedupe
key for the queue and the embeddings identity tuple.
"""
return _render_source_detail(request, source_id, edit=edit)
_IN_FLIGHT_STATUSES = [
SourceStatus.PENDING,
SourceStatus.PULLED,
@ -229,11 +192,17 @@ async def accept_source(request: Request, source_id: int):
if not source:
raise HTTPException(status_code=404, detail="Source not found")
source.status = SourceStatus.ACCEPTED
source_data = _source_to_dict(source)
extraction_data = _extraction_to_dict(source.extraction) if source.extraction else None
return _render_source_detail(
return templates.TemplateResponse(
request,
source_id,
flash="Accepted — will be included in next wiki compile.",
"source_detail.html",
{
"source": source_data,
"extraction": extraction_data,
"flash": "Accepted — will be included in next wiki compile.",
},
)
@ -246,80 +215,19 @@ async def reject_source(request: Request, source_id: int):
if not source:
raise HTTPException(status_code=404, detail="Source not found")
source.status = SourceStatus.FAILED
source_data = _source_to_dict(source)
extraction_data = _extraction_to_dict(source.extraction) if source.extraction else None
return _render_source_detail(request, source_id, flash="Rejected.")
@app.post("/sources/{source_id}/edit", response_class=HTMLResponse)
async def edit_source_metadata(
request: Request,
source_id: int,
domain: str = Form(...),
title: Optional[str] = Form(None),
focus: Optional[str] = Form(None),
):
"""HTMX edit-mode save. Updates domain/title/focus only — URL is
intentionally not editable (see service-layer comment)."""
db = get_database()
try:
with db.session() as sess:
update_source_metadata(
sess,
source_id=source_id,
domain=domain,
title=title,
focus=focus,
)
except LookupError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except ValueError as exc:
# Bounce back to edit mode with the user's in-progress values so
# they don't have to retype them.
return _render_source_detail(
return templates.TemplateResponse(
request,
source_id,
edit=True,
error=str(exc),
form_overrides={
"domain": domain,
"title": title or "",
"focus": focus or "",
"source_detail.html",
{
"source": source_data,
"extraction": extraction_data,
"flash": "Rejected.",
},
)
return _render_source_detail(request, source_id, flash="Saved.")
@app.post("/sources/{source_id}/retry", response_class=HTMLResponse)
async def retry_source(
request: Request,
source_id: int,
mode: str = Form(RETRY_FULL),
):
"""HTMX retry/re-run. `mode=full` → PENDING (re-pull + re-process).
`mode=extract` → TRANSCRIBED (re-extract only; requires an existing
transcript). Always clears the claim pair and error_message."""
if mode not in RETRY_MODES:
raise HTTPException(status_code=400, detail=f"unknown mode {mode!r}")
db = get_database()
try:
with db.session() as sess:
source = reset_source_for_retry(
sess, source_id=source_id, mode=mode
)
new_status = source.status.value
except LookupError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except ValueError as exc:
return _render_source_detail(request, source_id, error=str(exc))
if mode == RETRY_EXTRACT:
flash = f"Re-queued for extraction (status → {new_status})."
else:
flash = f"Re-queued from scratch (status → {new_status})."
return _render_source_detail(request, source_id, flash=flash)
# ---------------------------------------------------------------------------
# JSON API endpoints
@ -378,7 +286,6 @@ def _source_to_dict(s: Source) -> dict:
"focus": s.focus,
"ingested_at": s.ingested_at.strftime("%Y-%m-%d %H:%M") if s.ingested_at else "",
"has_extraction": s.extraction is not None,
"has_transcript": bool((s.transcript_text or "").strip()),
"error_message": s.error_message,
}

88
src/second_brain/web/templates/source_detail.html
View File

@ -11,74 +11,17 @@
</svg>
Back to sources
</a>
<div class="flex items-start justify-between gap-4">
<div class="min-w-0">
<h2 class="text-2xl font-semibold text-gray-900 break-words">{{ source.title }}</h2>
<h2 class="text-2xl font-semibold text-gray-900">{{ source.title }}</h2>
<a href="{{ source.url }}" target="_blank"
class="text-sm text-indigo-500 hover:text-indigo-700 break-all">{{ source.url }}</a>
</div>
{% if not edit_mode %}
<a href="/sources/{{ source.id }}?edit=1"
class="shrink-0 px-3 py-1.5 text-sm border border-gray-300 text-gray-700 rounded hover:bg-gray-50 transition-colors">
Edit
</a>
{% endif %}
</div>
</div>
{% if flash %}
<div class="mb-4 px-4 py-3 bg-green-50 border border-green-200 text-green-800 rounded text-sm">
{{ flash }}
</div>
{% endif %}
{% if error %}
<div class="mb-4 px-4 py-3 bg-red-50 border border-red-200 text-red-800 rounded text-sm">
{{ error }}
</div>
{% endif %}
{% if edit_mode %}
<!-- Edit mode: domain / title / focus. URL is intentionally read-only -->
<form hx-post="/sources/{{ source.id }}/edit"
hx-target="body"
hx-swap="innerHTML"
class="mb-8 bg-white rounded-lg border border-gray-200 p-5 space-y-4">
<div>
<label class="block text-xs font-semibold text-gray-500 uppercase tracking-wide mb-1">URL <span class="font-normal normal-case text-gray-400">(read-only — dedupe key)</span></label>
<input type="text" value="{{ source.url }}" disabled
class="w-full px-3 py-2 border border-gray-200 bg-gray-50 text-gray-500 text-sm rounded">
</div>
<div>
<label for="edit-title" class="block text-xs font-semibold text-gray-500 uppercase tracking-wide mb-1">Title</label>
<input type="text" id="edit-title" name="title" value="{{ source.title or '' }}"
class="w-full px-3 py-2 border border-gray-300 text-sm rounded focus:outline-none focus:ring-2 focus:ring-indigo-300">
</div>
<div>
<label for="edit-domain" class="block text-xs font-semibold text-gray-500 uppercase tracking-wide mb-1">Domain</label>
<select id="edit-domain" name="domain"
class="w-full px-3 py-2 border border-gray-300 text-sm rounded focus:outline-none focus:ring-2 focus:ring-indigo-300">
{% for d in domains %}
<option value="{{ d }}" {% if d == source.domain %}selected{% endif %}>{{ d }}</option>
{% endfor %}
</select>
</div>
<div>
<label for="edit-focus" class="block text-xs font-semibold text-gray-500 uppercase tracking-wide mb-1">Focus</label>
<textarea id="edit-focus" name="focus" rows="3"
class="w-full px-3 py-2 border border-gray-300 text-sm rounded focus:outline-none focus:ring-2 focus:ring-indigo-300">{{ source.focus or '' }}</textarea>
</div>
<div class="flex gap-2 pt-1">
<button type="submit"
class="px-4 py-2 bg-indigo-600 text-white text-sm font-semibold rounded hover:bg-indigo-700 transition-colors">
Save
</button>
<a href="/sources/{{ source.id }}"
class="px-4 py-2 border border-gray-300 text-gray-700 text-sm font-semibold rounded hover:bg-gray-50 transition-colors">
Cancel
</a>
</div>
</form>
{% else %}
<!-- Meta row -->
<div class="flex flex-wrap gap-2 mb-6">
{% set status_colors = {
@ -117,9 +60,9 @@
</div>
{% endif %}
<!-- Actions row: accept/reject (when analyzed) + retry (always) -->
<div class="flex flex-wrap gap-3 mb-8">
<!-- Accept / Reject actions (only when analyzed) -->
{% if source.status == 'analyzed' %}
<div class="flex gap-3 mb-8">
<button hx-post="/sources/{{ source.id }}/accept"
hx-target="body"
hx-swap="innerHTML"
@ -132,31 +75,6 @@
class="px-5 py-2 bg-red-500 text-white text-sm font-semibold rounded hover:bg-red-600 transition-colors">
Reject
</button>
{% endif %}
<!-- Re-run: full reset → PENDING. Re-pulls and re-processes from scratch. -->
<button hx-post="/sources/{{ source.id }}/retry"
hx-vals='{"mode": "full"}'
hx-target="body"
hx-swap="innerHTML"
hx-confirm="Re-run this source from scratch? Status will reset to PENDING and the pipeline will re-pull{% if source.source_type == 'video' %} and re-transcribe (expensive){% endif %}."
class="px-5 py-2 bg-amber-500 text-white text-sm font-semibold rounded hover:bg-amber-600 transition-colors">
Re-run from scratch
</button>
{% if source.has_transcript %}
<!-- Light retry: skip the re-pull, just re-run LLM extraction. Works
for any source with stored body text (videos have transcripts,
articles have the trafilatura-extracted body). -->
<button hx-post="/sources/{{ source.id }}/retry"
hx-vals='{"mode": "extract"}'
hx-target="body"
hx-swap="innerHTML"
hx-confirm="Re-extract only — keep the existing transcript/body and re-run the LLM extraction?"
class="px-5 py-2 bg-amber-100 text-amber-800 text-sm font-semibold rounded hover:bg-amber-200 border border-amber-300 transition-colors">
Re-extract only
</button>
{% endif %}
</div>
{% endif %}

258
tests/test_source_edit_retry.py
View File

@ -1,258 +0,0 @@
"""Tests for the source edit-metadata + retry-reset service helpers.
Live-DB pattern matches `tests/test_claim.py` / `tests/test_playlist.py`:
both groups auto-skip when no petalbrain connection is configured. The
service layer is where edit/retry logic lives (the web routes + CLI are
thin wrappers around it), so this is where the meaningful coverage is.
"""
from __future__ import annotations
import os
import uuid
import psycopg
import pytest
def _db_reachable() -> bool:
url = os.environ.get("SECOND_BRAIN_DATABASE_URL") or os.environ.get(
"HERBYLAB_DATABASE_URL"
)
if not url:
return False
raw = url.replace("postgresql+psycopg://", "postgresql://", 1)
try:
with psycopg.connect(raw, connect_timeout=2):
return True
except Exception:
return False
pytestmark = pytest.mark.skipif(
not _db_reachable(),
reason="needs SECOND_BRAIN_DATABASE_URL pointed at a real petalbrain",
)
def _seed_source(db, **overrides):
"""Insert a Source row and return (id, url). Caller cleans up by id."""
from second_brain.models import Source, SourceStatus, SourceType, utcnow
tag = uuid.uuid4().hex[:8]
defaults = dict(
url=f"https://example.test/edit-retry/{tag}",
title=f"seed-{tag}",
domain="development",
source_type=SourceType.ARTICLE,
status=SourceStatus.PENDING,
ingested_at=utcnow(),
)
defaults.update(overrides)
with db.session() as sess:
src = Source(**defaults)
sess.add(src)
sess.flush()
return src.id, src.url
def _cleanup(db, source_id):
from second_brain.models import Source
with db.session() as sess:
sess.query(Source).filter(Source.id == source_id).delete(
synchronize_session=False
)
# ---------------------------------------------------------------------------
# update_source_metadata
# ---------------------------------------------------------------------------
def test_update_metadata_writes_domain_title_focus():
from second_brain.database import Database, reset_database_singleton
from second_brain.models import Source
from second_brain.sources_service import update_source_metadata
reset_database_singleton()
db = Database()
sid, _ = _seed_source(db)
try:
with db.session() as sess:
update_source_metadata(
sess,
source_id=sid,
domain="homelab",
title="new title",
focus="new focus directive",
)
with db.session() as sess:
row = sess.query(Source).filter(Source.id == sid).first()
assert row.domain == "homelab"
assert row.title == "new title"
assert row.focus == "new focus directive"
assert row.updated_at is not None
finally:
_cleanup(db, sid)
reset_database_singleton()
def test_update_metadata_blank_strings_become_null():
from second_brain.database import Database, reset_database_singleton
from second_brain.models import Source
from second_brain.sources_service import update_source_metadata
reset_database_singleton()
db = Database()
sid, _ = _seed_source(db)
try:
with db.session() as sess:
update_source_metadata(
sess, source_id=sid, domain="development", title=" ", focus=""
)
with db.session() as sess:
row = sess.query(Source).filter(Source.id == sid).first()
assert row.title is None
assert row.focus is None
finally:
_cleanup(db, sid)
reset_database_singleton()
def test_update_metadata_rejects_unknown_domain():
from second_brain.database import Database, reset_database_singleton
from second_brain.sources_service import update_source_metadata
reset_database_singleton()
db = Database()
sid, _ = _seed_source(db)
try:
with db.session() as sess:
with pytest.raises(ValueError, match="unknown domain"):
update_source_metadata(sess, source_id=sid, domain="not-a-domain")
finally:
_cleanup(db, sid)
reset_database_singleton()
def test_update_metadata_missing_source_raises_lookup():
from second_brain.database import Database, reset_database_singleton
from second_brain.sources_service import update_source_metadata
reset_database_singleton()
db = Database()
try:
with db.session() as sess:
with pytest.raises(LookupError):
update_source_metadata(
sess, source_id=-99999, domain="development"
)
finally:
reset_database_singleton()
# ---------------------------------------------------------------------------
# reset_source_for_retry
# ---------------------------------------------------------------------------
def test_retry_full_resets_to_pending_and_clears_claim():
from second_brain.database import Database, reset_database_singleton
from second_brain.models import Source, SourceStatus, utcnow
from second_brain.sources_service import reset_source_for_retry
reset_database_singleton()
db = Database()
sid, _ = _seed_source(
db,
status=SourceStatus.FAILED,
error_message="boom",
claimed_by="worker-x",
claimed_at=utcnow(),
)
try:
with db.session() as sess:
reset_source_for_retry(sess, source_id=sid, mode="full")
with db.session() as sess:
row = sess.query(Source).filter(Source.id == sid).first()
assert row.status == SourceStatus.PENDING
assert row.claimed_by is None
assert row.claimed_at is None
assert row.error_message is None
finally:
_cleanup(db, sid)
reset_database_singleton()
def test_retry_extract_sets_transcribed_when_transcript_present():
from second_brain.database import Database, reset_database_singleton
from second_brain.models import Source, SourceStatus
from second_brain.sources_service import reset_source_for_retry
reset_database_singleton()
db = Database()
sid, _ = _seed_source(
db,
status=SourceStatus.ANALYZED,
transcript_text="word " * 50,
error_message="prior failure",
)
try:
with db.session() as sess:
reset_source_for_retry(sess, source_id=sid, mode="extract")
with db.session() as sess:
row = sess.query(Source).filter(Source.id == sid).first()
assert row.status == SourceStatus.TRANSCRIBED
assert row.error_message is None
assert row.transcript_text # transcript preserved
finally:
_cleanup(db, sid)
reset_database_singleton()
def test_retry_extract_without_transcript_raises():
from second_brain.database import Database, reset_database_singleton
from second_brain.models import SourceStatus
from second_brain.sources_service import reset_source_for_retry
reset_database_singleton()
db = Database()
sid, _ = _seed_source(db, status=SourceStatus.PENDING, transcript_text=None)
try:
with db.session() as sess:
with pytest.raises(ValueError, match="requires an existing transcript"):
reset_source_for_retry(sess, source_id=sid, mode="extract")
finally:
_cleanup(db, sid)
reset_database_singleton()
def test_retry_unknown_mode_raises():
from second_brain.database import Database, reset_database_singleton
from second_brain.sources_service import reset_source_for_retry
reset_database_singleton()
db = Database()
sid, _ = _seed_source(db)
try:
with db.session() as sess:
with pytest.raises(ValueError, match="unknown retry mode"):
reset_source_for_retry(sess, source_id=sid, mode="bogus")
finally:
_cleanup(db, sid)
reset_database_singleton()
def test_retry_missing_source_raises_lookup():
from second_brain.database import Database, reset_database_singleton
from second_brain.sources_service import reset_source_for_retry
reset_database_singleton()
db = Database()
try:
with db.session() as sess:
with pytest.raises(LookupError):
reset_source_for_retry(sess, source_id=-99999, mode="full")
finally:
reset_database_singleton()