Capture ThinkStorm project: codebase state, workflows, and access control policies

tests/test_api.py

@@ -0,0 +1,129 @@
+"""
+REST API Endpoint Tests
+Tests Anonymous intake, auth sessions, claims, and role-based permissions.
+"""
+
+import pytest
+from starlette.testclient import TestClient
+from thinkstorm.main import app
+from thinkstorm.database import init_db
+from thinkstorm.config import config
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_app():
+    init_db()
+
+def test_api_anonymous_submission():
+    client = TestClient(app)
+    resp = client.post("/api/ideas", json={"text": "Decentralized task scheduler with automated cron jobs and webhooks."})
+    assert resp.status_code == 201
+    data = resp.json()
+    assert data["id"].startswith("TS-")
+    assert data["lifecycle_state"] == "SUBMITTED"
+
+def test_api_ideas_list():
+    client = TestClient(app)
+    
+    # 1. Unauthenticated -> 401 Unauthorized
+    resp_unauth = client.get("/api/ideas")
+    assert resp_unauth.status_code == 401
+
+    # 2. Authenticated -> 200 OK list
+    login_resp = client.post("/api/auth/login", json={"username": "researcher", "password": "thinkstorm_user_2026"})
+    assert login_resp.status_code == 200
+    token = login_resp.json()["token"]
+    
+    resp_auth = client.get("/api/ideas", headers={"Authorization": f"Bearer {token}"})
+    assert resp_auth.status_code == 200
+    data = resp_auth.json()
+    assert isinstance(data, list)
+
+def test_api_auth_and_claim_flow():
+    client = TestClient(app)
+    
+    # 1. Submit idea
+    sub_resp = client.post("/api/ideas", json={"text": "Real-time vector search indexer for personal notes."})
+    idea_id = sub_resp.json()["id"]
+
+    # Make idea available in db
+    from thinkstorm.database import get_db
+    with get_db() as conn:
+        conn.execute("UPDATE ideas SET lifecycle_state = 'AVAILABLE' WHERE id = ?", (idea_id,))
+
+    # 2. Try to claim anonymously -> 401
+    claim_unauth = client.post(f"/api/ideas/{idea_id}/claim")
+    assert claim_unauth.status_code == 401
+
+    # 3. Login as researcher
+    login_resp = client.post("/api/auth/login", json={"username": "researcher", "password": "thinkstorm_user_2026"})
+    assert login_resp.status_code == 200
+    token = login_resp.json()["token"]
+
+    # 4. Claim idea with session
+    claim_auth = client.post(f"/api/ideas/{idea_id}/claim", headers={"Authorization": f"Bearer {token}"})
+    assert claim_auth.status_code == 200
+    assert "claimed" in claim_auth.json()["message"].lower()
+
+def test_admin_endpoints_permission():
+    client = TestClient(app)
+
+    # 1. Anonymously access admin -> 401 or 403
+    adm_unauth = client.get("/api/admin/prompts")
+    assert adm_unauth.status_code in (401, 403)
+
+    # 2. Login as admin
+    login_adm = client.post("/api/auth/login", json={"username": "admin", "password": config.admin_bootstrap_key})
+    assert login_adm.status_code == 200
+    adm_token = login_adm.json()["token"]
+
+    # 3. Admin access prompts
+    adm_prompts = client.get("/api/admin/prompts", headers={"Authorization": f"Bearer {adm_token}"})
+    assert adm_prompts.status_code == 200
+    prompts = adm_prompts.json()
+    assert len(prompts) > 0
+    # Full prompt template visible to admin
+    assert prompts[0]["system_prompt"] != "[REDACTED: ADMIN ONLY]"
+
+def test_unauthenticated_cannot_browse_ideas():
+    client = TestClient(app)
+    
+    # 1. Anonymous submission is allowed
+    sub_resp = client.post("/api/ideas", json={"text": "New idea submitted by anonymous visitor."})
+    assert sub_resp.status_code == 201
+    idea_id = sub_resp.json()["id"]
+
+    # 2. Unauthenticated GET /ideas (HTML Browse page) -> Redirect to /login
+    html_ideas_resp = client.get("/ideas", follow_redirects=False)
+    assert html_ideas_resp.status_code in (302, 307)
+    assert html_ideas_resp.headers.get("location") == "/login"
+
+    # 3. Unauthenticated GET /ideas/{idea_id} (HTML Dossier page) -> Redirect to /login
+    html_detail_resp = client.get(f"/ideas/{idea_id}", follow_redirects=False)
+    assert html_detail_resp.status_code in (302, 307)
+    assert html_detail_resp.headers.get("location") == "/login"
+
+    # 4. Unauthenticated GET /api/ideas (API Browse list) -> 401 Unauthorized
+    api_ideas_resp = client.get("/api/ideas")
+    assert api_ideas_resp.status_code == 401
+    assert "authentication required" in api_ideas_resp.json()["detail"].lower()
+
+    # 5. Unauthenticated GET /api/ideas/{idea_id} (API Dossier detail) -> 401 Unauthorized
+    api_detail_resp = client.get(f"/api/ideas/{idea_id}")
+    assert api_detail_resp.status_code == 401
+    assert "authentication required" in api_detail_resp.json()["detail"].lower()
+
+    # 6. Authenticated user can browse and view ideas
+    login_resp = client.post("/api/auth/login", json={"username": "researcher", "password": "thinkstorm_user_2026"})
+    assert login_resp.status_code == 200
+    token = login_resp.json()["token"]
+    auth_headers = {"Authorization": f"Bearer {token}"}
+
+    # Authenticated API list -> 200
+    auth_list_resp = client.get("/api/ideas", headers=auth_headers)
+    assert auth_list_resp.status_code == 200
+    assert isinstance(auth_list_resp.json(), list)
+
+    # Authenticated API detail -> 200
+    auth_detail_resp = client.get(f"/api/ideas/{idea_id}", headers=auth_headers)
+    assert auth_detail_resp.status_code == 200
+    assert auth_detail_resp.json()["id"] == idea_id

