diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 88a31a2c..bc58c52c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,7 +20,7 @@ jobs: uses: actions/checkout@v6 - name: Set up Node.js - uses: actions/setup-node@v6 + uses: actions/setup-node@v7 with: node-version: '20' cache: 'npm' @@ -51,7 +51,7 @@ jobs: uses: actions/checkout@v6 - name: Set up Python - uses: actions/setup-python@v6 + uses: actions/setup-python@v7 with: python-version: '3.10' 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/criteria/README.md b/backend/api/criteria/README.md new file mode 100644 index 00000000..d9e93054 --- /dev/null +++ b/backend/api/criteria/README.md @@ -0,0 +1,42 @@ +# Criteria configuration backend contract + +This module is the Phase 1 foundation for the extended criteria UI. It introduces +canonical criteria configuration without replacing existing screening consumers. + +## Canonical storage + +- `criteria`: strict `schema_version: 2` JSON. +- `criteria_yaml`: deterministic YAML generated from the same model. +- `criteria_parsed`: compatibility projection for current screening and extraction + code, plus stable `items` with IDs. +- `criteria_revision`: optimistic concurrency token supplied as `expected_revision`. + +All four values are updated atomically by the canonical save operation. + +## API + +| Method | Path | Purpose | +| --- | --- | --- | +| `GET` | `/sr/{sr_id}/criteria-config` | Canonical config or read-only legacy migration preview. | +| `POST` | `/sr/{sr_id}/criteria-config/validate` | Validate canonical JSON. | +| `POST` | `/sr/{sr_id}/criteria-config/import-yaml` | Parse v2 or legacy YAML without saving. | +| `PUT` | `/sr/{sr_id}/criteria-config` | Save with revision and invalidation guards. | +| `GET` | `/sr/{sr_id}/criteria-config/export-yaml` | Export deterministic v2 YAML. | + +Validation failures return HTTP 422 with field paths. Stale revisions, +screening-data invalidation, and unconfirmed legacy migration return HTTP 409. + +## Migration and compatibility + +Legacy conversion is deterministic. IDs derive from source text with ordered +collision suffixes. Explicit `(exclude)` or `[exclude]` markers become `exclude`; +other answers default to `include`, with diagnostics for every inference. + +Load and import operations never persist migration output. Any write of migration +output that contains inferred semantics must include the exact +`migration_fingerprint` returned by the current preview. This applies to review +creation, canonical save, and the compatibility YAML update endpoint. + +Triggers may reference only earlier questions or selection parameters and must +target one of that source's options. Text parameters cannot be trigger sources. +The existing explicit `force` guard remains required when screening data exists. diff --git a/backend/api/criteria/__init__.py b/backend/api/criteria/__init__.py new file mode 100644 index 00000000..107f0d5d --- /dev/null +++ b/backend/api/criteria/__init__.py @@ -0,0 +1,7 @@ +"""Canonical criteria configuration domain.""" +from __future__ import annotations + +from .models import CriteriaConfigV2 +from .service import CriteriaConfigurationService + +__all__ = ['CriteriaConfigV2', 'CriteriaConfigurationService'] diff --git a/backend/api/criteria/models.py b/backend/api/criteria/models.py new file mode 100644 index 00000000..390c4eaa --- /dev/null +++ b/backend/api/criteria/models.py @@ -0,0 +1,171 @@ +"""Strict version 2 criteria configuration models and domain validation.""" +from __future__ import annotations + +from enum import Enum +from typing import Annotated +from typing import Literal + +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field +from pydantic import model_validator +from pydantic import StringConstraints + +CriteriaId = Annotated[ + str, StringConstraints( + pattern=r'^[a-z][a-z0-9_-]{2,63}$', + ), +] +NonEmptyText = Annotated[ + str, StringConstraints( + strip_whitespace=True, min_length=1, max_length=10000, + ), +] + + +class StrictModel(BaseModel): + model_config = ConfigDict(extra='forbid') + + +class Decision(str, Enum): + INCLUDE = 'include' + EXCLUDE = 'exclude' + + +class TriggerCondition(StrictModel): + source_item_id: CriteriaId + option_id: CriteriaId + + +class Trigger(StrictModel): + all: list[TriggerCondition] = Field(default_factory=list, max_length=20) + + @model_validator(mode='after') + def conditions_are_unique(self) -> Trigger: + pairs = [ + (condition.source_item_id, condition.option_id) + for condition in self.all + ] + if len(pairs) != len(set(pairs)): + raise ValueError('duplicate trigger conditions are not allowed') + return self + + +class CitationFields(StrictModel): + l1_include: list[NonEmptyText] = Field( + default_factory=list, max_length=100, + ) + doi: NonEmptyText | None = None + + @model_validator(mode='after') + def include_fields_are_unique(self) -> CitationFields: + if len(self.l1_include) != len(set(self.l1_include)): + raise ValueError( + 'citation_fields.l1_include must not contain duplicates', + ) + return self + + +class ScreeningAnswer(StrictModel): + id: CriteriaId + label: NonEmptyText + context: str | None = Field(default=None, max_length=10000) + decision: Decision + + +class ScreeningQuestion(StrictModel): + id: CriteriaId + question: NonEmptyText + answers: list[ScreeningAnswer] = Field(min_length=2, max_length=50) + trigger: Trigger = Field(default_factory=Trigger) + + @model_validator(mode='after') + def answer_ids_are_unique(self) -> ScreeningQuestion: + ids = [answer.id for answer in self.answers] + if len(ids) != len(set(ids)): + raise ValueError('answer IDs must be unique within a question') + return self + + +class ParameterOption(StrictModel): + id: CriteriaId + label: NonEmptyText + context: str | None = Field(default=None, max_length=10000) + + +class TextParameter(StrictModel): + id: CriteriaId + name: NonEmptyText + description: NonEmptyText + type: Literal['text'] + unit_instructions: str | None = Field(default=None, max_length=10000) + calculation: str | None = Field(default=None, max_length=10000) + trigger: Trigger = Field(default_factory=Trigger) + legacy_category: str | None = Field(default=None, max_length=1000) + + +class SelectionParameter(StrictModel): + id: CriteriaId + name: NonEmptyText + description: NonEmptyText + type: Literal['selection'] + selection_mode: Literal['single', 'multiple'] + options: list[ParameterOption] = Field(min_length=1, max_length=100) + unit_instructions: str | None = Field(default=None, max_length=10000) + calculation: str | None = Field(default=None, max_length=10000) + trigger: Trigger = Field(default_factory=Trigger) + legacy_category: str | None = Field(default=None, max_length=1000) + + @model_validator(mode='after') + def option_ids_are_unique(self) -> SelectionParameter: + ids = [option.id for option in self.options] + if len(ids) != len(set(ids)): + raise ValueError('option IDs must be unique within a parameter') + return self + + +Parameter = Annotated[ + TextParameter | + SelectionParameter, Field(discriminator='type'), +] + + +class CriteriaConfigV2(StrictModel): + schema_version: Literal[2] = 2 + citation_fields: CitationFields = Field(default_factory=CitationFields) + l1: list[ScreeningQuestion] = Field(default_factory=list, max_length=100) + l2: list[ScreeningQuestion] = Field(default_factory=list, max_length=100) + parameters: list[Parameter] = Field(default_factory=list, max_length=200) + + @model_validator(mode='after') + def validate_global_identity_and_triggers(self) -> CriteriaConfigV2: + items = [*self.l1, *self.l2, *self.parameters] + ids = [item.id for item in items] + if len(ids) != len(set(ids)): + raise ValueError('item IDs must be unique across the review') + + seen: dict[str, set[str] | None] = {} + for item in items: + for condition in item.trigger.all: + if condition.source_item_id not in seen: + raise ValueError( + f'trigger source {condition.source_item_id!r} must reference an earlier item', + ) + options = seen[condition.source_item_id] + if options is None: + raise ValueError( + 'text parameters cannot be trigger sources', + ) + if condition.option_id not in options: + raise ValueError( + f'option {condition.option_id!r} does not belong to trigger source ' + f'{condition.source_item_id!r}', + ) + + if isinstance(item, ScreeningQuestion): + seen[item.id] = {answer.id for answer in item.answers} + elif isinstance(item, SelectionParameter): + seen[item.id] = {option.id for option in item.options} + else: + seen[item.id] = None + return self diff --git a/backend/api/criteria/runtime.py b/backend/api/criteria/runtime.py new file mode 100644 index 00000000..947083bb --- /dev/null +++ b/backend/api/criteria/runtime.py @@ -0,0 +1,54 @@ +"""Runtime helpers for applying canonical criteria to stored citation values.""" +from __future__ import annotations + +from typing import Any + +from ..services.cit_db_service import snake_case_column + + +def _selected(row: dict[str, Any], item: dict[str, Any]) -> str | None: + """Return the latest human/LLM selected label for a criteria item.""" + slug = snake_case_column( + str(item.get('question') or item.get('name') or ''), + ) + prefixes = ('human_param_', 'llm_param_') if item.get( + 'name', + ) else ('human_', 'llm_') + for prefix in prefixes: + value = row.get(f'{prefix}{slug}') + if isinstance(value, str): + return value + if isinstance(value, dict): + for key in ('selected', 'value'): + if isinstance(value.get(key), str): + return value[key] + return None + + +def item_is_visible(criteria: dict[str, Any], item: dict[str, Any], row: dict[str, Any]) -> bool: + """Evaluate an item's AND trigger conditions against a citation row. + + Missing source answers fail closed: a dependent item must not be screened or + extracted before its prerequisite has a matching answer. + """ + items = [ + *(criteria.get('l1') or []), *( + criteria.get('l2') + or [] + ), *(criteria.get('parameters') or []), + ] + by_id = {str(candidate.get('id')): candidate for candidate in items if isinstance(candidate, dict)} + for condition in ((item.get('trigger') or {}).get('all') or []): + source = by_id.get(str(condition.get('source_item_id'))) + if not source: + return False + selected = _selected(row, source) + option = next( + ( + answer for answer in [*(source.get('answers') or []), *(source.get('options') or [])] + if str(answer.get('id')) == str(condition.get('option_id')) + ), None, + ) + if not option or selected != option.get('label'): + return False + return True diff --git a/backend/api/criteria/service.py b/backend/api/criteria/service.py new file mode 100644 index 00000000..4c565021 --- /dev/null +++ b/backend/api/criteria/service.py @@ -0,0 +1,349 @@ +"""Parsing, migration, serialization, and compatibility projection for criteria.""" +from __future__ import annotations + +import hashlib +import json +import re +import unicodedata +from copy import deepcopy +from typing import Any +from typing import Literal + +import yaml +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field + +from .models import CriteriaConfigV2 + +LEGACY_KEYS = {'include', 'criteria', 'l2_criteria', 'parameters'} +LEGACY_ONLY_KEYS = {'include', 'criteria', 'l2_criteria'} +MAX_YAML_BYTES = 1_000_000 + + +class Diagnostic(BaseModel): + model_config = ConfigDict(extra='forbid') + + severity: Literal['info', 'warning'] + code: str + source_path: list[str | int] = Field(default_factory=list) + target_path: list[str | int] = Field(default_factory=list) + message: str + requires_confirmation: bool = False + + +class MigrationStats(BaseModel): + l1: int = 0 + l2: int = 0 + parameters: int = 0 + + +class CriteriaLoadResult(BaseModel): + criteria: CriteriaConfigV2 + source_format: Literal['criteria_v2', 'legacy_yaml_v1'] + diagnostics: list[Diagnostic] = Field(default_factory=list) + requires_confirmation: bool = False + fingerprint: str | None = None + stats: MigrationStats | None = None + + +class _IdAllocator: + def __init__(self) -> None: + self.used: set[str] = set() + + def allocate(self, prefix: str, value: str, fallback: str) -> str: + normalized = unicodedata.normalize( + 'NFKD', value, + ).encode('ascii', 'ignore').decode() + slug = re.sub( + r'[^a-z0-9]+', '_', normalized.lower(), + ).strip('_') or fallback + base = f'{prefix}{slug}'[:63].rstrip('_') + if len(base) < 3: + base = f'{prefix}{fallback}'[:63] + candidate = base + suffix = 2 + while candidate in self.used: + marker = f'_{suffix}' + candidate = f'{base[:64 - len(marker)]}{marker}' + suffix += 1 + self.used.add(candidate) + return candidate + + +class CriteriaConfigurationService: + """Pure canonical criteria operations shared by API and persistence adapters.""" + + def parse_yaml(self, content: str, *, source_kind: str = 'yaml_import') -> CriteriaLoadResult: + if len(content.encode('utf-8')) > MAX_YAML_BYTES: + raise ValueError('criteria YAML exceeds the 1 MB limit') + try: + raw = yaml.safe_load(content) + except yaml.YAMLError as exc: + raise ValueError(f'invalid criteria YAML: {exc}') from exc + if not isinstance(raw, dict): + raise ValueError('criteria YAML root must be a mapping') + return self.normalize(raw, source_kind=source_kind) + + def normalize(self, raw: dict[str, Any], *, source_kind: str = 'backend_load') -> CriteriaLoadResult: + version = raw.get('schema_version') + if version == 2: + mixed = LEGACY_ONLY_KEYS.intersection(raw) + if mixed: + raise ValueError( + f'v2 criteria cannot contain legacy keys: {sorted(mixed)}', + ) + normalized = deepcopy(raw) + diagnostics = self._move_question_context_to_answers(normalized) + return CriteriaLoadResult( + criteria=CriteriaConfigV2.model_validate(normalized), + source_format='criteria_v2', + diagnostics=diagnostics, + ) + if version is not None: + raise ValueError( + f'unsupported criteria schema_version: {version!r}', + ) + if not LEGACY_KEYS.intersection(raw): + raise ValueError('unrecognized criteria format') + return self._migrate_legacy(raw, source_kind=source_kind) + + @staticmethod + def _move_question_context_to_answers(raw: dict[str, Any]) -> list[Diagnostic]: + """Preserve obsolete v2 question guidance by applying it to every answer.""" + diagnostics: list[Diagnostic] = [] + for stage in ('l1', 'l2'): + items = raw.get(stage) + if not isinstance(items, list): + continue + for question_index, item in enumerate(items): + if not isinstance(item, dict) or 'context' not in item: + continue + context = item.pop('context') + if not isinstance(context, str) or not context.strip(): + continue + answers = item.get('answers') + if not isinstance(answers, list): + continue + for answer in answers: + if not isinstance(answer, dict): + continue + answer_context = answer.get('context') + parts = [context.strip()] + if isinstance(answer_context, str) and answer_context.strip(): + parts.append(answer_context.strip()) + answer['context'] = '\n\n'.join(parts) + diagnostics.append( + Diagnostic( + severity='warning', + code='question_context_moved_to_answers', + source_path=[stage, question_index, 'context'], + target_path=[stage, question_index, 'answers'], + message='Question-level context was copied to each answer because screening guidance is answer-specific.', + ), + ) + return diagnostics + + def export_yaml(self, criteria: CriteriaConfigV2 | dict[str, Any]) -> str: + model = criteria if isinstance( + criteria, CriteriaConfigV2, + ) else CriteriaConfigV2.model_validate(criteria) + payload = model.model_dump(mode='json', exclude_none=True) + return yaml.safe_dump(payload, sort_keys=False, allow_unicode=True, width=100) + + def build_compatibility_projection( + self, criteria: CriteriaConfigV2 | dict[str, Any], + ) -> dict[str, Any]: + model = criteria if isinstance( + criteria, CriteriaConfigV2, + ) else CriteriaConfigV2.model_validate(criteria) + + def stage(items): + return { + 'questions': [item.question for item in items], + 'possible_answers': [[answer.label for answer in item.answers] for item in items], + 'additional_infos': [ + '\n\n'.join( + f'For articles matching <{answer.label}> we answer "{answer.label}":\n' + f'<{answer.label}>\n{answer.context or ""}\n' + for answer in item.answers + ) + for item in items + ], + 'items': [item.model_dump(mode='json', exclude_none=True) for item in items], + } + + parameter_items = [ + item.model_dump( + mode='json', exclude_none=True, + ) for item in model.parameters + ] + categories: list[str] = [] + grouped: dict[str, list[Any]] = {} + for item in model.parameters: + category = item.legacy_category or 'Parameters' + if category not in grouped: + categories.append(category) + grouped[category] = [] + grouped[category].append(item) + return { + 'schema_version': 2, + 'l1': {'include': model.citation_fields.l1_include, **stage(model.l1)}, + 'l2': stage(model.l2), + 'parameters': { + 'categories': categories, + 'possible_parameters': [[item.name for item in grouped[category]] for category in categories], + 'descriptions': [ + [f'Parameter {item.name} are described as {item.description}.' for item in grouped[category]] + for category in categories + ], + 'items': parameter_items, + }, + } + + def _migrate_legacy(self, legacy: dict[str, Any], *, source_kind: str) -> CriteriaLoadResult: + # Reserved for source-specific telemetry; migration remains pure. + del source_kind + diagnostics: list[Diagnostic] = [] + item_ids = _IdAllocator() + + include = legacy.get('include', []) + if not isinstance(include, list) or any(not isinstance(value, str) or not value.strip() for value in include): + raise ValueError( + 'legacy include must be a list of non-empty strings', + ) + normalized_include = list( + dict.fromkeys( + value.strip() for value in include + ), + ) + if len(normalized_include) != len(include): + diagnostics.append( + Diagnostic( + severity='warning', code='duplicate_include_fields_removed', + source_path=['include'], target_path=['citation_fields', 'l1_include'], + message='Duplicate citation fields were removed while preserving order.', + ), + ) + + def questions(key: str, stage: str) -> list[dict[str, Any]]: + block = legacy.get(key, {}) + if block is None: + block = {} + if not isinstance(block, dict): + raise ValueError(f'legacy {key} must be a mapping') + output = [] + for question_index, (question, answers) in enumerate(block.items()): + if not isinstance(question, str) or not question.strip() or not isinstance(answers, dict): + raise ValueError( + f'legacy {key} questions and answers must be mappings of non-empty strings', + ) + option_ids = _IdAllocator() + converted_answers = [] + decisions: set[str] = set() + for answer_index, (label, context) in enumerate(answers.items()): + if not isinstance(label, str) or not label.strip() or not isinstance(context, str): + raise ValueError( + f'legacy {key} answer labels and contexts must be strings', + ) + exclude = bool( + re.search( + r'(?i)(?:\(exclude\)|\[exclude\]|^\s*exclude\s*$)', label, + ), + ) + decision = 'exclude' if exclude else 'include' + decisions.add(decision) + diagnostics.append( + Diagnostic( + severity='warning' if exclude else 'info', + code='decision_inferred_exclude' if exclude else 'decision_defaulted_include', + source_path=[key, question, label], + target_path=[ + stage, question_index, + 'answers', answer_index, 'decision', + ], + message=( + 'This answer was marked Exclude from a legacy text marker.' if exclude + else 'This answer defaulted to Include because no legacy exclusion marker was present.' + ), + requires_confirmation=exclude, + ), + ) + converted_answers.append({ + 'id': option_ids.allocate('', label, f'answer_{answer_index + 1}'), + 'label': label.strip(), 'context': context.strip() or None, 'decision': decision, + }) + if len(converted_answers) < 2: + raise ValueError( + f'legacy {key} question {question!r} must have at least two answers', + ) + if decisions != {'include', 'exclude'}: + diagnostics.append( + Diagnostic( + severity='warning', code='unbalanced_screening_decisions', + source_path=[key, question], target_path=[ + stage, question_index, 'answers', + ], + message='This question does not contain both Include and Exclude answers.', + requires_confirmation=True, + ), + ) + output.append({ + 'id': item_ids.allocate(f'q_{stage}_', question, f'question_{question_index + 1}'), + 'question': question.strip(), 'answers': converted_answers, 'trigger': {'all': []}, + }) + return output + + parameters = legacy.get('parameters', {}) + if parameters is None: + parameters = {} + if not isinstance(parameters, dict): + raise ValueError('legacy parameters must be a mapping') + converted_parameters = [] + parameter_index = 0 + for category, values in parameters.items(): + if not isinstance(category, str) or not category.strip() or not isinstance(values, dict): + raise ValueError( + 'legacy parameter categories must map to parameter mappings', + ) + for name, description in values.items(): + if not isinstance(name, str) or not name.strip() or not isinstance(description, str) or not description.strip(): + raise ValueError( + 'legacy parameter names and descriptions must be non-empty strings', + ) + parameter_index += 1 + converted_parameters.append({ + 'id': item_ids.allocate('p_', f'{category}_{name}', f'parameter_{parameter_index}'), + 'name': name.strip(), 'description': description.strip(), 'type': 'text', + 'unit_instructions': None, 'calculation': None, 'trigger': {'all': []}, + 'legacy_category': category.strip(), + }) + + l1 = questions('criteria', 'l1') + l2 = questions('l2_criteria', 'l2') + criteria = CriteriaConfigV2.model_validate({ + 'schema_version': 2, + 'citation_fields': {'l1_include': normalized_include, 'doi': None}, + 'l1': l1, 'l2': l2, 'parameters': converted_parameters, + }) + canonical_source = json.dumps( + legacy, sort_keys=True, separators=(',', ':'), ensure_ascii=False, + ) + fingerprint = f'sha256:{hashlib.sha256(canonical_source.encode()).hexdigest()}' + return CriteriaLoadResult( + criteria=criteria, + source_format='legacy_yaml_v1', + diagnostics=diagnostics, + requires_confirmation=any( + item.requires_confirmation for item in diagnostics + ), + fingerprint=fingerprint, + stats=MigrationStats( + l1=len(l1), l2=len( + l2, + ), parameters=len(converted_parameters), + ), + ) + + +criteria_configuration_service = CriteriaConfigurationService() diff --git a/backend/api/extract/prompts.py b/backend/api/extract/prompts.py index 7a5d8a0e..b9166967 100644 --- a/backend/api/extract/prompts.py +++ b/backend/api/extract/prompts.py @@ -2,6 +2,8 @@ PARAMETER_PROMPT_JSON = """ You are an expert information extractor for scientific full-text articles. You will be given: - A short description of a parameter to extract (what the parameter is and how it is defined). +- Optional reporting/unit instructions and calculation instructions. +- For selection parameters, the allowed options and their guidance. - The full text of a paper with each sentence numbered like: [0] First sentence. [1] Second sentence. etc. - Optionally, numbered tables (as markdown) and numbered figure captions (with the corresponding figure images provided alongside this message). @@ -25,6 +27,9 @@ Inputs available for formatting: - {parameter_name} (a short name for the parameter) - {parameter_description} (detailed description of what to look for) +- {unit_instructions} +- {calculation} +- {options} - {fulltext} (the numbered sentences string; e.g. "[0] First sentence\n[1] Next sentence\n...") - {tables} (numbered markdown tables) - {figures} (numbered figure captions) @@ -35,6 +40,9 @@ Do not output anything besides the JSON object. \n\nParameter name: {parameter_name} \nParameter description: {parameter_description} +\nUnit/reporting instructions: {unit_instructions} +\nCalculation instructions: {calculation} +\nAllowed selection options:\n{options} \n\nFull text (numbered sentences):\n{fulltext} \n\nTables (numbered):\n{tables} \n\nFigures (numbered; captions correspond to images provided alongside this message):\n{figures} diff --git a/backend/api/extract/router.py b/backend/api/extract/router.py index 184d9f73..f36cb258 100644 --- a/backend/api/extract/router.py +++ b/backend/api/extract/router.py @@ -27,6 +27,7 @@ from ..core.config import settings from ..core.docint_coords import normalize_bounding_regions_to_boxes from ..core.security import get_current_active_user +from ..criteria.runtime import item_is_visible from ..services.azure_docint_client import azure_docint_client from ..services.azure_openai_client import azure_openai_client from ..services.cit_db_service import cits_dp_service @@ -54,6 +55,10 @@ class ParameterExtractRequest(BaseModel): parameter_description: str = Field( ..., description='Human-friendly description of what to extract', ) + unit_instructions: str = '' + calculation: str = '' + options: list[str] = Field(default_factory=list) + option_contexts: dict[str, str] = Field(default_factory=dict) model: str | None = Field(None, description='Model to use') temperature: float | None = Field(0.0, ge=0.0, le=1.0) max_tokens: int | None = Field(512, ge=1, le=4000) @@ -131,6 +136,17 @@ async def extract_parameter_endpoint( table_name = (screening or {}).get('table_name') or 'citations' + # Enforce builder conditional visibility at the execution boundary. + canonical = sr.get('criteria') if isinstance( + sr.get('criteria'), dict, + ) else {} + parameter = next( + ( + item for item in (canonical.get('parameters') or []) + if isinstance(item, dict) and item.get('name') == payload.parameter_name + ), None, + ) + # Obtain fulltext: prefer payload, otherwise read from DB row fulltext = payload.fulltext row = None @@ -161,6 +177,12 @@ async def extract_parameter_endpoint( detail='Full text not provided and not available for this citation', ) + if parameter and not item_is_visible(canonical, parameter, row or {}): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail='This parameter is conditionally hidden by the configured criteria.', + ) + # Build tables/figures context: prefer payload fields, else fetch from DB row (if loaded) tables_text = payload.tables figures_text = payload.figures @@ -242,6 +264,11 @@ async def extract_parameter_endpoint( prompt = PARAMETER_PROMPT_JSON.format( parameter_name=payload.parameter_name, parameter_description=payload.parameter_description, + unit_instructions=payload.unit_instructions, + calculation=payload.calculation, + options='\n'.join( + f'- {option}: {payload.option_contexts.get(option, "")}' for option in payload.options + ), fulltext=fulltext, tables=tables_text, figures=figures_text, @@ -701,6 +728,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 +866,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 3d86019a..7694defb 100644 --- a/backend/api/screen/router.py +++ b/backend/api/screen/router.py @@ -24,6 +24,7 @@ from ..core.cit_utils import load_sr_and_check from ..core.config import settings from ..core.security import get_current_active_user +from ..criteria.runtime import item_is_visible from ..services.azure_openai_client import azure_openai_client from ..services.cit_db_service import cits_dp_service from ..services.cit_db_service import snake_case @@ -815,6 +816,24 @@ async def classify_citation( status_code=status.HTTP_404_NOT_FOUND, detail='Citation not found', ) + # Direct classify calls must honor canonical conditional visibility too. + canonical = sr.get('criteria') if isinstance( + sr.get('criteria'), dict, + ) else {} + stage = 'l2' if (payload.screening_step or '').lower() == 'l2' else 'l1' + source_item = next( + ( + item for item in (canonical.get(stage) or []) + if isinstance(item, dict) and item.get('question') == payload.question + ), + None, + ) + if source_item and not item_is_visible(canonical, source_item, row): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail='This criterion is conditionally hidden by the configured criteria.', + ) + # Build or use provided citation text (fall back to combined title/abstract when not provided) citation_text = payload.citation_text or citations_router._build_combined_citation_from_row( row, payload.include_columns, @@ -1097,6 +1116,22 @@ async def human_classify_citation( status_code=status.HTTP_404_NOT_FOUND, detail='Citation not found', ) + canonical = sr.get('criteria') if isinstance( + sr.get('criteria'), dict, + ) else {} + source_item = next( + ( + item for item in [*(canonical.get('l1') or []), *(canonical.get('l2') or [])] + if isinstance(item, dict) and item.get('question') == payload.question + ), + None, + ) + if source_item and not item_is_visible(canonical, source_item, row): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail='This criterion is conditionally hidden by the configured criteria.', + ) + citation_text = payload.citation_text confidence = payload.confidence @@ -1254,6 +1289,25 @@ async def _call_llm(prompt: str) -> tuple[str, dict[str, Any], int]: if not isinstance(q, str) or not q.strip(): continue + canonical = sr.get('criteria') if isinstance( + sr.get('criteria'), dict, + ) else {} + question_item = next( + ( + item for item in (canonical.get('l1') or []) + if isinstance(item, dict) and item.get('question') == q + ), + None, + ) + if question_item and not item_is_visible(canonical, question_item, row): + results.append({ + 'question': q, + 'criterion_key': snake_case(q, max_len=56), + 'skipped': True, + 'reason': 'conditional_trigger_not_met', + }) + continue + opts = possible[i] if i < len( possible, ) and isinstance(possible[i], list) else [] @@ -1681,6 +1735,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] = [] @@ -1832,6 +1896,25 @@ async def _call_llm(prompt: str) -> tuple[str, dict[str, Any], int]: if not isinstance(q, str) or not q.strip(): continue + canonical = sr.get('criteria') if isinstance( + sr.get('criteria'), dict, + ) else {} + question_item = next( + ( + item for item in (canonical.get(source_step) or []) + if isinstance(item, dict) and item.get('question') == q + ), + None, + ) + if question_item and not item_is_visible(canonical, question_item, row): + results.append({ + 'question': q, + 'criterion_key': snake_case(q, max_len=56), + 'skipped': True, + 'reason': 'conditional_trigger_not_met', + }) + continue + if source_step == 'l1': opts_raw = l1_possible[idx] if idx < len( l1_possible, @@ -1878,6 +1961,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 +1970,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 +1998,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 +2027,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 +2172,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 +2180,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 +2201,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, @@ -2215,7 +2366,10 @@ async def get_screening_metrics( try: ids = await run_in_threadpool( screening_eligibility_service.list_eligible_ids, - criteria=cp, table_name=table_name, target_stage=step_norm, + criteria=cp, + table_name=table_name, + target_stage=step_norm, + repair_decisions=False, ) except Exception as e: raise HTTPException( @@ -2768,7 +2922,10 @@ async def get_screening_calibration( try: ids = await run_in_threadpool( screening_eligibility_service.list_eligible_ids, - criteria=cp, table_name=table_name, target_stage=step_norm, + criteria=cp, + table_name=table_name, + target_stage=step_norm, + repair_decisions=False, ) except Exception as e: raise HTTPException( @@ -3070,7 +3227,10 @@ async def get_live_confidence_histogram( try: ids = await run_in_threadpool( screening_eligibility_service.list_eligible_ids, - criteria=cp, table_name=table_name, target_stage=step_norm, + criteria=cp, + table_name=table_name, + target_stage=step_norm, + repair_decisions=False, ) except Exception as e: raise HTTPException( @@ -3213,7 +3373,10 @@ async def get_calibration_samples( try: ids = await run_in_threadpool( screening_eligibility_service.list_eligible_ids, - criteria=cp, table_name=table_name, target_stage=step_norm, + criteria=cp, + table_name=table_name, + target_stage=step_norm, + repair_decisions=False, ) except Exception as e: raise HTTPException( 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/citation_field_service.py b/backend/api/services/citation_field_service.py new file mode 100644 index 00000000..24678512 --- /dev/null +++ b/backend/api/services/citation_field_service.py @@ -0,0 +1,65 @@ +"""Discover user-authored citation columns for criteria configuration.""" +from __future__ import annotations + +from typing import Any + +from .cit_db_service import cits_dp_service + +SYSTEM_NAMES = { + 'id', 'combined_citation', + 'human_l1_decision', 'human_l2_decision', +} +SYSTEM_PREFIXES = ('llm_', 'human_', 'fulltext', 'parameters_', 'l1_', 'l2_') + + +def _is_user_field(name: str) -> bool: + folded = name.casefold() + return folded not in SYSTEM_NAMES and not any(folded.startswith(prefix) for prefix in SYSTEM_PREFIXES) + + +def build_citation_field_contract(review: dict[str, Any], columns: list[dict[str, Any]]) -> dict[str, Any]: + fields = [] + for column in columns: + name = str( + column.get('column_name') + or column.get('name') or '', + ).strip() + if not name or not _is_user_field(name): + continue + fields.append({ + 'name': name, 'data_type': str( + column.get('data_type') or 'text', + ), + }) + + criteria = review.get('criteria') if isinstance( + review.get( + 'criteria', + ), dict, + ) else review.get('criteria_parsed') or {} + configured = criteria.get('citation_fields') if isinstance( + criteria.get('citation_fields'), dict, + ) else {} + selected = [ + str(value) for value in configured.get( + 'l1_include', + ) or criteria.get('include') or [] + ] + doi = configured.get('doi') + available = {field['name'] for field in fields} + unavailable = [ + value for value in [*selected, doi] + if value and value not in available + ] + return { + 'fields': fields, + 'unavailable_configured_fields': list(dict.fromkeys(unavailable)), + } + + +def discover_citation_fields(review: dict[str, Any]) -> dict[str, Any]: + table_name = (review.get('screening_db') or {}).get('table_name') + columns = cits_dp_service.get_table_columns( + table_name, + ) if table_name else [] + return build_citation_field_contract(review, 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/api/services/screening_eligibility_service.py b/backend/api/services/screening_eligibility_service.py index 9241baa4..ac300598 100644 --- a/backend/api/services/screening_eligibility_service.py +++ b/backend/api/services/screening_eligibility_service.py @@ -26,7 +26,11 @@ def _answer_object(value: Any) -> dict[str, Any]: def selected_answer(row: dict[str, Any], question: str) -> str | None: - """Return the effective answer using human-over-AI precedence.""" + """Return the effective answer using human-over-main-AI precedence. + + The critical agent is advisory: disagreements are routed to human review + but do not change the main screening agent's progression decision. + """ core = snake_case(question, max_len=56) if not core: return None @@ -90,7 +94,15 @@ def list_eligible_ids( criteria: dict[str, Any] | None, table_name: str, target_stage: str, + repair_decisions: bool = True, ) -> list[int]: + """Return citation IDs eligible for ``target_stage``. + + Progression callers should keep ``repair_decisions=True`` so derived + decision caches are refreshed before they are used. Read-only reporting + callers can set it to ``False`` to avoid turning concurrent metrics + requests into repeated, table-wide writes. + """ stage = str(target_stage or '').strip().lower() if stage not in {'l1', 'l2', 'extract'}: raise ValueError(f'Unsupported screening stage: {target_stage!r}') @@ -98,9 +110,12 @@ def list_eligible_ids( if stage == 'l1': return self.repository.list_citation_ids(None, table_name) - # Later stages depend on derived decisions. Repair first and fail loudly - # if repair fails; an empty list would be a misleading valid result. - self.repository.backfill_human_decisions(criteria or {}, table_name) + # Later stages depend on derived decisions. Progression paths repair + # first and fail loudly; reporting paths only read the existing cache. + if repair_decisions: + self.repository.backfill_human_decisions( + criteria or {}, table_name, + ) source_stage = 'l1' if stage == 'l2' else 'l2' return self.repository.list_citation_ids(source_stage, table_name) diff --git a/backend/api/services/sr_db_service.py b/backend/api/services/sr_db_service.py index f5027dd2..727f874c 100644 --- a/backend/api/services/sr_db_service.py +++ b/backend/api/services/sr_db_service.py @@ -23,6 +23,7 @@ from fastapi import status from ..core.config import settings +from ..criteria.service import criteria_configuration_service from .postgres_auth import postgres_server logger = logging.getLogger(__name__) @@ -56,6 +57,7 @@ def ensure_table_exists(self) -> None: criteria JSONB, criteria_yaml TEXT, criteria_parsed JSONB, + criteria_revision INTEGER NOT NULL DEFAULT 0, screening_thresholds JSONB, critical_prompt_additions JSONB, screening_db JSONB, @@ -67,6 +69,18 @@ def ensure_table_exists(self) -> None: # Runtime schema evolution for existing deployments. # (No migrations philosophy: add columns if missing.) + try: + cur.execute( + 'ALTER TABLE systematic_reviews ADD COLUMN IF NOT EXISTS criteria_revision INTEGER NOT NULL DEFAULT 0', + ) + except Exception: + try: + cur.execute( + 'ALTER TABLE systematic_reviews ADD COLUMN criteria_revision INTEGER NOT NULL DEFAULT 0', + ) + except Exception: + pass + try: cur.execute( 'ALTER TABLE systematic_reviews ADD COLUMN IF NOT EXISTS screening_thresholds JSONB', @@ -117,6 +131,11 @@ def build_criteria_parsed(self, criteria_obj: dict[str, Any] | None) -> dict[str if not criteria_obj or not isinstance(criteria_obj, dict): return parsed + if criteria_obj.get('schema_version') == 2: + return criteria_configuration_service.build_compatibility_projection( + criteria_obj, + ) + # include list (columns) include = criteria_obj.get('include', []) if isinstance(include, list): @@ -517,7 +536,8 @@ def update_criteria(self, sr_id: str, criteria_obj: dict[str, Any], criteria_str update_sql = """ UPDATE systematic_reviews - SET criteria = %s, criteria_yaml = %s, criteria_parsed = %s, updated_at = %s + SET criteria = %s, criteria_yaml = %s, criteria_parsed = %s, + criteria_revision = criteria_revision + 1, updated_at = %s WHERE id = %s """ @@ -565,6 +585,65 @@ def update_criteria(self, sr_id: str, criteria_obj: dict[str, Any], criteria_str if conn: pass + def save_criteria_config( + self, + sr_id: str, + criteria_obj: dict[str, Any], + criteria_str: str, + expected_revision: int, + requester_id: str, + ) -> dict[str, Any]: + """Atomically save all canonical criteria representations with optimistic locking.""" + sr = self.get_systematic_review(sr_id) + if not sr or not sr.get('visible', True): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail='Systematic review not found', + ) + if not self.user_has_sr_permission(sr_id, requester_id): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail='Not authorized to modify this systematic review', + ) + + projection = self.build_criteria_parsed(criteria_obj) + conn = None + try: + conn = postgres_server.conn + cur = conn.cursor() + cur.execute( + """ + UPDATE systematic_reviews + SET criteria = %s, criteria_yaml = %s, criteria_parsed = %s, + criteria_revision = criteria_revision + 1, updated_at = %s + WHERE id = %s AND criteria_revision = %s + RETURNING criteria_revision + """, + ( + json.dumps(criteria_obj), criteria_str, json.dumps( + projection, + ), + datetime.utcnow().isoformat(), sr_id, expected_revision, + ), + ) + row = cur.fetchone() + if not row: + conn.rollback() + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail='Criteria revision conflict. Reload the latest configuration and retry.', + ) + conn.commit() + return {'revision': row[0], 'criteria_parsed': projection} + except HTTPException: + raise + except Exception as exc: + if conn: + conn.rollback() + logger.exception( + 'Failed to save canonical criteria config: %s', exc, + ) + raise + def list_systematic_reviews_for_user(self, user_email: str) -> list[dict[str, Any]]: """ Return all SR documents where the user is a member (regardless of visible flag). diff --git a/backend/api/sr/router.py b/backend/api/sr/router.py index 62d9e1b6..cd18f9de 100644 --- a/backend/api/sr/router.py +++ b/backend/api/sr/router.py @@ -20,17 +20,63 @@ from fastapi import status from fastapi import UploadFile from fastapi.concurrency import run_in_threadpool +from fastapi.responses import PlainTextResponse from pydantic import BaseModel +from pydantic import ValidationError from ..core.cit_utils import load_sr_and_check from ..core.config import settings from ..core.security import get_current_active_user +from ..criteria.models import CriteriaConfigV2 +from ..criteria.service import criteria_configuration_service +from ..services.citation_field_service import discover_citation_fields from ..services.sr_db_service import srdb_service from ..services.user_db import user_db_service router = APIRouter() +class CriteriaConfigSaveRequest(BaseModel): + expected_revision: int + force: bool = False + migration_fingerprint: str | None = None + criteria: CriteriaConfigV2 + + +class CriteriaYamlImportRequest(BaseModel): + criteria_yaml: str + + +def _criteria_error(exc: ValueError) -> HTTPException: + if isinstance(exc, ValidationError): + errors = [ + { + 'path': list(error['loc']), + 'code': error['type'], + 'message': error['msg'], + } + for error in exc.errors(include_url=False) + ] + else: + errors = [ + {'path': [], 'code': 'invalid_criteria', 'message': str(exc)}, + ] + return HTTPException(status_code=422, detail={'errors': errors}) + + +def _require_migration_confirmation(result, fingerprint: str | None) -> None: + if result.source_format == 'legacy_yaml_v1' and result.requires_confirmation: + if fingerprint != result.fingerprint: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={ + 'code': 'migration_confirmation_required', + 'message': 'Preview and confirm the legacy migration before saving.', + 'fingerprint': result.fingerprint, + }, + ) + + class SystematicReviewCreate(BaseModel): name: str description: str | None = None @@ -79,6 +125,7 @@ async def create_systematic_review( description: str | None = Form(None), criteria_file: UploadFile | None = File(None), criteria_yaml: str | None = Form(None), + migration_fingerprint: str | None = Form(None), current_user: dict[str, Any] = Depends(get_current_active_user), ): """ @@ -111,19 +158,20 @@ async def create_systematic_review( criteria_str = criteria_yaml if criteria_str: - criteria_obj = yaml.safe_load(criteria_str) - # ensure it's a mapping/dict - if criteria_obj is None: - criteria_obj = {} - elif not isinstance(criteria_obj, dict): - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail='Parsed YAML criteria must be a mapping/object at the top level', - ) - except yaml.YAMLError as ye: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid YAML provided: {ye}", - ) + normalized = criteria_configuration_service.parse_yaml( + criteria_str, + ) + _require_migration_confirmation(normalized, migration_fingerprint) + criteria_obj = normalized.criteria.model_dump( + mode='json', exclude_none=True, + ) + criteria_str = criteria_configuration_service.export_yaml( + normalized.criteria, + ) + except (yaml.YAMLError, ValueError) as exc: + raise _criteria_error(exc) from exc + except HTTPException: + raise except Exception as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=str(e), @@ -317,6 +365,144 @@ async def list_systematic_reviews_for_user( return results +async def _load_criteria_review(sr_id: str, current_user: dict[str, Any]) -> dict[str, Any]: + review, _screening = await load_sr_and_check( + sr_id, current_user, srdb_service, require_screening=False, + ) + return review + + +@router.get('/{sr_id}/citation-fields') +async def get_citation_fields( + sr_id: str, + current_user: dict[str, Any] = Depends(get_current_active_user), +): + review = await _load_criteria_review(sr_id, current_user) + return await run_in_threadpool(discover_citation_fields, review) + + +@router.get('/{sr_id}/criteria-config') +async def get_criteria_config( + sr_id: str, + current_user: dict[str, Any] = Depends(get_current_active_user), +): + review = await _load_criteria_review(sr_id, current_user) + try: + if isinstance(review.get('criteria'), dict): + result = criteria_configuration_service.normalize( + review['criteria'], + ) + elif review.get('criteria_yaml'): + result = criteria_configuration_service.parse_yaml( + review['criteria_yaml'], source_kind='backend_load', + ) + else: + raise ValueError( + 'No usable criteria configuration is stored for this review.', + ) + except ValueError as exc: + raise _criteria_error(exc) from exc + + response: dict[str, Any] = { + 'criteria': result.criteria.model_dump(mode='json', exclude_none=True), + 'revision': review.get('criteria_revision', 0), + 'warnings': [item.model_dump(mode='json') for item in result.diagnostics], + } + if result.source_format == 'legacy_yaml_v1': + response['migration'] = { + 'status': 'preview', + 'source_format': result.source_format, + 'fingerprint': result.fingerprint, + 'requires_confirmation': result.requires_confirmation, + 'stats': result.stats.model_dump() if result.stats else None, + 'diagnostics': [item.model_dump(mode='json') for item in result.diagnostics], + } + return response + + +@router.post('/{sr_id}/criteria-config/validate') +async def validate_criteria_config( + sr_id: str, + criteria: CriteriaConfigV2, + current_user: dict[str, Any] = Depends(get_current_active_user), +): + await _load_criteria_review(sr_id, current_user) + return { + 'valid': True, + 'criteria': criteria.model_dump(mode='json', exclude_none=True), + 'warnings': [], + } + + +@router.post('/{sr_id}/criteria-config/import-yaml') +async def import_criteria_yaml( + sr_id: str, + payload: CriteriaYamlImportRequest, + current_user: dict[str, Any] = Depends(get_current_active_user), +): + await _load_criteria_review(sr_id, current_user) + try: + result = criteria_configuration_service.parse_yaml( + payload.criteria_yaml, + ) + except ValueError as exc: + raise _criteria_error(exc) from exc + return result.model_dump(mode='json', exclude_none=True) + + +@router.put('/{sr_id}/criteria-config') +async def save_criteria_config( + sr_id: str, + payload: CriteriaConfigSaveRequest, + current_user: dict[str, Any] = Depends(get_current_active_user), +): + review = await _load_criteria_review(sr_id, current_user) + if (review.get('screening_db') or {}).get('table_name') and not payload.force: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail='This SR already has screening data. Pass force=true to confirm criteria invalidation.', + ) + try: + stored = criteria_configuration_service.normalize(review['criteria']) + except (KeyError, ValueError): + stored = None + if stored: + _require_migration_confirmation(stored, payload.migration_fingerprint) + criteria = payload.criteria.model_dump(mode='json', exclude_none=True) + criteria_yaml = criteria_configuration_service.export_yaml( + payload.criteria, + ) + saved = await run_in_threadpool( + srdb_service.save_criteria_config, + sr_id, + criteria, + criteria_yaml, + payload.expected_revision, + current_user.get('email') or current_user.get('id'), + ) + return { + 'criteria': criteria, + 'revision': saved['revision'], + 'warnings': [], + 'invalidation': {'forced': payload.force, 'screening_data_present': bool(review.get('screening_db'))}, + } + + +@router.get('/{sr_id}/criteria-config/export-yaml', response_class=PlainTextResponse) +async def export_criteria_yaml( + sr_id: str, + current_user: dict[str, Any] = Depends(get_current_active_user), +): + response = await get_criteria_config(sr_id, current_user) + return PlainTextResponse( + criteria_configuration_service.export_yaml(response['criteria']), + media_type='application/yaml', + headers={ + 'Content-Disposition': f'attachment; filename="criteria-{sr_id}.yaml"', + }, + ) + + @router.get('/{sr_id}', response_model=SystematicReviewRead) async def get_systematic_review(sr_id: str, current_user: dict[str, Any] = Depends(get_current_active_user)): """ @@ -383,6 +569,7 @@ async def update_systematic_review_criteria( criteria_file: UploadFile | None = File(None), criteria_yaml: str | None = Form(None), force: str | None = Form(None), + migration_fingerprint: str | None = Form(None), current_user: dict[str, Any] = Depends(get_current_active_user), ): """ @@ -435,18 +622,18 @@ async def update_systematic_review_criteria( detail='Either criteria_file or criteria_yaml must be provided', ) - criteria_obj = yaml.safe_load(criteria_str) - if criteria_obj is None: - criteria_obj = {} - elif not isinstance(criteria_obj, dict): - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail='Parsed YAML criteria must be a mapping/object at the top level', - ) - except yaml.YAMLError as ye: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid YAML provided: {ye}", + normalized = criteria_configuration_service.parse_yaml(criteria_str) + _require_migration_confirmation(normalized, migration_fingerprint) + criteria_obj = normalized.criteria.model_dump( + mode='json', exclude_none=True, + ) + criteria_str = criteria_configuration_service.export_yaml( + normalized.criteria, ) + except (yaml.YAMLError, ValueError) as exc: + raise _criteria_error(exc) from exc + except HTTPException: + raise except Exception as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=str(e), 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/citation_field_service_test.py b/backend/tests/citation_field_service_test.py new file mode 100644 index 00000000..6dba2c3f --- /dev/null +++ b/backend/tests/citation_field_service_test.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import unittest + +from api.services.citation_field_service import build_citation_field_contract + + +class CitationFieldServiceTests(unittest.TestCase): + def test_preserves_order_filters_runtime_columns_and_reports_unavailable(self): + result = build_citation_field_contract( + { + 'criteria': { + 'citation_fields': { + 'l1_include': ['Title', 'Missing'], 'doi': 'DOI', + }, + }, + }, + [{'column_name': name, 'data_type': 'text'} for name in [ + 'id', 'Title', 'Abstract', 'DOI', 'human_answer', 'fulltext_url', + ]], + ) + self.assertEqual( + result['fields'], [ + {'name': 'Title', 'data_type': 'text'}, + {'name': 'Abstract', 'data_type': 'text'}, + {'name': 'DOI', 'data_type': 'text'}, + ], + ) + self.assertEqual(result['unavailable_configured_fields'], ['Missing']) + + def test_supports_review_without_upload(self): + self.assertEqual(build_citation_field_contract({}, [])['fields'], []) diff --git a/backend/tests/criteria_config_router_test.py b/backend/tests/criteria_config_router_test.py new file mode 100644 index 00000000..e61464dc --- /dev/null +++ b/backend/tests/criteria_config_router_test.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import unittest +from unittest.mock import AsyncMock +from unittest.mock import patch + +from api.criteria.models import CriteriaConfigV2 +from api.sr.router import CriteriaConfigSaveRequest +from api.sr.router import CriteriaYamlImportRequest +from api.sr.router import get_citation_fields +from api.sr.router import get_criteria_config +from api.sr.router import import_criteria_yaml +from api.sr.router import save_criteria_config +from fastapi import HTTPException + + +EMPTY_CONFIG = { + 'schema_version': 2, + 'citation_fields': {'l1_include': []}, + 'l1': [], + 'l2': [], + 'parameters': [], +} +USER = {'id': 'owner-id', 'email': 'owner@example.test'} + + +class CriteriaConfigRouterTests(unittest.IsolatedAsyncioTestCase): + @patch('api.sr.router.discover_citation_fields') + @patch('api.sr.router._load_criteria_review', new_callable=AsyncMock) + async def test_citation_fields_uses_authorized_review(self, load_review: AsyncMock, discover) -> None: + review = {'screening_db': {'table_name': 'review_citations'}} + load_review.return_value = review + discover.return_value = { + 'fields': [], 'unavailable_configured_fields': [], + } + response = await get_citation_fields('review-id', USER) + load_review.assert_awaited_once_with('review-id', USER) + discover.assert_called_once_with(review) + self.assertEqual(response['fields'], []) + + @patch('api.sr.router._load_criteria_review', new_callable=AsyncMock) + async def test_get_normalizes_legacy_without_persisting(self, load_review: AsyncMock) -> None: + load_review.return_value = { + 'criteria_revision': 4, + 'criteria': { + 'include': ['Title'], + 'criteria': {'Eligible?': {'Yes': 'Include', 'No (exclude)': 'Exclude'}}, + }, + } + + response = await get_criteria_config('review-id', USER) + + self.assertEqual(response['revision'], 4) + self.assertEqual(response['criteria']['schema_version'], 2) + self.assertEqual(response['migration']['status'], 'preview') + self.assertTrue( + response['migration'] + ['fingerprint'].startswith('sha256:'), + ) + + @patch('api.sr.router._load_criteria_review', new_callable=AsyncMock) + async def test_import_returns_422_for_invalid_yaml(self, load_review: AsyncMock) -> None: + load_review.return_value = {'criteria_revision': 0} + + with self.assertRaises(HTTPException) as caught: + await import_criteria_yaml( + 'review-id', CriteriaYamlImportRequest( + criteria_yaml='- not\n- a mapping', + ), USER, + ) + + self.assertEqual(caught.exception.status_code, 422) + self.assertEqual( + caught.exception.detail['errors'][0]['code'], 'invalid_criteria', + ) + + @patch('api.sr.router._load_criteria_review', new_callable=AsyncMock) + async def test_save_requires_force_when_screening_data_exists(self, load_review: AsyncMock) -> None: + load_review.return_value = { + 'screening_db': {'table_name': 'citations'}, + } + payload = CriteriaConfigSaveRequest( + expected_revision=2, + force=False, + criteria=CriteriaConfigV2.model_validate(EMPTY_CONFIG), + ) + + with self.assertRaises(HTTPException) as caught: + await save_criteria_config('review-id', payload, USER) + + self.assertEqual(caught.exception.status_code, 409) + + @patch('api.sr.router._load_criteria_review', new_callable=AsyncMock) + async def test_save_requires_matching_legacy_migration_fingerprint(self, load_review: AsyncMock) -> None: + load_review.return_value = { + 'criteria': { + 'criteria': {'Eligible?': {'Yes': 'Include', 'No (exclude)': 'Exclude'}}, + }, + } + payload = CriteriaConfigSaveRequest( + expected_revision=0, + criteria=CriteriaConfigV2.model_validate(EMPTY_CONFIG), + ) + + with self.assertRaises(HTTPException) as caught: + await save_criteria_config('review-id', payload, USER) + + self.assertEqual(caught.exception.status_code, 409) + self.assertEqual( + caught.exception.detail['code'], 'migration_confirmation_required', + ) + self.assertTrue( + caught.exception.detail['fingerprint'].startswith('sha256:'), + ) + + +if __name__ == '__main__': + unittest.main() diff --git a/backend/tests/criteria_configuration_service_test.py b/backend/tests/criteria_configuration_service_test.py new file mode 100644 index 00000000..1736f140 --- /dev/null +++ b/backend/tests/criteria_configuration_service_test.py @@ -0,0 +1,192 @@ +from __future__ import annotations + +import unittest +from pathlib import Path + +from api.criteria.models import CriteriaConfigV2 +from api.criteria.service import CriteriaConfigurationService +from pydantic import ValidationError + + +class CriteriaConfigurationServiceTests(unittest.TestCase): + def setUp(self) -> None: + self.service = CriteriaConfigurationService() + + def test_v2_yaml_round_trip_is_deterministic(self) -> None: + config = { + 'schema_version': 2, + 'citation_fields': {'l1_include': ['Title', 'Abstract'], 'doi': 'DOI'}, + 'l1': [{ + 'id': 'q_primary', + 'question': 'Primary research?', + 'answers': [ + {'id': 'yes', 'label': 'Yes', 'decision': 'include'}, + {'id': 'no_answer', 'label': 'No', 'decision': 'exclude'}, + ], + 'trigger': {'all': []}, + }], + 'l2': [], + 'parameters': [{ + 'id': 'p_design', + 'name': 'Design', + 'description': 'Reported study design.', + 'type': 'selection', + 'selection_mode': 'single', + 'options': [{'id': 'cohort', 'label': 'Cohort'}], + 'trigger': {'all': [{'source_item_id': 'q_primary', 'option_id': 'yes'}]}, + }], + } + + first = self.service.export_yaml(config) + loaded = self.service.parse_yaml(first) + second = self.service.export_yaml(loaded.criteria) + + self.assertEqual(first, second) + self.assertEqual( + loaded.criteria.model_dump( + mode='json', + ), CriteriaConfigV2.model_validate(config).model_dump(mode='json'), + ) + + def test_forward_and_text_parameter_trigger_sources_are_rejected(self) -> None: + base = { + 'schema_version': 2, 'citation_fields': {}, 'l1': [], 'l2': [], + 'parameters': [ + { + 'id': 'p_text', 'name': 'Text', 'description': 'Text value', 'type': 'text', + }, + { + 'id': 'p_later', 'name': 'Later', 'description': 'Later value', 'type': 'text', + 'trigger': {'all': [{'source_item_id': 'p_text', 'option_id': 'anything'}]}, + }, + ], + } + + with self.assertRaisesRegex(ValidationError, 'text parameters cannot be trigger sources'): + CriteriaConfigV2.model_validate(base) + + base['parameters'][0]['trigger'] = { + 'all': [{'source_item_id': 'p_later', 'option_id': 'anything'}], + } + with self.assertRaisesRegex(ValidationError, 'must reference an earlier item'): + CriteriaConfigV2.model_validate(base) + + def test_representative_legacy_yaml_migrates_deterministically(self) -> None: + path = Path(__file__).parents[1] / \ + 'api/sr/criteria_config_measles_updated.yaml' + content = path.read_text(encoding='utf-8') + + first = self.service.parse_yaml(content) + second = self.service.parse_yaml(content) + + self.assertEqual(first.criteria, second.criteria) + self.assertEqual(first.fingerprint, second.fingerprint) + self.assertEqual(first.stats.l1, 3) + self.assertEqual(first.stats.l2, 1) + self.assertGreater(first.stats.parameters, 1) + self.assertTrue(first.requires_confirmation) + self.assertIn( + 'decision_inferred_exclude', { + item.code for item in first.diagnostics + }, + ) + + def test_legacy_collision_ids_are_stable_and_unique(self) -> None: + legacy = ''' +criteria: + Same question: + "Yes": Include + "No (exclude)": Exclude +l2_criteria: + Same question: + "Yes": Include + "No (exclude)": Exclude +parameters: + Category: + A/B: First + A B: Second +''' + result = self.service.parse_yaml(legacy) + item_ids = [ + *[item.id for item in result.criteria.l1], + *[item.id for item in result.criteria.l2], + *[item.id for item in result.criteria.parameters], + ] + + self.assertEqual(len(item_ids), len(set(item_ids))) + self.assertEqual(item_ids[-1], f'{item_ids[-2]}_2') + + def test_mixed_v2_and_legacy_keys_are_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, 'cannot contain legacy keys'): + self.service.normalize({ + 'schema_version': 2, 'citation_fields': {}, 'l1': [], 'l2': [], + 'parameters': [], 'include': ['Title'], + }) + + def test_obsolete_question_context_is_preserved_on_every_answer(self) -> None: + result = self.service.normalize({ + 'schema_version': 2, + 'citation_fields': {}, + 'l1': [{ + 'id': 'q_primary', + 'question': 'Primary research?', + 'context': 'Guidance shared by the old question.', + 'answers': [ + { + 'id': 'yes', 'label': 'Yes', 'context': 'Yes-specific guidance.', + 'decision': 'include', + }, + {'id': 'no_answer', 'label': 'No', 'decision': 'exclude'}, + ], + }], + 'l2': [], + 'parameters': [], + }) + + self.assertEqual( + result.criteria.l1[0].answers[0].context, + 'Guidance shared by the old question.\n\nYes-specific guidance.', + ) + self.assertEqual( + result.criteria.l1[0].answers[1].context, + 'Guidance shared by the old question.', + ) + self.assertEqual( + [item.code for item in result.diagnostics], + ['question_context_moved_to_answers'], + ) + + def test_projection_retains_legacy_arrays_and_adds_stable_items(self) -> None: + migrated = self.service.normalize({ + 'include': ['Title'], + 'criteria': {'Eligible?': {'Yes': 'Include it', 'No (exclude)': 'Exclude it'}}, + 'parameters': {'Group': {'Rate': 'A reported rate'}}, + }) + + projection = self.service.build_compatibility_projection( + migrated.criteria, + ) + + self.assertEqual(projection['schema_version'], 2) + self.assertEqual(projection['l1']['questions'], ['Eligible?']) + self.assertEqual(projection['l1']['include'], ['Title']) + self.assertIn( + '\nInclude it\n', + projection['l1']['additional_infos'][0], + ) + self.assertIn( + '\nExclude it\n', + projection['l1']['additional_infos'][0], + ) + self.assertEqual( + projection['l1']['items'] + [0]['id'], migrated.criteria.l1[0].id, + ) + self.assertEqual( + projection['parameters']['items'] + [0]['id'], migrated.criteria.parameters[0].id, + ) + + +if __name__ == '__main__': + unittest.main() 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/backend/tests/screening_eligibility_service_test.py b/backend/tests/screening_eligibility_service_test.py index a54a7384..7a426cec 100644 --- a/backend/tests/screening_eligibility_service_test.py +++ b/backend/tests/screening_eligibility_service_test.py @@ -8,7 +8,11 @@ if 'psycopg' not in sys.modules: psycopg = types.ModuleType('psycopg') psycopg.connect = Mock() + psycopg_rows = types.ModuleType('psycopg.rows') + psycopg_rows.dict_row = Mock() + psycopg.rows = psycopg_rows sys.modules['psycopg'] = psycopg + sys.modules['psycopg.rows'] = psycopg_rows from api.services.screening_eligibility_service import ScreeningEligibilityService from api.services.screening_eligibility_service import compute_screening_decisions @@ -44,6 +48,20 @@ def test_ai_answer_is_used_when_human_answer_is_missing(self) -> None: ('include', 'include'), ) + def test_critical_ai_answer_does_not_override_screening_ai_answer(self) -> None: + row = { + 'llm_relevant_population': { + 'selected': 'Include', + 'critical': {'selected': 'Exclude'}, + }, + 'llm_eligible_study_design': {'selected': 'Include'}, + } + + self.assertEqual( + compute_screening_decisions(row, CRITERIA), + ('include', 'include'), + ) + def test_l2_decision_is_cumulative_and_missing_answers_exclude(self) -> None: row = {'human_relevant_population': {'selected': 'Include'}} @@ -93,6 +111,60 @@ def test_l1_does_not_run_unnecessary_repair(self) -> None: None, 'screening_table', ) + def test_l2_read_only_scope_does_not_repair_decisions(self) -> None: + repository = Mock() + repository.list_citation_ids.return_value = [4, 9] + service = ScreeningEligibilityService(repository) + + result = service.list_eligible_ids( + criteria=CRITERIA, + table_name='screening_table', + target_stage='l2', + repair_decisions=False, + ) + + self.assertEqual(result, [4, 9]) + repository.backfill_human_decisions.assert_not_called() + repository.list_citation_ids.assert_called_once_with( + 'l1', 'screening_table', + ) + + def test_extract_read_only_scope_uses_l2_decisions(self) -> None: + repository = Mock() + repository.list_citation_ids.return_value = [12] + service = ScreeningEligibilityService(repository) + + result = service.list_eligible_ids( + criteria=CRITERIA, + table_name='screening_table', + target_stage='extract', + repair_decisions=False, + ) + + self.assertEqual(result, [12]) + repository.backfill_human_decisions.assert_not_called() + repository.list_citation_ids.assert_called_once_with( + 'l2', 'screening_table', + ) + + def test_read_only_scope_is_not_affected_by_repair_failure(self) -> None: + repository = Mock() + repository.backfill_human_decisions.side_effect = RuntimeError( + 'repair failed', + ) + repository.list_citation_ids.return_value = [4] + service = ScreeningEligibilityService(repository) + + result = service.list_eligible_ids( + criteria=CRITERIA, + table_name='screening_table', + target_stage='l2', + repair_decisions=False, + ) + + self.assertEqual(result, [4]) + repository.backfill_human_decisions.assert_not_called() + def test_repair_failure_is_not_hidden_as_an_empty_scope(self) -> None: repository = Mock() repository.backfill_human_decisions.side_effect = RuntimeError( diff --git a/frontend/app/[lang]/can-sr/extract/page.tsx b/frontend/app/[lang]/can-sr/extract/page.tsx index a15f8773..13a4d7ca 100644 --- a/frontend/app/[lang]/can-sr/extract/page.tsx +++ b/frontend/app/[lang]/can-sr/extract/page.tsx @@ -1,6 +1,7 @@ 'use client' import CitationsListPage from '@/components/can-sr/CitationListPage' +import { flattenExtractionParameters } from '@/components/can-sr/extraction-parameters' import type { AiCall, @@ -11,6 +12,7 @@ type ParametersParsed = { categories: string[] possible_parameters: any[][] descriptions: any[][] + items?: any[] } const buildCitationAiCalls: BuildCitationAiCalls = async ({ @@ -42,7 +44,7 @@ const buildCitationAiCalls: BuildCitationAiCalls = async ({ // Load parameter definitions (criteria_parsed.parameters) for this SR. // This is per-citation today (simple + correct); we can cache later if needed. - let paramsFlat: Array<{ name: string; description: string }> = [] + let paramsFlat = [] as ReturnType try { const res = await fetch( `/api/can-sr/reviews/create?sr_id=${encodeURIComponent(srId)}&criteria_parsed=1`, @@ -50,29 +52,11 @@ const buildCitationAiCalls: BuildCitationAiCalls = async ({ ) const data = await res.json().catch(() => ({})) const parsed = data?.criteria_parsed || data?.criteria || {} - const paramsInfo: ParametersParsed | null = (parsed?.parameters as any) || null + const paramsInfo: ParametersParsed | null = + (parsed?.parameters as any) || null - if (paramsInfo?.categories && paramsInfo?.possible_parameters) { - const out: Array<{ name: string; description: string }> = [] - paramsInfo.categories.forEach((_cat, i) => { - const arr = paramsInfo.possible_parameters?.[i] || [] - const descs = paramsInfo.descriptions?.[i] || [] - arr.forEach((param: any, j: number) => { - const rawName = - typeof param === 'string' - ? param - : Array.isArray(param) - ? param[0] - : String(param) - const rawDesc = - typeof descs?.[j] === 'string' ? (descs[j] as string) : '' - const cleanDesc = rawDesc.replace(/<\/?desc>/g, '') - if (rawName && rawName.trim()) { - out.push({ name: rawName.trim(), description: cleanDesc || rawName }) - } - }) - }) - paramsFlat = out + if (paramsInfo) { + paramsFlat = flattenExtractionParameters(paramsInfo) } } catch (e) { console.warn('Failed to load parameters for extraction', e) @@ -113,6 +97,10 @@ const buildCitationAiCalls: BuildCitationAiCalls = async ({ body: JSON.stringify({ parameter_name: p.name, parameter_description: p.description, + unit_instructions: p.unit_instructions, + calculation: p.calculation, + options: p.options, + option_contexts: p.option_contexts, model, temperature: 0.0, max_tokens: 512, diff --git a/frontend/app/[lang]/can-sr/extract/view/page.tsx b/frontend/app/[lang]/can-sr/extract/view/page.tsx index acdb0b95..9132c076 100644 --- a/frontend/app/[lang]/can-sr/extract/view/page.tsx +++ b/frontend/app/[lang]/can-sr/extract/view/page.tsx @@ -5,9 +5,15 @@ import { useState, useEffect, useRef } from 'react' import GCHeader, { SRHeader } from '@/components/can-sr/headers' import { getAuthToken, getTokenType } from '@/lib/auth' import { ModelSelector } from '@/components/chat' -import PDFBoundingBoxViewer, { PDFBoundingBoxViewerHandle } from '@/components/can-sr/PDFBoundingBoxViewer' +import PDFBoundingBoxViewer, { + PDFBoundingBoxViewerHandle, +} from '@/components/can-sr/PDFBoundingBoxViewer' import { ChevronDown, ChevronRight, Wand2 } from 'lucide-react' import { useDictionary } from '@/app/[lang]/DictionaryProvider' +import { + flattenExtractionParameters, + type ExtractionParameter, +} from '@/components/can-sr/extraction-parameters' function snakeCase(name: string, maxLen = 63): string { if (!name) return '' @@ -42,11 +48,7 @@ export default function CanSrL2ScreenPage() { // Navigation list (Extract is filtered by L2 pass) const [citationIdList, setCitationIdList] = useState([]) - type ParametersParsed = { - categories: string[] - possible_parameters: string[][] - descriptions: string[][] - } + type ParametersParsed = { items: ExtractionParameter[] } const getAuthHeaders = (): Record => { const token = getAuthToken() @@ -77,7 +79,8 @@ export default function CanSrL2ScreenPage() { loadIds() }, [srId]) - const [parametersParsed, setParametersParsed] = useState(null) + const [parametersParsed, setParametersParsed] = + useState(null) const [paramValues, setParamValues] = useState>({}) const [savingParam, setSavingParam] = useState(null) const [saveStatus, setSaveStatus] = useState>({}) @@ -86,11 +89,18 @@ export default function CanSrL2ScreenPage() { setDescOpen((prev) => ({ ...prev, [name]: !prev[name] })) const [paramFound, setParamFound] = useState>({}) - const [aiStatus, setAiStatus] = useState>({}) + const [aiStatus, setAiStatus] = useState< + Record< + string, + 'queued' | 'extracting' | 'suggesting' | 'suggested' | 'error' + > + >({}) const [aiPanels, setAiPanels] = useState>({}) const [panelOpen, setPanelOpen] = useState>({}) const [fulltextCoords, setFulltextCoords] = useState(null) - const [fulltextPages, setFulltextPages] = useState<{ width: number; height: number }[] | null>(null) + const [fulltextPages, setFulltextPages] = useState< + { width: number; height: number }[] | null + >(null) const [fulltextStr, setFulltextStr] = useState(null) const viewerRef = useRef(null) @@ -99,9 +109,14 @@ export default function CanSrL2ScreenPage() { const [fulltextFigures, setFulltextFigures] = useState(null) const scrollToArtifact = (kind: 'table' | 'figure', idx: number) => { - const list = kind === 'table' ? (fulltextTables || []) : (fulltextFigures || []) + const list = kind === 'table' ? fulltextTables || [] : fulltextFigures || [] const item = list.find((x: any) => Number(x?.index) === Number(idx)) - console.log('[artifact-click]', { kind, idx, hasViewer: !!viewerRef.current, item }) + console.log('[artifact-click]', { + kind, + idx, + hasViewer: !!viewerRef.current, + item, + }) if (!item || !viewerRef.current) return const bbox = item?.bounding_box const first = Array.isArray(bbox) ? bbox[0] : null @@ -111,7 +126,10 @@ export default function CanSrL2ScreenPage() { } const [runningAllAI, setRunningAllAI] = useState(false) - const [runAllProgress, setRunAllProgress] = useState<{ done: number; total: number } | null>(null) + const [runAllProgress, setRunAllProgress] = useState<{ + done: number + total: number + } | null>(null) const [runAllError, setRunAllError] = useState(null) const citationKey = `${srId || ''}:${citationId || ''}` const currentCitationKeyRef = useRef(citationKey) @@ -206,21 +224,32 @@ export default function CanSrL2ScreenPage() { const suggestParam = async ( name: string, description: string, - context?: { srId: string; citationId: string; model: string; fullText?: string | null }, + context?: { + srId: string + citationId: string + model: string + fullText?: string | null + }, ): Promise<{ ok: boolean; error?: string }> => { - const runContext = context || (citationId && srId - ? { srId, citationId, model: selectedModel } - : null) + const runContext = + context || + (citationId && srId ? { srId, citationId, model: selectedModel } : null) if (!runContext) return { ok: false, error: 'Missing citation context' } const requestKey = `${runContext.srId}:${runContext.citationId}` - if (currentCitationKeyRef.current !== requestKey) return { ok: false, error: 'Citation changed' } - setAiStatus(prev => ({ ...prev, [name]: 'extracting' })) + if (currentCitationKeyRef.current !== requestKey) + return { ok: false, error: 'Citation changed' } + setAiStatus((prev) => ({ ...prev, [name]: 'extracting' })) try { const headers = getAuthHeaders() - const fullText = context && 'fullText' in context ? context.fullText : await ensureFullText() - if (!fullText) throw new Error('Full text is not available for this citation') - if (currentCitationKeyRef.current !== requestKey) return { ok: false, error: 'Citation changed' } - setAiStatus(prev => ({ ...prev, [name]: 'suggesting' })) + const fullText = + context && 'fullText' in context + ? context.fullText + : await ensureFullText() + if (!fullText) + throw new Error('Full text is not available for this citation') + if (currentCitationKeyRef.current !== requestKey) + return { ok: false, error: 'Citation changed' } + setAiStatus((prev) => ({ ...prev, [name]: 'suggesting' })) const res = await fetch( `/api/can-sr/extract?action=extract-parameter&sr_id=${encodeURIComponent( runContext.srId, @@ -239,21 +268,27 @@ export default function CanSrL2ScreenPage() { }, ) const data = await res.json().catch(() => ({})) - if (currentCitationKeyRef.current !== requestKey) return { ok: false, error: 'Citation changed' } + if (currentCitationKeyRef.current !== requestKey) + return { ok: false, error: 'Citation changed' } const ext = data?.extraction || data if (res.ok && ext) { - setParamValues(prev => ({ ...prev, [name]: ext?.value ?? '' })) - setParamFound(prev => ({ ...prev, [name]: !!ext?.found })) - setAiPanels(prev => ({ ...prev, [name]: ext })) - setPanelOpen(prev => ({ ...prev, [name]: false })) - setAiStatus(prev => ({ ...prev, [name]: 'suggested' })) + setParamValues((prev) => ({ ...prev, [name]: ext?.value ?? '' })) + setParamFound((prev) => ({ ...prev, [name]: !!ext?.found })) + setAiPanels((prev) => ({ ...prev, [name]: ext })) + setPanelOpen((prev) => ({ ...prev, [name]: false })) + setAiStatus((prev) => ({ ...prev, [name]: 'suggested' })) return { ok: true } } else { - setAiStatus(prev => ({ ...prev, [name]: 'error' })) - return { ok: false, error: data?.detail || data?.error || `Request failed (${res.status})` } + setAiStatus((prev) => ({ ...prev, [name]: 'error' })) + return { + ok: false, + error: + data?.detail || data?.error || `Request failed (${res.status})`, + } } } catch (err: any) { - if (currentCitationKeyRef.current === requestKey) setAiStatus(prev => ({ ...prev, [name]: 'error' })) + if (currentCitationKeyRef.current === requestKey) + setAiStatus((prev) => ({ ...prev, [name]: 'error' })) return { ok: false, error: err?.message || 'Parameter extraction failed' } } } @@ -261,19 +296,14 @@ export default function CanSrL2ScreenPage() { const runAllAI = async () => { if (!parametersParsed || runningAllAI || !srId || !citationId) return const requestKey = `${srId}:${citationId}` - const context = { srId, citationId, model: selectedModel, fullText: null as string | null } - - // Flatten rendered parameters with their descriptions - const params: Array<{ name: string; description: string }> = [] - parametersParsed.categories.forEach((_, i) => { - ;(parametersParsed.possible_parameters[i] || []).forEach((param, j) => { - const desc = parametersParsed.descriptions?.[i]?.[j] || '' - const cleanDesc = desc.replace(/<\/?desc>/g, '') - const paramName = - typeof param === 'string' ? param : Array.isArray(param) ? param[0] : String(param) - params.push({ name: paramName, description: cleanDesc }) - }) - }) + const context = { + srId, + citationId, + model: selectedModel, + fullText: null as string | null, + } + + const params = parametersParsed.items setRunningAllAI(true) setRunAllProgress({ done: 0, total: params.length }) @@ -282,22 +312,27 @@ export default function CanSrL2ScreenPage() { try { // Resolve full text exactly once and reuse it for every parameter request. context.fullText = await ensureFullText() - if (!context.fullText) throw new Error('Full text is not available for this citation') + if (!context.fullText) + throw new Error('Full text is not available for this citation') const failures: string[] = [] for (let idx = 0; idx < params.length; idx++) { if (currentCitationKeyRef.current !== requestKey) break const p = params[idx] const result = await suggestParam(p.name, p.description, context) - if (!result.ok && result.error !== 'Citation changed') failures.push(`${p.name}: ${result.error || 'failed'}`) + if (!result.ok && result.error !== 'Citation changed') + failures.push(`${p.name}: ${result.error || 'failed'}`) if (currentCitationKeyRef.current !== requestKey) break setRunAllProgress({ done: idx + 1, total: params.length }) } if (failures.length) { - setRunAllError(`${failures.length} of ${params.length} parameters failed. ${failures[0]}`) + setRunAllError( + `${failures.length} of ${params.length} parameters failed. ${failures[0]}`, + ) } } catch (err: any) { - if (currentCitationKeyRef.current === requestKey) setRunAllError(err?.message || 'Run All AI failed') + if (currentCitationKeyRef.current === requestKey) + setRunAllError(err?.message || 'Run All AI failed') } finally { if (currentCitationKeyRef.current === requestKey) setRunningAllAI(false) // keep progress around as “done”; user can see it completed @@ -316,22 +351,19 @@ export default function CanSrL2ScreenPage() { const data = await res.json().catch(() => ({})) const parsed = data?.criteria_parsed || data?.criteria || {} const paramsInfo = parsed?.parameters - if (paramsInfo && paramsInfo.categories && paramsInfo.possible_parameters) { - setParametersParsed(paramsInfo) + if (paramsInfo && (paramsInfo.items || paramsInfo.categories)) { + const items = flattenExtractionParameters(paramsInfo) + setParametersParsed({ items }) const defaults: Record = {} - paramsInfo.possible_parameters.forEach((arr: string[]) => { - arr.forEach((name: string) => { - defaults[name] = defaults[name] || '' - }) + items.forEach(({ name }) => { + defaults[name] = defaults[name] || '' }) - setParamValues(prev => ({ ...defaults, ...prev })) + setParamValues((prev) => ({ ...defaults, ...prev })) const defaultFound: Record = {} - paramsInfo.possible_parameters.forEach((arr: string[]) => { - arr.forEach((name: string) => { - defaultFound[name] = defaultFound[name] ?? false - }) + items.forEach(({ name }) => { + defaultFound[name] = defaultFound[name] ?? false }) - setParamFound(prev => ({ ...defaultFound, ...prev })) + setParamFound((prev) => ({ ...defaultFound, ...prev })) } } catch (err) { console.warn('Failed to load parameters', err) @@ -370,27 +402,25 @@ export default function CanSrL2ScreenPage() { } } - parametersParsed.possible_parameters.forEach((arr: string[]) => { - arr.forEach((name: string) => { - const humanCol = humanParamColumn(name) - const llmCol = snakeCaseParamLLM(name) - const humanVal = parseJson((row as any)[humanCol]) - const llmVal = parseJson((row as any)[llmCol]) - - // Prefer human value over LLM for the textbox/found flag - if (humanVal && typeof humanVal === 'object') { - nextFound[name] = !!(humanVal.found ?? humanVal.value) - nextValues[name] = humanVal.value ?? '' - } else if (llmVal && typeof llmVal === 'object') { - nextFound[name] = !!llmVal.found - nextValues[name] = llmVal.value ?? '' - } + parametersParsed.items.forEach(({ name }) => { + const humanCol = humanParamColumn(name) + const llmCol = snakeCaseParamLLM(name) + const humanVal = parseJson((row as any)[humanCol]) + const llmVal = parseJson((row as any)[llmCol]) + + // Prefer human value over LLM for the textbox/found flag + if (humanVal && typeof humanVal === 'object') { + nextFound[name] = !!(humanVal.found ?? humanVal.value) + nextValues[name] = humanVal.value ?? '' + } else if (llmVal && typeof llmVal === 'object') { + nextFound[name] = !!llmVal.found + nextValues[name] = llmVal.value ?? '' + } - // Always populate AI panel with LLM extraction (explanation/evidence) when available - if (llmVal && typeof llmVal === 'object') { - nextAIPanels[name] = llmVal - } - }) + // Always populate AI panel with LLM extraction (explanation/evidence) when available + if (llmVal && typeof llmVal === 'object') { + nextAIPanels[name] = llmVal + } }) setParamFound(nextFound) @@ -398,23 +428,37 @@ export default function CanSrL2ScreenPage() { setAiPanels(nextAIPanels) // extract coords/pages/fulltext and artifacts for PDF overlay - const ft = typeof (row as any).fulltext === 'string' ? (row as any).fulltext : null + const ft = + typeof (row as any).fulltext === 'string' + ? (row as any).fulltext + : null setFulltextStr(ft) fullTextCacheRef.current = ft - const tablesAny = parseJson((row as any).fulltext_tables) ?? (row as any).fulltext_tables + const tablesAny = + parseJson((row as any).fulltext_tables) ?? + (row as any).fulltext_tables setFulltextTables(Array.isArray(tablesAny) ? tablesAny : null) - const figsAny = parseJson((row as any).fulltext_figures) ?? (row as any).fulltext_figures + const figsAny = + parseJson((row as any).fulltext_figures) ?? + (row as any).fulltext_figures setFulltextFigures(Array.isArray(figsAny) ? figsAny : null) - const coordsAny = parseJson((row as any).fulltext_coords) ?? (row as any).fulltext_coords + const coordsAny = + parseJson((row as any).fulltext_coords) ?? + (row as any).fulltext_coords setFulltextCoords(Array.isArray(coordsAny) ? coordsAny : null) - const pagesAny = parseJson((row as any).fulltext_pages) ?? (row as any).fulltext_pages + const pagesAny = + parseJson((row as any).fulltext_pages) ?? (row as any).fulltext_pages setFulltextPages(Array.isArray(pagesAny) ? pagesAny : null) } catch (err: any) { - if (err?.name === 'AbortError' || currentCitationKeyRef.current !== requestKey) return + if ( + err?.name === 'AbortError' || + currentCitationKeyRef.current !== requestKey + ) + return console.warn('Failed to prefill citation params', err) } })() @@ -448,25 +492,38 @@ export default function CanSrL2ScreenPage() { } } - const ft = typeof (row as any).fulltext === 'string' ? (row as any).fulltext : null + const ft = + typeof (row as any).fulltext === 'string' + ? (row as any).fulltext + : null setFulltextStr(ft) fullTextCacheRef.current = ft - const tablesAny = parseJson((row as any).fulltext_tables) ?? (row as any).fulltext_tables + const tablesAny = + parseJson((row as any).fulltext_tables) ?? + (row as any).fulltext_tables setFulltextTables(Array.isArray(tablesAny) ? tablesAny : null) - const figsAny = parseJson((row as any).fulltext_figures) ?? (row as any).fulltext_figures + const figsAny = + parseJson((row as any).fulltext_figures) ?? + (row as any).fulltext_figures setFulltextFigures(Array.isArray(figsAny) ? figsAny : null) - const coordsAny = parseJson((row as any).fulltext_coords) ?? (row as any).fulltext_coords + const coordsAny = + parseJson((row as any).fulltext_coords) ?? + (row as any).fulltext_coords setFulltextCoords(Array.isArray(coordsAny) ? coordsAny : null) - const pagesAny = parseJson((row as any).fulltext_pages) ?? (row as any).fulltext_pages + const pagesAny = + parseJson((row as any).fulltext_pages) ?? (row as any).fulltext_pages setFulltextPages(Array.isArray(pagesAny) ? pagesAny : null) // If coords/pages are missing, trigger backend extraction to populate them, then refetch const needExtract = - !Array.isArray(coordsAny) || coordsAny.length === 0 || !Array.isArray(pagesAny) || pagesAny.length === 0 + !Array.isArray(coordsAny) || + coordsAny.length === 0 || + !Array.isArray(pagesAny) || + pagesAny.length === 0 if (needExtract) { try { @@ -486,23 +543,38 @@ export default function CanSrL2ScreenPage() { const row2 = await res3.json().catch(() => ({})) if (currentCitationKeyRef.current !== requestKey) return - const ft2 = typeof (row2 as any).fulltext === 'string' ? (row2 as any).fulltext : null + const ft2 = + typeof (row2 as any).fulltext === 'string' + ? (row2 as any).fulltext + : null setFulltextStr(ft2) fullTextCacheRef.current = ft2 - const coordsAny2 = parseJson((row2 as any).fulltext_coords) ?? (row2 as any).fulltext_coords + const coordsAny2 = + parseJson((row2 as any).fulltext_coords) ?? + (row2 as any).fulltext_coords setFulltextCoords(Array.isArray(coordsAny2) ? coordsAny2 : null) - const pagesAny2 = parseJson((row2 as any).fulltext_pages) ?? (row2 as any).fulltext_pages + const pagesAny2 = + parseJson((row2 as any).fulltext_pages) ?? + (row2 as any).fulltext_pages setFulltextPages(Array.isArray(pagesAny2) ? pagesAny2 : null) } } catch (err: any) { - if (err?.name === 'AbortError' || currentCitationKeyRef.current !== requestKey) return + if ( + err?.name === 'AbortError' || + currentCitationKeyRef.current !== requestKey + ) + return console.warn('Failed to extract fulltext for overlay', err) } } } catch (err: any) { - if (err?.name === 'AbortError' || currentCitationKeyRef.current !== requestKey) return + if ( + err?.name === 'AbortError' || + currentCitationKeyRef.current !== requestKey + ) + return console.warn('Failed to load citation overlay data', err) } })() @@ -510,13 +582,13 @@ export default function CanSrL2ScreenPage() { }, [srId, citationId]) const updateValue = (name: string, val: string) => { - setParamValues(prev => ({ ...prev, [name]: val })) + setParamValues((prev) => ({ ...prev, [name]: val })) } const saveParam = async (name: string) => { if (!citationId || !srId) return setSavingParam(name) - setSaveStatus(prev => ({ ...prev, [name]: 'saving' })) + setSaveStatus((prev) => ({ ...prev, [name]: 'saving' })) try { const headers = getAuthHeaders() const res = await fetch( @@ -537,9 +609,9 @@ export default function CanSrL2ScreenPage() { }, ) await res.json().catch(() => ({})) - setSaveStatus(prev => ({ ...prev, [name]: res.ok ? 'saved' : 'error' })) + setSaveStatus((prev) => ({ ...prev, [name]: res.ok ? 'saved' : 'error' })) } catch { - setSaveStatus(prev => ({ ...prev, [name]: 'error' })) + setSaveStatus((prev) => ({ ...prev, [name]: 'error' })) } finally { setSavingParam(null) } @@ -568,41 +640,43 @@ export default function CanSrL2ScreenPage() { } /> -
+
{/* Workspace (left) */}
{/*
*/} - {/*
+ {/*

Workspace

This area displays the full text (PDF) and is a flexible workspace for viewing and selecting parameter regions.

*/} - + {/*
*/}
{/* Selection sidebar (right) */}