diff --git a/backend/Dockerfile b/backend/Dockerfile index a4acd193..be071f57 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -9,7 +9,7 @@ WORKDIR /app RUN apt-get clean && \ rm -rf /var/lib/apt/lists/* && \ apt-get update && \ - apt-get install -y \ + apt-get install -y --no-install-recommends \ gcc \ g++ \ git \ diff --git a/backend/api/citations/router.py b/backend/api/citations/router.py index 436f45b5..161452ea 100644 --- a/backend/api/citations/router.py +++ b/backend/api/citations/router.py @@ -31,7 +31,7 @@ rispy = None from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, status -from fastapi.responses import Response +from fastapi.responses import Response, StreamingResponse from fastapi.concurrency import run_in_threadpool from pydantic import BaseModel @@ -920,8 +920,15 @@ async def upload_citation_fulltext( sr_id: str, citation_id: int, file: UploadFile = File(...), + mode: str = 'replace', current_user: dict[str, Any] = Depends(get_current_active_user), ): + mode = str(mode or '').strip().lower() + if mode not in {'replace', 'supplementary'}: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="mode must be 'replace' or 'supplementary'", + ) # The backend is authoritative: a PDF-linkage worker and manual upload must # never compete for an SR. A final conditional worker update remains the # race-safe fallback for uploads that began just before job creation. @@ -1002,16 +1009,24 @@ async def upload_citation_fulltext( ) try: - from ..services.fulltext_attachment_service import attach_fulltext_document - result = await attach_fulltext_document( - citation_id=citation_id, - table_name=table_name, - user_id=str(current_user['id']), - filename=file.filename, - content=content, - source='manual', - replace=True, + from ..services.fulltext_attachment_service import ( + add_supplementary_document, attach_fulltext_document, ) + if mode == 'supplementary': + result = await add_supplementary_document( + citation_id=citation_id, table_name=table_name, + user_id=str(current_user['id']), filename=file.filename, content=content, + ) + else: + result = await attach_fulltext_document( + citation_id=citation_id, + table_name=table_name, + user_id=str(current_user['id']), + filename=file.filename, + content=content, + source='manual', + replace=True, + ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as e: @@ -1033,6 +1048,100 @@ async def upload_citation_fulltext( } +@router.get('/{sr_id}/citations/{citation_id}/fulltext-documents') +async def get_fulltext_documents( + sr_id: str, citation_id: int, + current_user: dict[str, Any] = Depends(get_current_active_user), +): + _, screening = await load_sr_and_check(sr_id, current_user, srdb_service) + table_name = (screening or {}).get('table_name') or 'citations' + row = await run_in_threadpool(cits_dp_service.get_citation_by_id, citation_id, table_name) + if not row: + raise HTTPException(status_code=404, detail='Citation not found') + from ..services.fulltext_attachment_service import ensure_legacy_document, list_fulltext_documents + await run_in_threadpool(ensure_legacy_document, citation_id, table_name, row) + documents = await run_in_threadpool(list_fulltext_documents, citation_id, table_name) + return {'documents': documents, 'title': row.get('title') or row.get('citation') or ''} + + +@router.get('/{sr_id}/citations/{citation_id}/fulltext-documents/{document_id}/download') +async def download_fulltext_document( + sr_id: str, citation_id: int, document_id: str, + current_user: dict[str, Any] = Depends(get_current_active_user), +): + """Download a citation attachment after checking SR membership and linkage.""" + _, screening = await load_sr_and_check(sr_id, current_user, srdb_service) + table_name = (screening or {}).get('table_name') or 'citations' + row = await run_in_threadpool(cits_dp_service.get_citation_by_id, citation_id, table_name) + if not row: + raise HTTPException(status_code=404, detail='Citation not found') + from ..services.fulltext_attachment_service import ensure_legacy_document, get_fulltext_document + from ..services.storage import storage_service + await run_in_threadpool(ensure_legacy_document, citation_id, table_name, row) + document = await run_in_threadpool( + get_fulltext_document, citation_id, table_name, document_id, + ) + if not document: + raise HTTPException( + status_code=404, detail='Full-text document not found', + ) + try: + content, filename = await storage_service.get_bytes_by_path(document['storage_path']) + except FileNotFoundError as exc: + raise HTTPException( + status_code=404, detail='Full-text file not found in storage', + ) from exc + except ValueError as exc: + raise HTTPException( + status_code=400, detail='Invalid full-text storage path', + ) from exc + return StreamingResponse( + iter([content]), media_type='application/pdf', + headers={ + 'Content-Disposition': f'inline; filename="{filename or document["filename"]}"', + 'Cache-Control': 'no-store, max-age=0', + }, + ) + + +class ActivateDocumentRequest(BaseModel): + document_id: str + + +@router.post('/{sr_id}/citations/{citation_id}/fulltext-documents/activate') +async def activate_document( + sr_id: str, citation_id: int, payload: ActivateDocumentRequest, + current_user: dict[str, Any] = Depends(get_current_active_user), +): + _, screening = await load_sr_and_check(sr_id, current_user, srdb_service) + table_name = (screening or {}).get('table_name') or 'citations' + from ..services.fulltext_attachment_service import activate_fulltext_document + activated = await run_in_threadpool( + activate_fulltext_document, citation_id, table_name, payload.document_id, + ) + if not activated: + raise HTTPException( + status_code=404, detail='Full-text document not found', + ) + return {'status': 'success'} + + +@router.delete('/{sr_id}/citations/{citation_id}/fulltext-documents/{document_id}') +async def delete_document( + sr_id: str, citation_id: int, document_id: str, + current_user: dict[str, Any] = Depends(get_current_active_user), +): + _, screening = await load_sr_and_check(sr_id, current_user, srdb_service) + table_name = (screening or {}).get('table_name') or 'citations' + from ..services.fulltext_attachment_service import delete_fulltext_document + deleted, was_active = await delete_fulltext_document(citation_id, table_name, document_id) + if not deleted: + raise HTTPException( + status_code=404, detail='Full-text document not found', + ) + return {'status': 'success', 'was_active': was_active} + + # Helper to list fulltext URLs - delegated to backend.api.core.postgres.list_fulltext_urls @@ -1076,6 +1185,9 @@ async def hard_delete_screening_resources(sr_id: str, current_user: dict[str, An # 1) collect fulltext URLs from the screening DB try: urls = await run_in_threadpool(cits_dp_service.list_fulltext_urls, table_name) + from ..services.fulltext_attachment_service import list_registered_storage_paths + registered_urls = await run_in_threadpool(list_registered_storage_paths, table_name) + urls = list(dict.fromkeys([*(urls or []), *(registered_urls or [])])) except RuntimeError as rexc: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(rexc), @@ -1154,6 +1266,13 @@ async def hard_delete_screening_resources(sr_id: str, current_user: dict[str, An # 3) drop the screening table try: await run_in_threadpool(cits_dp_service.drop_table, table_name) + try: + from ..services.fulltext_attachment_service import delete_document_registry + await run_in_threadpool(delete_document_registry, table_name) + except Exception: + # The citation table is authoritative; stale registry rows are harmless + # and can be reconciled separately if registry cleanup is unavailable. + pass table_dropped = True except RuntimeError as rexc: raise HTTPException( diff --git a/backend/api/extract/router.py b/backend/api/extract/router.py index 184d9f73..41384c19 100644 --- a/backend/api/extract/router.py +++ b/backend/api/extract/router.py @@ -701,6 +701,20 @@ async def _run_docint(): [f"[{i}] {x}" for i, x in enumerate(full_text_arr)], ) + # A structurally valid PDF can still yield no machine-readable text + # (for example, an image-only scan or an unsupported text encoding). + # Do not persist this as a successful extraction: downstream screening + # would otherwise misdiagnose absent context as an LLM response failure. + if not full_text_str.strip(): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=( + 'No machine-readable text could be extracted from this PDF. ' + 'The file may be scanned, image-only, or use an unsupported ' + 'encoding. Upload a searchable/OCR-processed PDF and try again.' + ), + ) + # ------------------------- # Upload Azure DI artifacts # ------------------------- @@ -825,6 +839,17 @@ async def _run_docint(): coords_for_overlay = list(annotations) + list(artifact_coords) updated1 = await run_in_threadpool(cits_dp_service.update_text_column, citation_id, 'fulltext', full_text_str, table_name) updated2 = await run_in_threadpool(cits_dp_service.update_text_column, citation_id, 'fulltext_md5', current_md5, table_name) + try: + from ..services.fulltext_attachment_service import ( + ensure_legacy_document, record_active_extracted_text, + ) + await run_in_threadpool(ensure_legacy_document, citation_id, table_name, row) + await run_in_threadpool( + record_active_extracted_text, citation_id, table_name, full_text_str, + ) + except Exception: + # Multi-document indexing is additive and must not fail extraction. + pass updated3 = await run_in_threadpool(cits_dp_service.update_jsonb_column, citation_id, 'fulltext_coords', coords_for_overlay, table_name) updated4 = await run_in_threadpool(cits_dp_service.update_jsonb_column, citation_id, 'fulltext_pages', pages, table_name) updated5 = await run_in_threadpool(cits_dp_service.update_jsonb_column, citation_id, 'fulltext_figures', fulltext_figures, table_name) diff --git a/backend/api/screen/agentic_utils.py b/backend/api/screen/agentic_utils.py index ed15513d..80b78d09 100644 --- a/backend/api/screen/agentic_utils.py +++ b/backend/api/screen/agentic_utils.py @@ -119,7 +119,26 @@ def parse_agent_xml(text: str) -> ParsedAgentXML: class AgentResponseError(ValueError): - """Raised when an LLM response still violates its stage contract after repair.""" + """Raised when an LLM response still violates its stage contract after repair. + + The repaired response is retained so callers that support partial suggestions + can persist the usable fields and show precise warnings to reviewers. + """ + + def __init__( + self, + message: str, + *, + raw_response: str = '', + parsed: ParsedAgentXML | None = None, + metadata: object | None = None, + missing_fields: list[str] | None = None, + ) -> None: + super().__init__(message) + self.raw_response = raw_response + self.parsed = parsed + self.metadata = metadata + self.missing_fields = list(missing_fields or []) def validate_agent_response( @@ -138,7 +157,12 @@ def validate_agent_response( return missing -def build_repair_prompt(*, raw_response: str, stage: AgentStage) -> str: +def build_repair_prompt( + *, + raw_response: str, + stage: AgentStage, + original_prompt: str = '', +) -> str: """Ask the model to reformat an incomplete response without changing its judgment.""" if stage == 'screening': schema = ( @@ -151,10 +175,16 @@ def build_repair_prompt(*, raw_response: str, stage: AgentStage) -> str: 'exact selected option\n' 'number from 0 to 1' ) + context = original_prompt.strip() return f"""Your previous response did not match the required output contract. -Preserve the same judgment, but return ONLY these XML tags with valid, non-empty values: +Use the original task and allowed options below. Preserve the same judgment when it +is recoverable; otherwise make the best supported judgment from the original task. +Return ONLY these XML tags with valid, non-empty values: {schema} +Original task and allowed options: +{context or '(not available)'} + Previous response: {raw_response} """ @@ -173,7 +203,11 @@ async def call_and_parse_agent_response( if not missing: return raw, parsed, metadata, False - repair_prompt = build_repair_prompt(raw_response=raw, stage=stage) + repair_prompt = build_repair_prompt( + raw_response=raw, + stage=stage, + original_prompt=prompt, + ) repaired_raw, repaired_metadata = await call_llm(repair_prompt) repaired = parse_agent_xml(repaired_raw) repaired_missing = validate_agent_response(repaired, stage=stage) @@ -181,6 +215,10 @@ async def call_and_parse_agent_response( fields = ', '.join(repaired_missing) raise AgentResponseError( f'{stage.capitalize()} agent response missing or invalid fields after repair: {fields}', + raw_response=repaired_raw, + parsed=repaired, + metadata=repaired_metadata, + missing_fields=repaired_missing, ) return repaired_raw, repaired, repaired_metadata, True diff --git a/backend/api/screen/router.py b/backend/api/screen/router.py index e2b20cf7..e4bcbb04 100644 --- a/backend/api/screen/router.py +++ b/backend/api/screen/router.py @@ -1681,6 +1681,16 @@ async def run_fulltext_agentic( ) fulltext = str((row or {}).get('fulltext') or '') fulltext_md5 = str((row or {}).get('fulltext_md5') or '') + # Screen against every extracted document for the citation. Document + # boundaries prevent the model from conflating main and supplementary text. + try: + from ..services.fulltext_attachment_service import combined_fulltext + fulltext = await run_in_threadpool( + combined_fulltext, citation_id, table_name, fulltext, + ) + except Exception: + # Registry support is additive; legacy single-PDF screening must remain available. + pass # Tables/Figures context from row tables_md_lines: list[str] = [] @@ -1878,6 +1888,8 @@ async def _call_with_metadata(prompt: str): raw, usage, latency = await _call_llm(prompt) return raw, (usage, latency) + screening_missing_fields: list[str] = [] + screening_repair_failed = False try: screening_raw, screening_parsed, screening_meta, _ = await call_and_parse_agent_response( screening_prompt, @@ -1885,11 +1897,22 @@ async def _call_with_metadata(prompt: str): call_llm=_call_with_metadata, ) except AgentResponseError as exc: - raise HTTPException( - status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc), - ) from exc + if exc.parsed is None or exc.metadata is None: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc), + ) from exc + screening_raw = exc.raw_response + screening_parsed = exc.parsed + screening_meta = exc.metadata + screening_missing_fields = list(exc.missing_fields) + screening_repair_failed = True screening_usage, screening_latency = screening_meta screening_answer = resolve_option(screening_parsed.answer, opts) + screening_confidence = ( + None + if 'confidence' in screening_missing_fields + else screening_parsed.confidence + ) try: screening_run_id = await run_in_threadpool( @@ -1902,12 +1925,14 @@ async def _call_with_metadata(prompt: str): 'criterion_key': criterion_key, 'stage': 'screening', 'answer': screening_answer, - 'confidence': screening_parsed.confidence, + 'confidence': screening_confidence, 'rationale': screening_parsed.rationale, 'raw_response': screening_raw, 'guardrails': { **_build_guardrails(screening_parsed, raw_text=screening_raw, stage='screening'), 'fulltext_md5': fulltext_md5, + 'repair_failed': screening_repair_failed, + 'missing_fields': screening_missing_fields, }, 'model': payload.model, 'prompt_version': payload.prompt_version, @@ -1929,6 +1954,54 @@ async def _call_with_metadata(prompt: str): detail=f"Failed to persist screening run: {e}", ) + # A critical audit requires a valid selected option and confidence. Keep + # the partial screening suggestion, but do not manufacture an audit from + # missing inputs. + if 'answer' in screening_missing_fields or 'confidence' in screening_missing_fields: + try: + llm_col = snake_case_column(q) + partial_payload = { + 'selected': screening_answer or '', + 'confidence': screening_confidence, + 'explanation': screening_parsed.rationale or '', + 'evidence_sentences': screening_parsed.evidence_sentences or [], + 'evidence_tables': screening_parsed.evidence_tables or [], + 'evidence_figures': screening_parsed.evidence_figures or [], + 'source': 'agentic', + 'pipeline': 'fulltext', + 'fulltext_md5': fulltext_md5, + 'partial': True, + 'repair_failed': True, + 'missing_fields': screening_missing_fields, + } + await run_in_threadpool( + cits_dp_service.update_jsonb_column, + citation_id, + llm_col, + partial_payload, + table_name, + ) + except Exception: + pass + results.append({ + 'question': q, + 'criterion_key': criterion_key, + 'screening': { + 'run_id': screening_run_id, + 'answer': screening_answer or None, + 'confidence': screening_confidence, + 'rationale': screening_parsed.rationale or None, + 'parse_ok': False, + 'partial': True, + 'missing_fields': screening_missing_fields, + 'evidence_sentences': screening_parsed.evidence_sentences, + 'evidence_tables': screening_parsed.evidence_tables, + 'evidence_figures': screening_parsed.evidence_figures, + }, + 'critical': None, + }) + continue + # 2) critical critical_additions = '' try: @@ -2026,7 +2099,7 @@ async def _call_with_metadata(prompt: str): llm_col = snake_case_column(q) llm_payload = { 'selected': screening_answer, - 'confidence': screening_parsed.confidence, + 'confidence': screening_confidence, 'explanation': screening_parsed.rationale or '', 'evidence_sentences': screening_parsed.evidence_sentences or [], 'evidence_tables': screening_parsed.evidence_tables or [], @@ -2034,6 +2107,9 @@ async def _call_with_metadata(prompt: str): 'source': 'agentic', 'pipeline': 'fulltext', 'fulltext_md5': fulltext_md5, + 'partial': bool(screening_missing_fields), + 'repair_failed': screening_repair_failed, + 'missing_fields': screening_missing_fields, } await run_in_threadpool( cits_dp_service.update_jsonb_column, @@ -2052,9 +2128,11 @@ async def _call_with_metadata(prompt: str): 'screening': { 'run_id': screening_run_id, 'answer': screening_answer, - 'confidence': screening_parsed.confidence, + 'confidence': screening_confidence, 'rationale': screening_parsed.rationale, - 'parse_ok': screening_parsed.parse_ok, + 'parse_ok': not screening_missing_fields, + 'partial': bool(screening_missing_fields), + 'missing_fields': screening_missing_fields, 'evidence_sentences': screening_parsed.evidence_sentences, 'evidence_tables': screening_parsed.evidence_tables, 'evidence_figures': screening_parsed.evidence_figures, diff --git a/backend/api/services/cit_db_service.py b/backend/api/services/cit_db_service.py index f3a6ecba..5a9ab5bc 100644 --- a/backend/api/services/cit_db_service.py +++ b/backend/api/services/cit_db_service.py @@ -1734,8 +1734,8 @@ def attach_fulltext_atomic( 'fulltext','fulltext_coords','fulltext_pages', 'fulltext_figures','fulltext_tables', 'llm_l2_decision','human_l2_decision' - ) OR column_name LIKE 'llm_param_%' - OR column_name LIKE 'human_param_%')""", + ) OR column_name LIKE 'llm_param_%%' + OR column_name LIKE 'human_param_%%')""", (table_name,), ) clear_columns = [ diff --git a/backend/api/services/fulltext_attachment_service.py b/backend/api/services/fulltext_attachment_service.py index 3d6af2c8..7037d89a 100644 --- a/backend/api/services/fulltext_attachment_service.py +++ b/backend/api/services/fulltext_attachment_service.py @@ -5,12 +5,13 @@ import os import re from dataclasses import dataclass +from typing import Any from fastapi.concurrency import run_in_threadpool from .cit_db_service import cits_dp_service -from .storage import storage_service from .postgres_auth import postgres_server +from .storage import storage_service MAX_PDF_BYTES = int(os.getenv('PDF_LINKAGE_MAX_BYTES', str(50 * 1024 * 1024))) @@ -23,6 +24,295 @@ class AttachmentResult: document_id: str | None = None +def _ensure_document_table() -> None: + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + """CREATE TABLE IF NOT EXISTS citation_fulltext_documents ( + table_name TEXT NOT NULL, + citation_id BIGINT NOT NULL, + document_id TEXT NOT NULL, + filename TEXT NOT NULL, + storage_path TEXT NOT NULL, + file_md5 TEXT NOT NULL, + document_type TEXT NOT NULL DEFAULT 'supplementary', + is_active BOOLEAN NOT NULL DEFAULT FALSE, + extracted_text TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (table_name, citation_id, document_id) + )""", + ) + conn.commit() + + +def list_fulltext_documents(citation_id: int, table_name: str) -> list[dict]: + _ensure_document_table() + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + """SELECT document_id, filename, storage_path, file_md5, document_type, + is_active, extracted_text IS NOT NULL, created_at + FROM citation_fulltext_documents + WHERE table_name=%s AND citation_id=%s + ORDER BY is_active DESC, created_at""", + (table_name, int(citation_id)), + ) + return [ + { + 'document_id': row[0], 'filename': row[1], 'storage_path': row[2], + 'file_md5': row[3], 'document_type': row[4], 'is_active': bool(row[5]), + 'is_extracted': bool(row[6]), 'created_at': row[7].isoformat() if row[7] else None, + } + for row in (cur.fetchall() or []) + ] + + +def ensure_legacy_document(citation_id: int, table_name: str, row: dict) -> None: + """Backfill the registry for PDFs attached before multi-document support.""" + path = str(row.get('fulltext_url') or '') + file_md5 = str(row.get('fulltext_md5') or '') + if not path: + return + _ensure_document_table() + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + 'SELECT 1 FROM citation_fulltext_documents WHERE table_name=%s AND citation_id=%s AND storage_path=%s', + (table_name, int(citation_id), path), + ) + if cur.fetchone(): + return + tail = path.rsplit('/', 1)[-1] + document_id, _, filename = tail.partition('_') + _register_document( + citation_id, table_name, document_id or f'legacy-{citation_id}', + filename or tail or f'fulltext_{citation_id}.pdf', path, + file_md5 or 'unknown', 'main', True, + ) + + +def _register_document( + citation_id: int, table_name: str, document_id: str, filename: str, + storage_path: str, file_md5: str, document_type: str, is_active: bool, +) -> None: + _ensure_document_table() + conn = postgres_server.conn + cur = conn.cursor() + if is_active: + cur.execute( + 'UPDATE citation_fulltext_documents SET is_active=FALSE WHERE table_name=%s AND citation_id=%s', + (table_name, int(citation_id)), + ) + cur.execute( + """INSERT INTO citation_fulltext_documents + (table_name, citation_id, document_id, filename, storage_path, file_md5, document_type, is_active) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s) + ON CONFLICT (table_name, citation_id, document_id) DO UPDATE SET + filename=EXCLUDED.filename, storage_path=EXCLUDED.storage_path, + file_md5=EXCLUDED.file_md5, document_type=EXCLUDED.document_type, + is_active=EXCLUDED.is_active""", + ( + table_name, int(citation_id), document_id, filename, + storage_path, file_md5, document_type, is_active, + ), + ) + conn.commit() + + +async def add_supplementary_document( + *, citation_id: int, table_name: str, user_id: str, filename: str, content: bytes, +) -> AttachmentResult: + file_md5 = validate_pdf(content) + safe_name = os.path.basename( + filename, + ) or f'supplementary_{citation_id}.pdf' + document_id = await storage_service.upload_user_document( + user_id=user_id, filename=safe_name, file_content=content, + ) + if not document_id: + raise RuntimeError('storage_upload_failed') + path = f'{storage_service.container_name}/users/{user_id}/documents/{document_id}_{safe_name}' + try: + await run_in_threadpool( + _register_document, citation_id, table_name, str( + document_id, + ), safe_name, + path, file_md5, 'supplementary', False, + ) + except Exception: + await _delete_path(path) + raise + return AttachmentResult(True, 'linked', path, str(document_id)) + + +def activate_fulltext_document(citation_id: int, table_name: str, document_id: str) -> bool: + _ensure_document_table() + conn = postgres_server.conn + try: + cur = conn.cursor() + cur.execute( + """SELECT storage_path, file_md5 FROM citation_fulltext_documents + WHERE table_name=%s AND citation_id=%s AND document_id=%s FOR UPDATE""", + (table_name, int(citation_id), document_id), + ) + row = cur.fetchone() + if not row: + conn.rollback() + return False + cur.execute( + 'UPDATE citation_fulltext_documents SET is_active=FALSE WHERE table_name=%s AND citation_id=%s', + (table_name, int(citation_id)), + ) + cur.execute( + 'UPDATE citation_fulltext_documents SET is_active=TRUE WHERE table_name=%s AND citation_id=%s AND document_id=%s', + (table_name, int(citation_id), document_id), + ) + # The legacy columns remain the compatibility contract for viewer/extraction. + cur.execute( + f'''UPDATE "{table_name}" SET fulltext_url=%s, fulltext_md5=%s, + fulltext=NULL, fulltext_coords=NULL, fulltext_pages=NULL, + fulltext_figures=NULL, fulltext_tables=NULL WHERE id=%s''', + (row[0], row[1], int(citation_id)), + ) + conn.commit() + return True + except Exception: + conn.rollback() + raise + + +def record_active_extracted_text(citation_id: int, table_name: str, text: str) -> None: + _ensure_document_table() + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + """UPDATE citation_fulltext_documents SET extracted_text=%s + WHERE table_name=%s AND citation_id=%s AND is_active=TRUE""", + (text, table_name, int(citation_id)), + ) + conn.commit() + + +_NUMBERED_SENTENCE_RE = re.compile( + r'(?m)^\s*\[(\d+)\]\s*(.*?)(?=\n\s*\[\d+\]\s|\Z)', re.DOTALL, +) + + +def format_combined_fulltext(rows: list[tuple[str, str, str]], fallback: str) -> str: + """Build document boundaries and globally unique evidence sentence indices.""" + if not rows: + return fallback + next_index = 0 + documents: list[str] = [] + for name, kind, text in rows: + sentences = [ + match.group(2).strip() + for match in _NUMBERED_SENTENCE_RE.finditer(text or '') + ] + if sentences: + numbered = [] + for sentence in sentences: + numbered.append(f'[{next_index}] {sentence}') + next_index += 1 + body = '\n\n'.join(numbered) + else: + body = str(text or '').strip() + documents.append(f'=== DOCUMENT: {name} ({kind}) ===\n{body}') + return '\n\n'.join(documents) + + +def get_fulltext_document(citation_id: int, table_name: str, document_id: str) -> dict[str, Any] | None: + """Resolve a document only inside an already-authorized citation scope.""" + _ensure_document_table() + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + """SELECT document_id, filename, storage_path, file_md5, document_type, + is_active, extracted_text IS NOT NULL + FROM citation_fulltext_documents + WHERE table_name=%s AND citation_id=%s AND document_id=%s""", + (table_name, int(citation_id), document_id), + ) + row = cur.fetchone() + if not row: + return None + return { + 'document_id': row[0], 'filename': row[1], 'storage_path': row[2], + 'file_md5': row[3], 'document_type': row[4], 'is_active': bool(row[5]), + 'is_extracted': bool(row[6]), + } + + +def list_registered_storage_paths(table_name: str) -> list[str]: + """List all attachment blobs for review cleanup, including supplements.""" + _ensure_document_table() + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + 'SELECT DISTINCT storage_path FROM citation_fulltext_documents WHERE table_name=%s', + (table_name,), + ) + return [str(row[0]) for row in (cur.fetchall() or []) if row and row[0]] + + +def delete_document_registry(table_name: str) -> int: + _ensure_document_table() + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + 'DELETE FROM citation_fulltext_documents WHERE table_name=%s', ( + table_name, + ), + ) + deleted = cur.rowcount or 0 + conn.commit() + return deleted + + +def combined_fulltext(citation_id: int, table_name: str, fallback: str) -> str: + _ensure_document_table() + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + """SELECT filename, document_type, extracted_text + FROM citation_fulltext_documents + WHERE table_name=%s AND citation_id=%s AND extracted_text IS NOT NULL + ORDER BY is_active DESC, created_at""", + (table_name, int(citation_id)), + ) + rows = cur.fetchall() or [] + return format_combined_fulltext(rows, fallback) + + +async def delete_fulltext_document(citation_id: int, table_name: str, document_id: str) -> tuple[bool, bool]: + _ensure_document_table() + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + """SELECT storage_path, is_active FROM citation_fulltext_documents + WHERE table_name=%s AND citation_id=%s AND document_id=%s""", + (table_name, int(citation_id), document_id), + ) + row = cur.fetchone() + if not row: + return False, False + path, was_active = str(row[0]), bool(row[1]) + cur.execute( + 'DELETE FROM citation_fulltext_documents WHERE table_name=%s AND citation_id=%s AND document_id=%s', + (table_name, int(citation_id), document_id), + ) + if was_active: + cur.execute( + f'''UPDATE "{table_name}" SET fulltext_url=NULL, fulltext_md5=NULL, + fulltext=NULL, fulltext_coords=NULL, fulltext_pages=NULL, + fulltext_figures=NULL, fulltext_tables=NULL WHERE id=%s''', + (int(citation_id),), + ) + conn.commit() + await _delete_path(path) + return True, was_active + + def validate_pdf(content: bytes) -> str: if not content or len(content) > MAX_PDF_BYTES: raise ValueError('invalid_pdf_size') @@ -96,7 +386,9 @@ def pending() -> list[str]: def forget(path: str) -> None: conn = postgres_server.conn cur = conn.cursor() - cur.execute('DELETE FROM fulltext_blob_cleanup WHERE storage_path=%s', (path,)) + cur.execute( + 'DELETE FROM fulltext_blob_cleanup WHERE storage_path=%s', (path,), + ) conn.commit() removed = 0 @@ -149,7 +441,22 @@ async def attach_fulltext_document( if not result.get('attached'): await _delete_path(path) return AttachmentResult(False, str(result.get('reason') or 'not_attached')) + await run_in_threadpool( + _register_document, citation_id, table_name, str( + document_id, + ), safe_name, + path, file_md5, 'main', True, + ) old_url = result.get('old_url') if old_url and old_url != path: + def forget_replaced() -> None: + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + 'DELETE FROM citation_fulltext_documents WHERE table_name=%s AND citation_id=%s AND storage_path=%s', + (table_name, int(citation_id), str(old_url)), + ) + conn.commit() + await run_in_threadpool(forget_replaced) await _delete_path(str(old_url)) - return AttachmentResult(True, 'linked', path, str(document_id)) \ No newline at end of file + return AttachmentResult(True, 'linked', path, str(document_id)) diff --git a/backend/deploy.sh b/backend/deploy.sh index d9bddbf0..aac3e25a 100755 --- a/backend/deploy.sh +++ b/backend/deploy.sh @@ -130,7 +130,7 @@ fi # Build images if requested if [ "$BUILD" = true ]; then echo -e "${BLUE}🔨 Building Docker images...${NC}" - docker compose build --no-cache + docker compose build echo -e "${GREEN}✅ Images built successfully${NC}" fi diff --git a/backend/tests/fulltext_attachment_service_test.py b/backend/tests/fulltext_attachment_service_test.py new file mode 100644 index 00000000..c4cc4c0f --- /dev/null +++ b/backend/tests/fulltext_attachment_service_test.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from api.services.fulltext_attachment_service import format_combined_fulltext +from api.services.fulltext_attachment_service import validate_pdf + + +def test_validate_pdf_accepts_pdf_signature_after_whitespace(): + digest = validate_pdf(b'\n\t%PDF-1.7\nbody') + assert len(digest) == 32 + + +def test_validate_pdf_rejects_non_pdf_bytes(): + try: + validate_pdf(b'not a pdf') + except ValueError as exc: + assert str(exc) == 'invalid_pdf' + else: + raise AssertionError('Expected invalid PDF to be rejected') + + +def test_combined_fulltext_has_unambiguous_document_boundaries(): + result = format_combined_fulltext( + [ + ('study.pdf', 'main', '[0] Main study methods.'), + ( + 'appendix.pdf', 'supplementary', + '[0] Supplementary outcome table.', + ), + ], + 'fallback', + ) + + assert '=== DOCUMENT: study.pdf (main) ===' in result + assert '=== DOCUMENT: appendix.pdf (supplementary) ===' in result + assert result.index('study.pdf') < result.index('appendix.pdf') + assert '[0] Main study methods.' in result + assert '[1] Supplementary outcome table.' in result + + +def test_combined_fulltext_reindexes_each_documents_local_sentence_ids(): + result = format_combined_fulltext( + [ + ('main.pdf', 'main', '[0] First.\n\n[1] Second.'), + ('supp.pdf', 'supplementary', '[0] Third.\n\n[1] Fourth.'), + ], + 'fallback', + ) + + assert result.count('[0]') == 1 + assert '[2] Third.' in result + assert '[3] Fourth.' in result + + +def test_combined_fulltext_uses_legacy_fallback_without_extracted_documents(): + assert format_combined_fulltext( + [], '[0] Legacy text.', + ) == '[0] Legacy text.' diff --git a/backend/tests/screen_agentic_utils_test.py b/backend/tests/screen_agentic_utils_test.py index 24be87f4..4ce754af 100644 --- a/backend/tests/screen_agentic_utils_test.py +++ b/backend/tests/screen_agentic_utils_test.py @@ -60,6 +60,8 @@ async def call_llm(prompt): assert raw.endswith('') assert len(prompts) == 2 assert 'Previous response:' in prompts[1] + assert 'Original task and allowed options:' in prompts[1] + assert 'original prompt' in prompts[1] @pytest.mark.asyncio @@ -67,7 +69,13 @@ async def test_screening_raises_after_failed_repair(): async def call_llm(_prompt): return 'Yes0.8', None - with pytest.raises(AgentResponseError, match='rationale'): + with pytest.raises(AgentResponseError, match='rationale') as exc_info: await call_and_parse_agent_response( 'original prompt', stage='screening', call_llm=call_llm, ) + + assert exc_info.value.missing_fields == ['rationale'] + assert exc_info.value.parsed is not None + assert exc_info.value.parsed.answer == 'Yes' + assert exc_info.value.parsed.confidence == 0.8 + assert exc_info.value.raw_response.startswith('Yes') diff --git a/frontend/app/[lang]/can-sr/l2-screen/view/page.tsx b/frontend/app/[lang]/can-sr/l2-screen/view/page.tsx index 63d6f0c7..fa76e100 100644 --- a/frontend/app/[lang]/can-sr/l2-screen/view/page.tsx +++ b/frontend/app/[lang]/can-sr/l2-screen/view/page.tsx @@ -896,7 +896,12 @@ export default function CanSrL2ScreenViewPage() { srId={srId || ''} citationId={citationId ?? ''} conversionId={null} - fileName={"Fulltext"} + fileName={String( + citation?.title || + citation?.citation || + citation?.article_title || + 'Full text' + )} coords={fulltextCoords || []} pages={fulltextPages || []} aiPanels={panelsKeyed.panels} @@ -1050,6 +1055,18 @@ export default function CanSrL2ScreenViewPage() { const displayExplanationRaw = (scrRationale || aiExpl || '').trim() const displayExplanation = extractXmlTag(displayExplanationRaw, 'rationale') || displayExplanationRaw + const screeningGuardrails = parseObject((scr as any)?.guardrails) + const missingFields = Array.isArray(aiData?.missing_fields) + ? aiData.missing_fields.map((field: unknown) => String(field)) + : Array.isArray(screeningGuardrails?.missing_fields) + ? screeningGuardrails.missing_fields.map((field: unknown) => String(field)) + : [] + const missingLabels = missingFields.map((field: string) => { + if (field === 'answer') return 'selection' + if (field === 'confidence') return 'confidence' + if (field === 'rationale') return 'rationale' + return field + }) return (
+ {missingLabels.length > 0 ? ( +
+ The AI response was incomplete after one automatic repair. Missing:{' '} + {missingLabels.join(', ')}. Review the available fields and rerun this criterion. +
+ ) : null}
{dict.screening.confidence}{' '} - {Number.isFinite(displayConfidence) + {missingFields.includes('confidence') + ? 'Not returned by AI' + : Number.isFinite(displayConfidence) ? String(displayConfidence) : String(aiData.confidence ?? '')}
{dict.screening.explanation}
- {displayExplanation || dict.screening.noExplanation} + {missingFields.includes('rationale') + ? 'Rationale was not returned by the AI after repair.' + : displayExplanation || dict.screening.noExplanation}
diff --git a/frontend/app/api/can-sr/citations/full-text/route.ts b/frontend/app/api/can-sr/citations/full-text/route.ts index 1230a499..19399dc0 100644 --- a/frontend/app/api/can-sr/citations/full-text/route.ts +++ b/frontend/app/api/can-sr/citations/full-text/route.ts @@ -16,9 +16,8 @@ import { BACKEND_URL } from '@/lib/config' * -> stream the stored PDF associated with the citation (calls backend citation endpoint to resolve document_id/storage_path, * then proxies to backend download endpoint or streams the storage URL) * - * - GET /api/can-sr/citations/full-text?document_id= - * -> stream the document directly from backend download endpoint: - * GET {BACKEND_URL}/api/files/documents/{document_id}/download + * - GET with sr_id, citation_id, and document_id streams a citation-scoped + * attachment after backend membership and linkage checks. * * Authentication: forwards Authorization header for all backend calls (required). */ @@ -56,6 +55,7 @@ export async function GET(request: NextRequest) { const srId = params.get('sr_id') const citationId = params.get('citation_id') const documentIdParam = params.get('document_id') + const action = params.get('action') const authHeader = request.headers.get('authorization') if (!authHeader) { @@ -65,6 +65,15 @@ export async function GET(request: NextRequest) { ) } + if (action === 'documents' && srId && citationId) { + const res = await fetch( + `${BACKEND_URL}/api/cite/${encodeURIComponent(srId)}/citations/${encodeURIComponent(citationId)}/fulltext-documents`, + { headers: { Authorization: authHeader }, cache: 'no-store' }, + ) + const data = await res.json().catch(() => ({})) + return NextResponse.json(data, { status: res.status }) + } + let fileFetchUrl: string | null = null const fetchOptions: Record = { method: 'GET', @@ -73,11 +82,15 @@ export async function GET(request: NextRequest) { }, } - if (documentIdParam) { - // Prefer direct document_id -> backend files download route - fileFetchUrl = `${BACKEND_URL}/api/files/documents/${encodeURIComponent( - documentIdParam, - )}/download` + if (documentIdParam && srId && citationId) { + // Resolve inside the authorized SR/citation scope. This permits review + // collaborators to view attachments uploaded by another member. + fileFetchUrl = `${BACKEND_URL}/api/cite/${encodeURIComponent(srId)}/citations/${encodeURIComponent(citationId)}/fulltext-documents/${encodeURIComponent(documentIdParam)}/download` + } else if (documentIdParam) { + return NextResponse.json( + { error: 'sr_id and citation_id are required with document_id' }, + { status: 400 }, + ) } else if (srId && citationId) { // Fetch the citation row to locate document_id or storage_path const citationUrl = `${BACKEND_URL}/api/cite/${encodeURIComponent( @@ -241,8 +254,8 @@ export async function POST(request: NextRequest) { try { const params = request.nextUrl.searchParams const srId = params.get('sr_id') + const action = params.get('action') const citationId = params.get('citation_id') - const action = params.get('action') // optional; if 'extract' will call backend extract endpoint if (!srId || !citationId) { return NextResponse.json( @@ -252,6 +265,23 @@ export async function POST(request: NextRequest) { } const authHeader = request.headers.get('authorization') + if (!authHeader) { + return NextResponse.json({ error: 'Authorization header is required' }, { status: 401 }) + } + + if (action === 'activate') { + const payload = await request.json() + const res = await fetch( + `${BACKEND_URL}/api/cite/${encodeURIComponent(srId)}/citations/${encodeURIComponent(citationId)}/fulltext-documents/activate`, + { + method: 'POST', + headers: { Authorization: authHeader, 'Content-Type': 'application/json' }, + body: JSON.stringify(payload), + }, + ) + const data = await res.json().catch(() => ({})) + return NextResponse.json(data, { status: res.status }) + } // If action=extract, call backend extract endpoint which will download PDF from storage and run Grobid if (action === 'extract') { @@ -259,13 +289,6 @@ export async function POST(request: NextRequest) { srId, )}/citations/${encodeURIComponent(citationId)}/extract-fulltext` - if (!authHeader) { - return NextResponse.json( - { error: 'Authorization header is required' }, - { status: 401 }, - ) - } - const res = await fetch(url, { method: 'POST', headers: { @@ -297,16 +320,10 @@ export async function POST(request: NextRequest) { const backendForm = new FormData() backendForm.append('file', file) + const mode = params.get('mode') || 'replace' const url = `${BACKEND_URL}/api/cite/${encodeURIComponent( srId, - )}/citations/${encodeURIComponent(citationId)}/upload-fulltext` - - if (!authHeader) { - return NextResponse.json( - { error: 'Authorization header is required' }, - { status: 401 }, - ) - } + )}/citations/${encodeURIComponent(citationId)}/upload-fulltext?mode=${encodeURIComponent(mode)}` const res = await fetch(url, { method: 'POST', @@ -333,3 +350,21 @@ export async function POST(request: NextRequest) { ) } } + +export async function DELETE(request: NextRequest) { + const params = request.nextUrl.searchParams + const srId = params.get('sr_id') + const citationId = params.get('citation_id') + const documentId = params.get('document_id') + const authHeader = request.headers.get('authorization') + if (!authHeader) return NextResponse.json({ error: 'Authorization header is required' }, { status: 401 }) + if (!srId || !citationId || !documentId) { + return NextResponse.json({ error: 'sr_id, citation_id and document_id are required' }, { status: 400 }) + } + const res = await fetch( + `${BACKEND_URL}/api/cite/${encodeURIComponent(srId)}/citations/${encodeURIComponent(citationId)}/fulltext-documents/${encodeURIComponent(documentId)}`, + { method: 'DELETE', headers: { Authorization: authHeader } }, + ) + const data = await res.json().catch(() => ({})) + return NextResponse.json(data, { status: res.status }) +} diff --git a/frontend/components/can-sr/PDFBoundingBoxViewer.tsx b/frontend/components/can-sr/PDFBoundingBoxViewer.tsx index 2994eb2d..ef536233 100644 --- a/frontend/components/can-sr/PDFBoundingBoxViewer.tsx +++ b/frontend/components/can-sr/PDFBoundingBoxViewer.tsx @@ -25,6 +25,14 @@ interface PDFBoundingBoxViewerProps { defaultFitToWidth?: boolean } +type FulltextDocument = { + document_id: string + filename: string + document_type: string + is_active: boolean + is_extracted: boolean +} + /** * PDFBoundingBoxViewer * - Renders all pages stacked vertically using pdf.js @@ -105,7 +113,12 @@ const PDFBoundingBoxViewer = forwardRef(null) const [pdfMissing, setPdfMissing] = useState(false) const [uploading, setUploading] = useState(false) + const [processingDocumentId, setProcessingDocumentId] = useState(null) const [reloadKey, setReloadKey] = useState(0) + const [documents, setDocuments] = useState([]) + const [documentsLoaded, setDocumentsLoaded] = useState(false) + const [viewedDocumentId, setViewedDocumentId] = useState(null) + const [uploadMode, setUploadMode] = useState<'replace' | 'supplementary'>('replace') const uploadInputRef = useRef(null) const pageRefs = useRef<{ [key: number]: HTMLCanvasElement | null }>({}) const containerRef = useRef(null) @@ -123,6 +136,35 @@ const PDFBoundingBoxViewer = forwardRef => { + const token = getAuthToken() + const tokenType = getTokenType() + return token ? { Authorization: `${tokenType} ${token}` } : {} + } + + useEffect(() => { + setDocumentsLoaded(false) + const refreshDocuments = async () => { + const res = await fetch( + `/api/can-sr/citations/full-text?action=documents&sr_id=${encodeURIComponent(String(srId))}&citation_id=${encodeURIComponent(String(citationId))}`, + { headers: authHeaders(), cache: 'no-store' }, + ) + if (!res.ok) { + setDocumentsLoaded(true) + return + } + const payload = await res.json() + const nextDocuments = Array.isArray(payload?.documents) ? payload.documents : [] + setDocuments(nextDocuments) + setViewedDocumentId((current) => { + if (current && nextDocuments.some((document: FulltextDocument) => document.document_id === current)) return current + return nextDocuments.find((document: FulltextDocument) => document.is_active)?.document_id || null + }) + setDocumentsLoaded(true) + } + void refreshDocuments() + }, [srId, citationId, reloadKey]) + useImperativeHandle(ref, () => ({ scrollToPage: (pageNum: number) => { const container = containerRef.current @@ -213,6 +255,7 @@ const PDFBoundingBoxViewer = forwardRef { + if (!documentsLoaded) return setLoading(true) setError(null) setPdfMissing(false) @@ -220,14 +263,12 @@ const PDFBoundingBoxViewer = forwardRef = {} - if (token) headers['Authorization'] = `${tokenType} ${token}` + const headers: Record = authHeaders() - const url = `/api/can-sr/citations/full-text?sr_id=${encodeURIComponent(String(srId))}&citation_id=${encodeURIComponent( - String(citationId), - )}` + const selectedDocument = documents.find((document) => document.document_id === viewedDocumentId) + const url = selectedDocument + ? `/api/can-sr/citations/full-text?sr_id=${encodeURIComponent(String(srId))}&citation_id=${encodeURIComponent(String(citationId))}&document_id=${encodeURIComponent(selectedDocument.document_id)}` + : `/api/can-sr/citations/full-text?sr_id=${encodeURIComponent(String(srId))}&citation_id=${encodeURIComponent(String(citationId))}` const res = await fetch(url, { headers, cache: 'no-store' }) if (res.status === 404) { @@ -254,7 +295,7 @@ const PDFBoundingBoxViewer = forwardRef ({})) throw new Error(payload?.detail || payload?.error || `${dict.pdf.uploadFailed} (${res.status})`) } - setReloadKey((value) => value + 1) + if (uploadMode === 'replace') window.location.reload() + else setReloadKey((value) => value + 1) } catch (err: any) { setError(err?.message || dict.pdf.uploadFailed) } finally { @@ -311,6 +353,41 @@ const PDFBoundingBoxViewer = forwardRef { + setProcessingDocumentId(documentId) + setError(null) + try { + const activateRes = await fetch( + `/api/can-sr/citations/full-text?action=activate&sr_id=${encodeURIComponent(String(srId))}&citation_id=${encodeURIComponent(String(citationId))}`, + { method: 'POST', headers: { ...authHeaders(), 'Content-Type': 'application/json' }, body: JSON.stringify({ document_id: documentId }) }, + ) + if (!activateRes.ok) throw new Error('Failed to select PDF for processing') + const extractRes = await fetch( + `/api/can-sr/citations/full-text?action=extract&sr_id=${encodeURIComponent(String(srId))}&citation_id=${encodeURIComponent(String(citationId))}`, + { method: 'POST', headers: authHeaders() }, + ) + if (!extractRes.ok) { + const payload = await extractRes.json().catch(() => ({})) + throw new Error(payload?.detail || payload?.error || 'PDF text extraction failed') + } + window.location.reload() + } catch (err: any) { + setError(err?.message || 'PDF processing failed') + setProcessingDocumentId(null) + } + } + + const deleteDocument = async (document: FulltextDocument) => { + if (!window.confirm(`Delete ${document.filename}?`)) return + const res = await fetch( + `/api/can-sr/citations/full-text?sr_id=${encodeURIComponent(String(srId))}&citation_id=${encodeURIComponent(String(citationId))}&document_id=${encodeURIComponent(document.document_id)}`, + { method: 'DELETE', headers: authHeaders() }, + ) + if (!res.ok) return setError('Failed to delete PDF') + if (document.is_active) window.location.reload() + else setReloadKey((value) => value + 1) + } + // Load PDF document via pdf.js useEffect(() => { const token = ++documentTokenRef.current @@ -489,6 +566,10 @@ const PDFBoundingBoxViewer = forwardRef { + const viewedDocument = documents.find((document) => document.document_id === viewedDocumentId) + // Coordinates and sentence IDs on the citation row belong to the active + // processed PDF. Never draw them over a different PDF selected only for viewing. + if (viewedDocument && !viewedDocument.is_active) return null if (!coords || !Array.isArray(coords) || coords.length === 0) return null const vp = pageViewports[pageNum] const dims = pages?.[pageNum - 1] @@ -715,6 +796,26 @@ const PDFBoundingBoxViewer = forwardRef : null}
+
+ {documents.map((document) => ( +
+ + {document.is_extracted ? 'Processed' : 'Needs processing'} + {!document.is_extracted ? ( + + ) : null} + +
+ ))} + + + +
+
{!pdfMissing && !error ?
@@ -772,13 +873,6 @@ const PDFBoundingBoxViewer = forwardRef {uploading ? dict.pdf.uploadingPDF : dict.pdf.uploadPDF} -
) : error ? (