tests/test_artifact_versioning.py

@@ -0,0 +1,73 @@
+import pytest
+import asyncio
+from fastapi.testclient import TestClient
+from thinkstorm.main import app
+from thinkstorm.database import init_db, get_db, get_utc_now
+from thinkstorm.processors.pipeline import execute_work_track_workflow
+
+@pytest.fixture(autouse=True)
+def setup_database():
+    init_db()
+
+client = TestClient(app)
+
+def test_work_track_artifact_versioning_and_deletion():
+    # 1. Login as admin
+    login_res = client.post("/api/auth/login", json={"username": "admin", "password": "admin-thinkstorm-pass-2026"})
+    assert login_res.status_code == 200
+
+    # 2. Submit an idea
+    sub_res = client.post("/api/ideas", json={"text": "Test Idea for Artifact Versioning and Deletion System"})
+    assert sub_res.status_code == 201
+    idea_id = sub_res.json()["id"]
+
+    # 3. Create a work track
+    track_res = client.post(
+        f"/api/ideas/{idea_id}/work-tracks",
+        json={"work_type_id": "ARTICLE", "name": "Article Test Track", "model_override": "auto/best-fast"}
+    )
+    assert track_res.status_code == 200
+    track_id = track_res.json()["id"]
+    assert track_res.json()["model_override"] == "auto/best-fast"
+
+    # 4. Execute work track workflow (Run 1 -> Version 1)
+    asyncio.run(execute_work_track_workflow(track_id, model_override="auto/best-fast"))
+
+    with get_db() as conn:
+        outs_v1 = conn.execute("SELECT id, name, version, is_current, model_used FROM work_track_outputs WHERE work_track_id = ? ORDER BY version DESC", (track_id,)).fetchall()
+        assert len(outs_v1) >= 2
+        for o in outs_v1:
+            assert o["version"] == 1
+            assert o["is_current"] == 1
+
+    # 5. Execute work track workflow again (Run 2 -> Version 2)
+    asyncio.run(execute_work_track_workflow(track_id, model_override="auto/best-reasoning"))
+
+    with get_db() as conn:
+        outs_v2 = conn.execute("SELECT id, name, version, is_current, model_used FROM work_track_outputs WHERE work_track_id = ? ORDER BY name ASC, version DESC", (track_id,)).fetchall()
+        # Should now have 4 records (2 for article.md: v2 and v1; 2 for outline.md: v2 and v1)
+        assert len(outs_v2) == 4
+        article_v2 = [o for o in outs_v2 if o["name"] == "article.md" and o["version"] == 2][0]
+        article_v1 = [o for o in outs_v2 if o["name"] == "article.md" and o["version"] == 1][0]
+        assert article_v2["is_current"] == 1
+        assert article_v1["is_current"] == 0
+
+    # 6. Verify GET /api/ideas/{id} serializes version metadata
+    detail_res = client.get(f"/api/ideas/{idea_id}")
+    assert detail_res.status_code == 200
+    data = detail_res.json()
+    track_data = [t for t in data["work_tracks"] if t["id"] == track_id][0]
+    assert len(track_data["outputs"]) == 4
+    output_ids = [o["id"] for o in track_data["outputs"]]
+
+    # 7. Delete artifact version 2 of article.md
+    del_res = client.delete(f"/api/ideas/work-tracks/outputs/{article_v2['id']}")
+    assert del_res.status_code == 200
+    assert "deleted successfully" in del_res.json()["message"]
+
+    # 8. Verify article_v1 is now promoted to is_current = 1
+    with get_db() as conn:
+        remaining_article = conn.execute("SELECT id, version, is_current FROM work_track_outputs WHERE work_track_id = ? AND name = 'article.md'", (track_id,)).fetchall()
+        assert len(remaining_article) == 1
+        assert remaining_article[0]["version"] == 1
+        assert remaining_article[0]["is_current"] == 1

tests/test_database.py

@@ -0,0 +1,36 @@
+"""
+Database & Sequence Generator Unit Tests
+"""
+
+import pytest
+import sqlite3
+from thinkstorm.database import init_db, get_db, next_sequence, hash_password
+from thinkstorm.models import UserRole
+
+def test_init_db_and_seeds():
+    init_db()
+    with get_db() as conn:
+        users = conn.execute("SELECT * FROM users").fetchall()
+        assert len(users) >= 2
+        
+        categories = conn.execute("SELECT * FROM categories").fetchall()
+        assert len(categories) >= 5
+
+        work_types = conn.execute("SELECT * FROM work_types").fetchall()
+        assert len(work_types) >= 3
+
+        prompts = conn.execute("SELECT * FROM prompt_definitions").fetchall()
+        assert len(prompts) >= 5
+
+def test_sequence_generators():
+    s1 = next_sequence("idea")
+    s2 = next_sequence("idea")
+    assert s1.startswith("TS-")
+    assert s2.startswith("TS-")
+    assert int(s2.split("-")[1]) == int(s1.split("-")[1]) + 1
+
+    w1 = next_sequence("work_track")
+    assert w1.startswith("WT-")
+
+    r1 = next_sequence("run")
+    assert r1.startswith("RUN-")

tests/test_lifecycle.py

@@ -0,0 +1,93 @@
+"""
+Idea Lifecycle & Workflow State Machine Tests
+PRD Section 10, 11, 28, 30, 31, 34:
+- Separation of Lifecycle State and Processing State
+- Claiming (AVAILABLE -> CLAIMED)
+- Activation (CLAIMED -> ACTIVE)
+- Work tracks & outputs
+"""
+
+import pytest
+import asyncio
+from thinkstorm.database import init_db, get_db, next_sequence, get_utc_now
+from thinkstorm.processors.pipeline import execute_intake_pipeline, execute_work_track_workflow
+from thinkstorm.models import LifecycleState, ProcessingState
+
+@pytest.mark.asyncio
+async def test_full_idea_intake_and_claim_lifecycle():
+    init_db()
+    idea_id = next_sequence("idea")
+    raw_text = f"Quantum biological sensor interface for monitoring mitochondrial ATP flux oscillations in real-time {idea_id} https://example.com/sensor"
+    now = get_utc_now()
+
+    # 1. Anonymous submission
+    with get_db() as conn:
+        conn.execute(
+            """
+            INSERT INTO ideas (id, original_text, submitted_at, title, summary, lifecycle_state, processing_state, enrichment_level, created_at, updated_at)
+            VALUES (?, ?, ?, 'Pending', 'Pending', 'SUBMITTED', 'QUEUED', 0, ?, ?)
+            """,
+            (idea_id, raw_text, now, now, now)
+        )
+
+    # 2. Run intake pipeline
+    await execute_intake_pipeline(idea_id)
+
+    # 3. Verify Idea reached AVAILABLE
+    with get_db() as conn:
+        idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
+        assert idea["lifecycle_state"] == LifecycleState.AVAILABLE.value
+        assert idea["processing_state"] == ProcessingState.IDLE.value
+        assert idea["enrichment_level"] >= 4
+        assert len(idea["title"]) > 0
+        assert idea["opengist_id"] is not None
+
+        # Verify raw original text was NEVER mutated
+        assert idea["original_text"] == raw_text
+
+        # Verify runs were recorded
+        runs = conn.execute("SELECT * FROM processor_runs WHERE idea_id = ?", (idea_id,)).fetchall()
+        assert len(runs) >= 4
+        total_tokens = sum(r["total_tokens"] for r in runs)
+        assert total_tokens > 0
+
+    # 4. Claim Idea (AVAILABLE -> CLAIMED)
+    claim_time = get_utc_now()
+    with get_db() as conn:
+        conn.execute(
+            "UPDATE ideas SET lifecycle_state = 'CLAIMED', claimed_by = 'researcher', claimed_at = ? WHERE id = ?",
+            (claim_time, idea_id)
+        )
+        claimed_idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
+        assert claimed_idea["lifecycle_state"] == LifecycleState.CLAIMED.value
+        assert claimed_idea["claimed_by"] == "researcher"
+
+    # 5. Activate Idea (CLAIMED -> ACTIVE)
+    with get_db() as conn:
+        conn.execute(
+            "UPDATE ideas SET lifecycle_state = 'ACTIVE' WHERE id = ?",
+            (idea_id,)
+        )
+        active_idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
+        assert active_idea["lifecycle_state"] == LifecycleState.ACTIVE.value
+
+    # 6. Add Work Track
+    track_id = next_sequence("work_track")
+    with get_db() as conn:
+        conn.execute(
+            """
+            INSERT INTO work_tracks (id, idea_id, work_type_id, name, state, created_at)
+            VALUES (?, ?, 'CODING_PROJECT', 'Core Architecture Blueprint', 'PLANNED', ?)
+            """,
+            (track_id, idea_id, get_utc_now())
+        )
+
+    # 7. Execute Work Track Workflow
+    await execute_work_track_workflow(track_id)
+
+    # 8. Verify Work Track completed and outputs produced
+    with get_db() as conn:
+        track = conn.execute("SELECT * FROM work_tracks WHERE id = ?", (track_id,)).fetchone()
+        assert track["state"] == "COMPLETED"
+        outputs = conn.execute("SELECT * FROM work_track_outputs WHERE work_track_id = ?", (track_id,)).fetchall()
+        assert len(outputs) >= 2

tests/test_prompts.py

@@ -0,0 +1,67 @@
+"""
+Prompt Versioning & Role Visibility Tests
+PRD Section 21, 23, 25:
+- Editing a prompt creates a new immutable version without mutating historical provenance.
+- Non-admin users cannot view full system/user prompt templates.
+"""
+
+import pytest
+from thinkstorm.database import init_db
+from thinkstorm.prompts.catalog import (
+    get_all_prompts, get_prompt_version, update_prompt, duplicate_prompt
+)
+
+def test_prompt_versioning_immutability():
+    init_db()
+    prompt_id = "test-versioning-prompt"
+    
+    # 1. Create a fresh test prompt definition
+    from thinkstorm.database import get_db
+    with get_db() as conn:
+        conn.execute("DELETE FROM prompt_versions WHERE prompt_definition_id = ?", (prompt_id,))
+        conn.execute("DELETE FROM prompt_definitions WHERE id = ?", (prompt_id,))
+    duplicate_prompt("normalize-idea", prompt_id, "Test Versioning Prompt", created_by="test_user")
+
+    # Verify baseline v1
+    v1 = get_prompt_version(prompt_id, version=1, is_admin=True)
+    assert v1 is not None
+    assert v1["version"] == 1
+    old_hash = v1["prompt_hash"]
+
+    # 2. Update prompt -> creates v2
+    new_version = update_prompt(
+        prompt_id=prompt_id,
+        system_prompt="Updated system prompt for normalizer v2.",
+        user_prompt_template="Analyze: {{submission_text}}",
+        model_policy="fast",
+        updated_by="admin_test"
+    )
+    assert new_version == 2
+
+    # 3. Check current is v2
+    current = get_prompt_version(prompt_id, is_admin=True)
+    assert current["version"] == 2
+    assert current["system_prompt"] == "Updated system prompt for normalizer v2."
+    assert current["prompt_hash"] != old_hash
+
+    # 4. Verify historical v1 remains unchanged and intact
+    v1_recheck = get_prompt_version(prompt_id, version=1, is_admin=True)
+    assert v1_recheck["version"] == 1
+    assert v1_recheck["prompt_hash"] == old_hash
+    assert "Updated system prompt" not in v1_recheck["system_prompt"]
+
+def test_non_admin_prompt_masking():
+    init_db()
+    prompt_id = "prior-art-search"
+
+    # Non-admin retrieval
+    public_view = get_prompt_version(prompt_id, is_admin=False)
+    assert public_view["system_prompt"] == "[REDACTED: ADMIN ONLY]"
+    assert public_view["user_prompt_template"] == "[REDACTED: ADMIN ONLY]"
+    assert public_view["name"] is not None
+    assert public_view["model_policy"] is not None
+
+    # Admin retrieval
+    admin_view = get_prompt_version(prompt_id, is_admin=True)
+    assert admin_view["system_prompt"] != "[REDACTED: ADMIN ONLY]"
+    assert admin_view["user_prompt_template"] != "[REDACTED: ADMIN ONLY]"

tests/test_trash.py

@@ -0,0 +1,133 @@
+"""
+ThinkStorm Trash & Empty Trash Tests
+Tests moving ideas to trash, hiding from default browsing, restoring from trash,
+permanently deleting single ideas, and emptying the entire trash bin with cascade cleanup.
+"""
+
+import pytest
+from starlette.testclient import TestClient
+from thinkstorm.main import app
+from thinkstorm.database import init_db, get_db
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_app():
+    init_db()
+
+def get_auth_token(username="admin", password="admin-thinkstorm-pass-2026"):
+    client = TestClient(app)
+    resp = client.post("/api/auth/login", json={"username": username, "password": password})
+    assert resp.status_code == 200
+    return resp.json()["token"]
+
+def test_trash_and_hide_from_default_catalog():
+    client = TestClient(app)
+    token = get_auth_token()
+    headers = {"Authorization": f"Bearer {token}"}
+
+    # 1. Create an idea and set to AVAILABLE
+    sub_resp = client.post("/api/ideas", json={"text": "A low-value idea that should be discarded: spam text test."})
+    idea_id = sub_resp.json()["id"]
+
+    with get_db() as conn:
+        conn.execute("UPDATE ideas SET lifecycle_state = 'AVAILABLE' WHERE id = ?", (idea_id,))
+
+    # Verify visible in default list
+    list_resp = client.get("/api/ideas", headers=headers)
+    idea_ids_default = [i["id"] for i in list_resp.json()]
+    assert idea_id in idea_ids_default
+
+    # 2. Move idea to Trash
+    trash_resp = client.post(f"/api/ideas/{idea_id}/trash", headers=headers)
+    assert trash_resp.status_code == 200
+    trash_data = trash_resp.json()
+    assert trash_data["lifecycle_state"] == "TRASHED"
+    assert trash_data["previous_lifecycle_state"] == "AVAILABLE"
+
+    # 3. Verify hidden from default ideas list and ALL filter
+    list_resp_after = client.get("/api/ideas", headers=headers)
+    idea_ids_after = [i["id"] for i in list_resp_after.json()]
+    assert idea_id not in idea_ids_after
+
+    list_all_resp = client.get("/api/ideas?state=ALL", headers=headers)
+    idea_ids_all = [i["id"] for i in list_all_resp.json()]
+    assert idea_id not in idea_ids_all
+
+    # 4. Verify visible when explicitly filtering by TRASHED
+    list_trashed_resp = client.get("/api/ideas?state=TRASHED", headers=headers)
+    idea_ids_trashed = [i["id"] for i in list_trashed_resp.json()]
+    assert idea_id in idea_ids_trashed
+
+def test_restore_idea_from_trash():
+    client = TestClient(app)
+    token = get_auth_token()
+    headers = {"Authorization": f"Bearer {token}"}
+
+    # 1. Create idea, set to CLAIMED, then trash it
+    sub_resp = client.post("/api/ideas", json={"text": "Idea that was mistakenly trashed and needs retrieval."})
+    idea_id = sub_resp.json()["id"]
+
+    with get_db() as conn:
+        conn.execute("UPDATE ideas SET lifecycle_state = 'CLAIMED', claimed_by = 'admin' WHERE id = ?", (idea_id,))
+
+    trash_resp = client.post(f"/api/ideas/{idea_id}/trash", headers=headers)
+    assert trash_resp.status_code == 200
+
+    # 2. Restore idea
+    restore_resp = client.post(f"/api/ideas/{idea_id}/restore", headers=headers)
+    assert restore_resp.status_code == 200
+    assert restore_resp.json()["lifecycle_state"] == "CLAIMED"
+
+    # 3. Verify idea is restored and visible
+    detail_resp = client.get(f"/api/ideas/{idea_id}", headers=headers)
+    assert detail_resp.status_code == 200
+    assert detail_resp.json()["lifecycle_state"] == "CLAIMED"
+
+def test_permanently_delete_single_idea():
+    client = TestClient(app)
+    token = get_auth_token()
+    headers = {"Authorization": f"Bearer {token}"}
+
+    # 1. Create idea and trash it
+    sub_resp = client.post("/api/ideas", json={"text": "Idea to be deleted permanently right now."})
+    idea_id = sub_resp.json()["id"]
+    client.post(f"/api/ideas/{idea_id}/trash", headers=headers)
+
+    # 2. Permanently delete
+    del_resp = client.delete(f"/api/ideas/{idea_id}/permanent", headers=headers)
+    assert del_resp.status_code == 200
+    assert "permanently deleted" in del_resp.json()["message"].lower()
+
+    # 3. Verify it no longer exists anywhere
+    detail_resp = client.get(f"/api/ideas/{idea_id}", headers=headers)
+    assert detail_resp.status_code == 404
+
+def test_empty_trash_bulk():
+    client = TestClient(app)
+    token = get_auth_token()
+    headers = {"Authorization": f"Bearer {token}"}
+
+    # 1. Create 3 ideas and trash all 3
+    trash_ids = []
+    for i in range(3):
+        resp = client.post("/api/ideas", json={"text": f"Junk idea {i+1} destined for trash empty test."})
+        i_id = resp.json()["id"]
+        client.post(f"/api/ideas/{i_id}/trash", headers=headers)
+        trash_ids.append(i_id)
+
+    # Verify all 3 are currently in trash
+    trashed_list = client.get("/api/ideas?state=TRASHED", headers=headers).json()
+    trashed_ids_in_db = [x["id"] for x in trashed_list]
+    for tid in trash_ids:
+        assert tid in trashed_ids_in_db
+
+    # 2. Empty Trash
+    empty_resp = client.post("/api/trash/empty", headers=headers)
+    assert empty_resp.status_code == 200
+    data = empty_resp.json()
+    assert data["deleted_count"] >= 3
+
+    # 3. Verify trash list is now empty of those IDs
+    trashed_after = client.get("/api/ideas?state=TRASHED", headers=headers).json()
+    trashed_after_ids = [x["id"] for x in trashed_after]
+    for tid in trash_ids:
+        assert tid not in trashed_after_ids

tests/test_url_safety.py

@@ -0,0 +1,55 @@
+"""
+URL Extraction & VirusTotal Safety Rule Tests
+PRD Section 14, 15:
+- Any VirusTotal malicious detection (malicious > 0) blocks automation and quarantines idea.
+- Safe URLs are approved.
+"""
+
+import pytest
+import asyncio
+from thinkstorm.processors.pipeline import process_url_safety, URL_REGEX
+from thinkstorm.models import URLSafetyState, AutomationPolicy
+from thinkstorm.database import init_db
+
+@pytest.mark.asyncio
+async def test_url_extraction():
+    text = "Check out this repo https://github.com/thinkstorm/core and also http://test.com/docs for details."
+    urls = URL_REGEX.findall(text)
+    assert len(urls) == 2
+    assert "https://github.com/thinkstorm/core" in urls
+    assert "http://test.com/docs" in urls
+
+@pytest.mark.asyncio
+async def test_safe_url_safety_check():
+    init_db()
+    idea_id = "TS-9901"
+    now = "2026-08-19T09:00:00Z"
+    from thinkstorm.database import get_db
+    with get_db() as conn:
+        conn.execute("INSERT OR IGNORE INTO ideas (id, original_text, submitted_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?)", (idea_id, "text", now, now, now))
+
+    text = "Great idea with safe link https://example.com/project"
+    url_records, quarantine_needed = await process_url_safety(idea_id, text)
+    
+    assert len(url_records) == 1
+    assert quarantine_needed is False
+    assert url_records[0].safety_state == URLSafetyState.SAFE
+    assert url_records[0].automation_policy == AutomationPolicy.APPROVED
+
+@pytest.mark.asyncio
+async def test_malicious_url_safety_check_blocks_and_quarantines():
+    init_db()
+    idea_id = "TS-9902"
+    now = "2026-08-19T09:00:00Z"
+    from thinkstorm.database import get_db
+    with get_db() as conn:
+        conn.execute("INSERT OR IGNORE INTO ideas (id, original_text, submitted_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?)", (idea_id, "text", now, now, now))
+
+    text = "Suspicious submission with bad link https://malicious-site.testmalicious/payload"
+    url_records, quarantine_needed = await process_url_safety(idea_id, text)
+    
+    assert len(url_records) == 1
+    assert quarantine_needed is True
+    assert url_records[0].safety_state == URLSafetyState.MALICIOUS
+    assert url_records[0].automation_policy == AutomationPolicy.BLOCKED
+    assert url_records[0].admin_review_required is True