Skip to content

tools

zocrates.agents.common.tools

ADK FunctionTool adapters for Zocrates core toolkit.

Binds pure business logic functions from toolkit.py to Google ADK session and invocation state (ToolContext), handling: - Scoping to active context items from Zotero UI - Evidence accumulation in temp:retrieved_evidence - Cross-tool deduplication in temp:seen_chunk_ids - Range-Aware Pre-Exclusion - Intra-tool asset deduplication in temp:seen_assets

ACTIVE_CONTEXT_ITEMS module-attribute 

ACTIVE_CONTEXT_ITEMS = 'active_context_items'

CACHED_TOCS module-attribute 

CACHED_TOCS = 'cached_tocs'

HARD_WORD_CEILING module-attribute 

HARD_WORD_CEILING = 7500

MAX_ABSOLUTE_RETRIEVAL_LIMIT module-attribute 

MAX_ABSOLUTE_RETRIEVAL_LIMIT = 20

MAX_RETRIEVAL_TOOLS_PER_RUN module-attribute 

MAX_RETRIEVAL_TOOLS_PER_RUN = 10

MAX_UNIQUE_CACHED_TOCS module-attribute 

MAX_UNIQUE_CACHED_TOCS = 10

TEMP_EXTENSION_COUNT module-attribute 

TEMP_EXTENSION_COUNT = 'temp:extension_count'

TEMP_EXTENSION_REQUESTED module-attribute 

TEMP_EXTENSION_REQUESTED = 'temp:extension_requested'

TEMP_EXTENSION_REQUEST_PASS module-attribute 

TEMP_EXTENSION_REQUEST_PASS = 'temp:extension_request_pass'

TEMP_RETRIEVAL_TOOL_BUDGET module-attribute 

TEMP_RETRIEVAL_TOOL_BUDGET = 'temp:retrieval_tool_budget'

TEMP_RETRIEVAL_TOOL_COUNT module-attribute 

TEMP_RETRIEVAL_TOOL_COUNT = 'temp:retrieval_tool_count'

TEMP_RETRIEVED_EVIDENCE module-attribute 

TEMP_RETRIEVED_EVIDENCE = 'temp:retrieved_evidence'

TEMP_SEARCH_REPORT module-attribute 

TEMP_SEARCH_REPORT = 'temp:search_report'

TEMP_SEEN_ASSETS module-attribute 

TEMP_SEEN_ASSETS = 'temp:seen_assets'

TEMP_SEEN_CHUNK_IDS module-attribute 

TEMP_SEEN_CHUNK_IDS = 'temp:seen_chunk_ids'

TEMP_SEEN_TOC_ITEM_KEYS module-attribute 

TEMP_SEEN_TOC_ITEM_KEYS = 'temp:seen_toc_item_keys'

TEMP_WORD_CEILING module-attribute 

TEMP_WORD_CEILING = 'temp:word_ceiling'

logger module-attribute 

logger = logging.getLogger(__name__)

_check_and_increment_tool_budget 

_check_and_increment_tool_budget(tool_context)

Check whether a retrieval tool call is permitted under budget and word ceilings.

Returns:

Type Description
tuple[bool, str | None, str | None]

tuple of (allowed: bool, error_message: str | None, budget_notice: str | None)
Source code in zocrates/agents/common/tools.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def _check_and_increment_tool_budget(
    tool_context: ToolContext | None,
) -> tuple[bool, str | None, str | None]:
    """Check whether a retrieval tool call is permitted under budget and word ceilings.

    Returns:
        tuple of (allowed: bool, error_message: str | None, budget_notice: str | None)
    """
    if tool_context is None:
        return True, None, None

    budget = int(tool_context.state.get(TEMP_RETRIEVAL_TOOL_BUDGET, MAX_RETRIEVAL_TOOLS_PER_RUN))
    word_ceiling = int(tool_context.state.get(TEMP_WORD_CEILING, HARD_WORD_CEILING))
    extension_count = int(tool_context.state.get(TEMP_EXTENSION_COUNT, 0))

    hitl_enabled = True
    max_extensions = 2
    trigger_pct = 0.8
    try:
        from zocrates.config import get_search_harness_config
        h_cfg = get_search_harness_config()
        hitl_enabled = bool(h_cfg.get("hitl_enabled", True))
        max_extensions = int(h_cfg.get("max_extension_count", 2))
        trigger_pct = float(h_cfg.get("word_ceiling_hitl_trigger_percent", 80)) / 100.0
    except Exception:
        pass

    can_extend = hitl_enabled and (extension_count < max_extensions)

    evidence = tool_context.state.get(TEMP_RETRIEVED_EVIDENCE, [])
    total_words = sum(
        c.get("word_count") if c.get("word_count") is not None else len(c.get("text", "").split())
        for c in evidence
    )
    if total_words >= word_ceiling:
        _ensure_preliminary_report(tool_context)
        action_msg = (
            "Call `request_search_extension` to request an extension, or `conclude_search` to finish."
            if can_extend
            else "Call `conclude_search` immediately to conclude retrieval."
        )
        return (
            False,
            (
                f"Word ceiling reached (~{total_words}/{word_ceiling} words). "
                f"You cannot call more retrieval tools. {action_msg}"
            ),
            None,
        )

    count = int(tool_context.state.get(TEMP_RETRIEVAL_TOOL_COUNT, 0))
    if count >= budget:
        _ensure_preliminary_report(tool_context)
        action_msg = (
            "Call `request_search_extension` to request an extension, or `conclude_search` to finish."
            if can_extend
            else "Call `conclude_search` immediately to conclude retrieval."
        )
        return (
            False,
            (
                f"Retrieval tool budget reached ({budget}/"
                f"{budget}). You cannot call more retrieval tools. {action_msg}"
            ),
            None,
        )

    new_count = count + 1
    tool_context.state[TEMP_RETRIEVAL_TOOL_COUNT] = new_count

    notice: str | None = None
    if new_count == budget:
        _ensure_preliminary_report(tool_context)
        action_msg = (
            "Call `request_search_extension` to request an extension, or `conclude_search` to conclude."
            if can_extend
            else "You must call `conclude_search` now to conclude retrieval."
        )
        notice = (
            f"[Notice: Retrieval tool budget exhausted ({budget}/{budget}). {action_msg}]"
        )
    elif new_count == (budget - 1):
        notice = (
            f"[Notice: {new_count}/{budget} retrieval tool calls used (1 call remaining). "
            f"Plan your final retrieval, request an extension via `request_search_extension`, or prepare to call `conclude_search`.]"
        )
    elif total_words >= int(trigger_pct * word_ceiling):
        notice = (
            f"[Notice: Evidence volume approaching ceiling (~{total_words}/{word_ceiling} words). "
            f"If crucial aspects remain uncovered, consider calling request_search_extension or targeting only essential sections.]"
        )
    elif new_count >= int(budget / 2):
        notice = (
            f"[Notice: {new_count}/{budget} retrieval tool calls used. "
            f"Continue your search, being mindful of your remaining retrieval budget as you proceed.]"
        )

    return True, None, notice

_ensure_preliminary_report 

_ensure_preliminary_report(tool_context, force=False)

Auto-generate preliminary report if not already set (or forced upon tool completion) when budget or word ceiling is exhausted.

Source code in zocrates/agents/common/tools.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def _ensure_preliminary_report(tool_context: ToolContext | None, force: bool = False) -> None:
    """Auto-generate preliminary report if not already set (or forced upon tool completion) when budget or word ceiling is exhausted."""
    if tool_context is None:
        return
    if not force and tool_context.state.get(TEMP_SEARCH_REPORT) is not None:
        return

    evidence = tool_context.state.get(TEMP_RETRIEVED_EVIDENCE, [])
    total_words = sum(
        c.get("word_count") if c.get("word_count") is not None else len(c.get("text", "").split())
        for c in evidence
    )
    if not evidence:
        report = {
            "summary": f"Retrieval tool limit reached ({MAX_RETRIEVAL_TOOLS_PER_RUN}/{MAX_RETRIEVAL_TOOLS_PER_RUN}) with 0 chunks retrieved.",
            "notes": "No evidence was retrieved.",
            "gaps": [],
            "source": "budget_exhaustion",
        }
    else:
        unique_headings: list[str] = []
        seen = set()
        for c in evidence:
            h = c.get("heading")
            if h and h not in seen:
                seen.add(h)
                unique_headings.append(h)
        chunk_types = sorted(list({c.get("chunk_type", "text") for c in evidence}))
        top_headings = unique_headings[:5]
        count = tool_context.state.get(TEMP_RETRIEVAL_TOOL_COUNT, 0)
        report = {
            "summary": (
                f"Retrieved {len(evidence)} chunks ({', '.join(chunk_types)}) "
                f"covering: {', '.join(top_headings) if top_headings else 'general content'} (~{total_words} words)."
            ),
            "notes": "Concluded via retrieval budget limit.",
            "gaps": [f"Unchecked literature beyond budget limit (~{total_words} words, {count} calls)"],
            "source": "budget_exhaustion",
        }
    tool_context.state[TEMP_SEARCH_REPORT] = report

_prune_cached_tocs 

_prune_cached_tocs(cached_tocs, max_entries=MAX_UNIQUE_CACHED_TOCS)

Keep only the most recently updated unique entries, pruning by unique bundle identity.

Source code in zocrates/agents/common/toolkit.py 
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
def _prune_cached_tocs(
    cached_tocs: dict[str, Any],
    max_entries: int = MAX_UNIQUE_CACHED_TOCS,
) -> dict[str, Any]:
    """Keep only the most recently updated unique entries, pruning by unique bundle identity."""
    unique_entries: dict[str, dict[str, Any]] = {}
    for k, v in cached_tocs.items():
        if isinstance(v, dict):
            res_key = v.get("resolved_key", k)
            if res_key not in unique_entries or v.get("updated_at", 0) > unique_entries[res_key].get("updated_at", 0):
                unique_entries[res_key] = v

    if len(unique_entries) <= max_entries:
        return cached_tocs

    sorted_unique = sorted(
        unique_entries.values(),
        key=lambda x: x.get("updated_at", 0),
        reverse=True,
    )[:max_entries]

    pruned: dict[str, Any] = {}
    for entry in sorted_unique:
        in_key = entry.get("input_key")
        res_key = entry.get("resolved_key")
        if in_key:
            pruned[in_key] = entry
        if res_key:
            pruned[res_key] = entry
    return pruned

_resolve_bundle_dir 

_resolve_bundle_dir(item_key)

Resolve bundle directory, active key, and optional disambiguation notice.

Three-step resolution: 1. Note on-demand: item_key matches a Zotero note → materialise its bundle first so staleness is always validated via the SHA-256 fast-path. 2. Direct check: cache_dir / item_key has document.md or toc.md. 3. Parent key fallback: look up converted child attachments (PDF-first sort); if none, fall back to materialising the first child note.

Parameters:

Name Type Description Default
item_key str

Zotero item key.

 required

Returns:

Type Description
tuple[Path, str, str | None] | None

(bundle_dir, resolved_key, notice) where notice is a non-empty
tuple[Path, str, str | None] | None

string when the parent item has multiple artifacts and the caller should
tuple[Path, str, str | None] | None

be informed about alternatives, or None if everything is unambiguous.
tuple[Path, str, str | None] | None

Returns None altogether when no bundle can be resolved.
Source code in zocrates/agents/common/toolkit.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
def _resolve_bundle_dir(item_key: str) -> tuple[Path, str, str | None] | None:
    """Resolve bundle directory, active key, and optional disambiguation notice.

    Three-step resolution:
    1. Note on-demand: ``item_key`` matches a Zotero note → materialise its
       bundle first so staleness is always validated via the SHA-256 fast-path.
    2. Direct check: ``cache_dir / item_key`` has ``document.md`` or ``toc.md``.
    3. Parent key fallback: look up converted child attachments (PDF-first sort);
       if none, fall back to materialising the first child note.

    Args:
        item_key: Zotero item key.

    Returns:
        ``(bundle_dir, resolved_key, notice)`` where ``notice`` is a non-empty
        string when the parent item has multiple artifacts and the caller should
        be informed about alternatives, or ``None`` if everything is unambiguous.
        Returns ``None`` altogether when no bundle can be resolved.
    """
    if not _validate_item_key(item_key):
        return None

    syncer = get_syncer()
    cache_dir = syncer.processor._cache_dir
    bundle_dir = cache_dir / item_key

    # Liveness check: If syncer.db is present and not syncer.db.is_item_active(item_key):
    #   Do NOT return bundle or materialize! Return None immediately so parent-deleted notes
    #   and directly-deleted items are not materialized or queried.
    if syncer.db and not syncer.db.is_item_active(item_key):
        return None

    # Step 1: Check whether item_key is a note first, so staleness is always
    # validated via materialize_note_bundle's SHA-256 fast-path even when a
    # bundle directory already exists on disk. On failure (or a non-note key)
    # execution falls through to the direct bundle check below.
    try:
        note = syncer.db.get_note_by_key(item_key)
    except Exception:
        note = None
    if note is not None and getattr(note, "note_key", None):
        try:
            from zocrates.rag.document_processor import materialize_note_bundle  # local import avoids cycles
            note_dir, resolved_note_key = materialize_note_bundle(note, syncer, cache_dir)
            return note_dir, resolved_note_key, None
        except Exception as exc:
            logger.error("Failed to materialise note bundle for %s: %s", item_key, exc)

    # Step 2: Direct bundle check (attachment or already-resolved key).
    if (bundle_dir / "document.md").exists() or (bundle_dir / "toc.md").exists():
        if bundle_dir.resolve().is_relative_to(cache_dir.resolve()):
            return bundle_dir, item_key, None

    # Step 3: Parent key fallback.
    item = syncer.db.get_item_by_key(item_key)
    if not item:
        return None

    attachments = syncer.db.get_attachments(item.item_id)
    converted_atts = [
        att for att in attachments
        if (cache_dir / att.item_key / "document.md").exists()
        and (cache_dir / att.item_key).resolve().is_relative_to(cache_dir.resolve())
    ]
    converted_atts.sort(key=lambda a: _CONTENT_TYPE_PRIORITY.get(a.content_type, 99))

    # Also gather child notes for disambiguation notice construction.
    try:
        notes = syncer.db.get_notes_for_item(item.item_id)
    except Exception:
        notes = []

    if converted_atts:
        primary = converted_atts[0]
        primary_dir = cache_dir / primary.item_key
        primary_name = primary.title or (primary.path.name if primary.path else primary.item_key)

        # Build notice when siblings exist.
        other_parts: list[str] = []
        for att in converted_atts[1:]:
            att_name = att.title or (att.path.name if att.path else att.item_key)
            other_parts.append(f"[Attachment Key: {att.item_key}] {att_name!r} [{att.content_type}]")
        for n in notes:
            n_name = n.title or "Note"
            other_parts.append(f"[Note Key: {n.note_key}] {n_name!r} [text/note]")

        notice: str | None = None
        if other_parts:
            notice = (
                f"Loaded primary attachment {primary_name!r}. "
                f"This item has additional artifacts: {', '.join(other_parts)}. "
                f"Call get_document_toc with a specific key to inspect them."
            )

        return primary_dir, primary.item_key, notice

    # No converted attachments: materialise the first note as last resort.
    if notes:
        try:
            from zocrates.rag.document_processor import materialize_note_bundle
            note_dir, resolved_note_key = materialize_note_bundle(notes[0], syncer, cache_dir)

            fallback_notice: str | None = None
            if len(notes) > 1:
                other_notes = [
                    f"[Note Key: {n.note_key}] {(n.title or 'Note')!r} [text/note]"
                    for n in notes[1:]
                ]
                fallback_notice = (
                    f"Loaded note {(notes[0].title or 'Note')!r}. "
                    f"This item has additional notes: {', '.join(other_notes)}. "
                    f"Call get_document_toc with a specific key to inspect them."
                )

            return note_dir, notes[0].note_key, fallback_notice
        except Exception as exc:
            logger.error(
                "Failed to materialise note bundle for %s (last-resort): %s",
                notes[0].note_key,
                exc,
            )

    return None

_validate_item_key 

_validate_item_key(item_key)

Validate item key to prevent path traversal.

Source code in zocrates/agents/common/toolkit.py 
31
32
33
def _validate_item_key(item_key: str) -> bool:
    """Validate item key to prevent path traversal."""
    return bool(item_key and _SAFE_KEY_RE.match(item_key))

conclude_search(summary, notes=None, gaps=None, tool_context=None)

Conclude the retrieval phase and hand off accumulated evidence to the synthesizer.

Call this tool as soon as you have collected the necessary evidence, or immediately upon exhausting your retrieval tool budget. Do NOT write final answers or syntheses.

Parameters:

Name Type Description Default
summary str

1-2 sentence overview of the evidence gathered.

 required
notes str | None

Optional observations regarding caveats or overall synthesis guidance.

 None
gaps list[str] | None

Optional list of missing facets or literature gaps. If there are no gaps, leave this parameter unspecified.

 None

Returns:

Type Description
dict[str, Any]

Confirmation dictionary indicating search has concluded.
Source code in zocrates/agents/common/tools.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def conclude_search(
    summary: str,
    notes: str | None = None,
    gaps: list[str] | None = None,
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Conclude the retrieval phase and hand off accumulated evidence to the synthesizer.

    Call this tool as soon as you have collected the necessary evidence, or immediately
    upon exhausting your retrieval tool budget. Do NOT write final answers or syntheses.

    Args:
        summary: 1-2 sentence overview of the evidence gathered.
        notes: Optional observations regarding caveats or overall synthesis guidance.
        gaps: Optional list of missing facets or literature gaps.
            If there are no gaps, leave this parameter unspecified.

    Returns:
        Confirmation dictionary indicating search has concluded.
    """
    if isinstance(gaps, str):
        gap_list = [g.strip() for g in re.split(r"[,;\n]+", gaps) if g.strip()]
    elif isinstance(gaps, (list, tuple, set)):
        gap_list = [str(g).strip() for g in gaps if g]
    else:
        gap_list = []

    report = {
        "summary": str(summary or "").strip(),
        "notes": str(notes or "").strip() if notes else None,
        "gaps": gap_list,
        "source": "subagent",
    }
    if tool_context is not None:
        tool_context.state[TEMP_SEARCH_REPORT] = report
        if getattr(tool_context, "actions", None) is not None:
            tool_context.actions.skip_summarization = True

    return {
        "status": "search_concluded",
        "message": "Retrieval phase concluded. Accumulated evidence will now be synthesized.",
        "report": report,
    }

extract_compact_toc_outline 

extract_compact_toc_outline(bundle_dir, title='', resolved_key='', parent_key='')

Extract a concise navigation outline from a document bundle.

Strips the long abstract and retains structured sections (with level-folding for dense documents) and asset ranges for figures and tables.

Source code in zocrates/agents/common/toolkit.py 
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
def extract_compact_toc_outline(
    bundle_dir: Path,
    title: str = "",
    resolved_key: str = "",
    parent_key: str = "",
) -> str:
    """Extract a concise navigation outline from a document bundle.

    Strips the long abstract and retains structured sections (with level-folding for
    dense documents) and asset ranges for figures and tables.
    """
    toc_json_path = bundle_dir / "toc.json"
    if toc_json_path.exists():
        try:
            with open(toc_json_path, encoding="utf-8") as f:
                data = json.load(f)

            effective_title = (title or data.get("title") or "").strip()
            key_label = resolved_key or data.get("item_key", "")
            eff_parent_key = parent_key or data.get("parent_item_key") or ""
            if eff_parent_key and eff_parent_key != key_label:
                key_bracket = f"[Key: {key_label}, Parent Key: {eff_parent_key}]"
            else:
                key_bracket = f"[Key: {key_label}]"

            title_part = f' "{effective_title}"' if effective_title else ""
            lines = [f"### {key_bracket}{title_part}"]

            sections = data.get("sections") or []

            # Abstract line injection if available in metadata and not already an explicit section
            abstract_text = (data.get("abstract") or "").strip()
            abs_source = data.get("abstract_source") or ""
            if abstract_text and abstract_text != "No abstract or summary available." and abs_source != "none":
                has_abs_sec = any(
                    str(s.get("heading", "")).lower().strip() in ("abstract", "summary")
                    for s in sections
                )
                if not has_abs_sec:
                    lines.append(f"- [abstract] Abstract / Summary (~{len(abstract_text.split())} words)")

            if len(sections) <= 15:
                filtered_sections = sections
            else:
                filtered_sections = [s for s in sections if s.get("level", 1) <= 2]

            for sec in filtered_sections:
                s_id = sec.get("id", "")
                heading = sec.get("heading", "")
                words = sec.get("word_count", 0)
                lines.append(f"- [{s_id}] {heading} (~{words} words)")

            figures = data.get("figures") or []
            if figures:
                if len(figures) <= 8:
                    lines.append("- Figures:")
                    for f in figures:
                        caption = (f.get("caption") or "").strip()
                        lines.append(f"  - [{f.get('id')}] {caption}")
                else:
                    first_id = figures[0].get("id", "f1")
                    last_id = figures[-1].get("id", f"f{len(figures)}")
                    lines.append(
                        f"- Figures: {len(figures)} figures present ([{first_id}] to [{last_id}]). "
                        "Can be directly read via read_document(asset_id='fX') or discovered via search_corpus."
                    )

            tables = data.get("tables") or []
            if tables:
                if len(tables) <= 6:
                    lines.append("- Tables:")
                    for t in tables:
                        caption = (t.get("caption") or "").strip()
                        lines.append(f"  - [{t.get('id')}] {caption}")
                else:
                    first_id = tables[0].get("id", "t1")
                    last_id = tables[-1].get("id", f"t{len(tables)}")
                    lines.append(
                        f"- Tables: {len(tables)} tables present ([{first_id}] to [{last_id}]). "
                        "Can be directly read via read_document(asset_id='tX')."
                    )

            return "\n".join(lines)
        except Exception as exc:
            logger.warning("Failed to parse toc.json in %s: %s", bundle_dir, exc)

    # Fallback to toc.md if toc.json is missing or corrupted
    eff_parent_key = parent_key
    if eff_parent_key and eff_parent_key != resolved_key:
        key_bracket = f"[Key: {resolved_key}, Parent Key: {eff_parent_key}]"
    else:
        key_bracket = f"[Key: {resolved_key}]"

    toc_md_path = bundle_dir / "toc.md"
    if toc_md_path.exists():
        try:
            content = toc_md_path.read_text(encoding="utf-8")
            effective_title = title.strip()
            title_part = f' "{effective_title}"' if effective_title else ""
            lines = [f"### {key_bracket}{title_part}"]

            # Preserve content starting from ## Table of Contents
            if "## Table of Contents" in content:
                toc_body = content.split("## Table of Contents", 1)[1]
                for raw_line in toc_body.splitlines():
                    stripped = raw_line.strip()
                    if stripped.startswith("- ["):
                        lines.append(stripped)
                    elif stripped.startswith("## Figures") or stripped.startswith("## Tables"):
                        lines.append(f"- {stripped.lstrip('#').strip()}")
                    elif stripped and not stripped.startswith("#"):
                        lines.append(f"  {stripped}")
            return "\n".join(lines)
        except Exception as exc:
            logger.warning("Failed to parse toc.md fallback in %s: %s", bundle_dir, exc)

    title_part = f' "{title}"' if title else ""
    return f"### {key_bracket}{title_part}"

get_common_tools 

get_common_tools()

Return the list of core ADK tools for research agents.

Source code in zocrates/agents/common/tools.py 
835
836
837
838
839
840
841
842
843
844
def get_common_tools() -> list[Any]:
    """Return the list of core ADK tools for research agents."""
    return [
        search_corpus,
        get_document_toc,
        read_document,
        get_related_items,
        conclude_search,
        request_search_extension,
    ]

get_document_toc 

get_document_toc(item_key, tool_context=None)

Retrieve the Title, Abstract, Table of Contents, Figures, and Tables for a given item.

Parameters:

Name Type Description Default
item_key str

Unique Zotero item key.

 required

Returns:

Type Description
dict[str, Any]

Dictionary with 'status', 'item_key', and 'toc_content'.
Source code in zocrates/agents/common/tools.py 
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
def get_document_toc(
    item_key: str,
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Retrieve the Title, Abstract, Table of Contents, Figures, and Tables for a given item.

    Args:
        item_key: Unique Zotero item key.

    Returns:
        Dictionary with 'status', 'item_key', and 'toc_content'.
    """
    allowed, err_msg, notice = _check_and_increment_tool_budget(tool_context)
    if not allowed:
        return {"status": "budget_exhausted", "error": err_msg}

    # Cycle suppression check for item_key before disk reading
    if tool_context is not None:
        seen_tocs = tool_context.state.get(TEMP_SEEN_TOC_ITEM_KEYS, [])
        if item_key in seen_tocs:
            res: dict[str, Any] = {
                "status": "already_inspected",
                "item_key": item_key,
                "message": (
                    f"TOC for {item_key} was already retrieved in this turn. "
                    "Refer to prior context or use read_document for specific sections."
                ),
            }
            if notice:
                _ensure_preliminary_report(tool_context, force=True)
                res["budget_notice"] = notice
            return res

    # Resolve bundle_dir and canonical resolved_key
    resolved_key = item_key
    bundle_dir = None
    try:
        resolved = _resolve_bundle_dir(item_key)
        if resolved:
            bundle_dir, r_key, _ = resolved
            resolved_key = r_key or resolved_key
    except Exception as exc:
        logger.debug("Failed to resolve bundle dir for TOC: %s", exc)

    # Check cached_tocs liveness: evict if item was deleted in Zotero
    if tool_context is not None:
        # ``or {}`` guards against an explicit None persisted in session state
        # (``.get(key, {})`` returns None when the key exists with a None value).
        cached_tocs = dict(tool_context.state.get(CACHED_TOCS) or {})
        if item_key in cached_tocs or resolved_key in cached_tocs:
            try:
                from zocrates.agents.common.toolkit import get_syncer
                syncer = get_syncer()
                is_active = syncer.db.is_item_active(resolved_key) if (syncer and syncer.db) else True
            except Exception:
                is_active = True
            if not is_active:
                cached_tocs.pop(item_key, None)
                cached_tocs.pop(resolved_key, None)
                tool_context.state[CACHED_TOCS] = cached_tocs
                return {
                    "status": "error",
                    "item_key": item_key,
                    "error": f"Document '{item_key}' has been deleted from the library.",
                }

    # Cycle suppression check for canonical resolved_key before disk reading
    if tool_context is not None:
        seen_tocs = tool_context.state.get(TEMP_SEEN_TOC_ITEM_KEYS, [])
        if resolved_key in seen_tocs:
            res = {
                "status": "already_inspected",
                "item_key": item_key,
                "message": (
                    f"TOC for {item_key} was already retrieved in this turn. "
                    "Refer to prior context or use read_document for specific sections."
                ),
            }
            if notice:
                _ensure_preliminary_report(tool_context, force=True)
                res["budget_notice"] = notice
            return res

    res = get_document_toc_core(item_key=item_key)

    if isinstance(res, dict) and "item_key" in res:
        resolved_key = res.get("item_key") or resolved_key

    if res.get("status") == "success" and tool_context is not None:
        # Track seen TOC keys for cycle suppression
        seen_tocs = list(tool_context.state.get(TEMP_SEEN_TOC_ITEM_KEYS, []))
        for k in (item_key, resolved_key):
            if k and k not in seen_tocs:
                seen_tocs.append(k)
        abstract_text = ""
        if bundle_dir:
            toc_json_path = bundle_dir / "toc.json"
            if toc_json_path.exists():
                try:
                    import json
                    with open(toc_json_path, encoding="utf-8") as f:
                        toc_data = json.load(f)
                        pk = toc_data.get("parent_item_key")
                        if pk and pk not in seen_tocs:
                            seen_tocs.append(pk)
                        raw_abs = (toc_data.get("abstract") or "").strip()
                        if raw_abs and raw_abs != "No abstract or summary available." and toc_data.get("abstract_source") != "none":
                            abstract_text = raw_abs
                except Exception:
                    pass
        tool_context.state[TEMP_SEEN_TOC_ITEM_KEYS] = seen_tocs

        input_key = item_key
        doc_title = res.get("title") or ""
        parent_key = input_key if input_key != resolved_key else ""

        # Accumulate retrieved abstract into TEMP_RETRIEVED_EVIDENCE so the synthesizer
        # is never starved of evidence when the search agent inspects TOC / abstract.
        if not abstract_text and res.get("toc_content"):
            match = re.search(r"## Abstract\s*\n(.*?)(?=\n## |\Z)", str(res["toc_content"]), re.DOTALL)
            if match:
                raw_abs = match.group(1).strip()
                if raw_abs and raw_abs != "No abstract or summary available.":
                    abstract_text = raw_abs

        if abstract_text:
            doc_authors = None
            doc_year = None
            if bundle_dir:
                try:
                    from zocrates.agents.common.toolkit import get_syncer
                    syncer = get_syncer()
                    if syncer and getattr(syncer, "processor", None):
                        meta = syncer.processor.get_metadata(resolved_key) or {}
                        doc_authors = meta.get("authors") or meta.get("creators")
                        doc_year = meta.get("year")
                except Exception:
                    pass
            if doc_authors is None or doc_year is None:
                active_items = tool_context.state.get(ACTIVE_CONTEXT_ITEMS) or []
                for item in active_items:
                    if isinstance(item, dict):
                        keys = [item.get("key"), item.get("item_key")]
                        for att in item.get("attachments", []):
                            keys.append(att.get("key"))
                        if resolved_key in keys or item_key in keys:
                            if doc_authors is None:
                                doc_authors = item.get("authors") or item.get("creators")
                            if doc_year is None:
                                doc_year = item.get("year")
                            break

            current_evidence = list(tool_context.state.get(TEMP_RETRIEVED_EVIDENCE, []))
            synthetic_chunk_id = f"{resolved_key}:abstract"
            if not any(c.get("chunk_id") == synthetic_chunk_id for c in current_evidence):
                chunk_dict: dict[str, Any] = {
                    "chunk_id": synthetic_chunk_id,
                    "item_key": resolved_key,
                    "title": doc_title or res.get("title", ""),
                    "chunk_type": "text",
                    "heading": "Abstract",
                    "text": abstract_text,
                    "word_count": len(abstract_text.split()),
                    "retrieval_method": "direct_read",
                    "cosine_similarity": None,
                    "rrf_score": None,
                    "rerank_score": None,
                    "image_path": None,
                    "structured_content": None,
                }
                if doc_authors is not None:
                    chunk_dict["authors"] = doc_authors
                if doc_year is not None:
                    chunk_dict["year"] = doc_year
                current_evidence.append(chunk_dict)
                tool_context.state[TEMP_RETRIEVED_EVIDENCE] = current_evidence
        outline_text = ""
        if bundle_dir:
            try:
                outline_text = extract_compact_toc_outline(
                    bundle_dir,
                    doc_title,
                    resolved_key,
                    parent_key=parent_key,
                )
            except Exception as exc:
                logger.debug("Failed to extract compact toc outline: %s", exc)

        if not outline_text and res.get("toc_content"):
            eff_pk = parent_key if parent_key != resolved_key else ""
            key_bracket = f"[Key: {resolved_key}, Parent Key: {eff_pk}]" if eff_pk else f"[Key: {resolved_key}]"
            title_part = f' "{doc_title}"' if doc_title else ""
            lines = [f"### {key_bracket}{title_part}"]
            for raw_line in str(res["toc_content"]).splitlines():
                s = raw_line.strip()
                if s.startswith("- [") or s.startswith("## Figures") or s.startswith("## Tables"):
                    lines.append(s)
            outline_text = "\n".join(lines)

        if outline_text:
            cached_entry = {
                "input_key": input_key,
                "resolved_key": resolved_key,
                "title": doc_title,
                "outline_text": outline_text,
                "updated_at": time.time(),
            }

            # ``or {}`` guards against an explicit None persisted in session state
            # (see matching comment above).
            cached_tocs = dict(tool_context.state.get(CACHED_TOCS) or {})
            cached_tocs[input_key] = cached_entry
            if resolved_key != input_key:
                cached_tocs[resolved_key] = cached_entry

            pruned_tocs = _prune_cached_tocs(cached_tocs, max_entries=MAX_UNIQUE_CACHED_TOCS)
            tool_context.state[CACHED_TOCS] = pruned_tocs

    # Dynamic TOC relation count injection (P1-3)
    if isinstance(res, dict) and res.get("status") == "success" and "toc_content" in res:
        rel_count = 0
        cache_dir = bundle_dir.parent if bundle_dir else None
        if not cache_dir:
            try:
                from zocrates.agents.common.toolkit import get_syncer
                syncer = get_syncer()
                if syncer and getattr(syncer, "processor", None):
                    cache_dir = getattr(syncer.processor, "_cache_dir", None)
            except Exception:
                pass

        if cache_dir:
            meta_db_path = cache_dir / "conversion_meta.db"
            if meta_db_path.exists():
                try:
                    with contextlib.closing(sqlite3.connect(str(meta_db_path), timeout=5.0)) as meta_conn:
                        meta_conn.execute("PRAGMA busy_timeout = 5000;")
                        row = meta_conn.execute(
                            "SELECT COUNT(DISTINCT target_key) FROM item_relations WHERE source_key = ?",
                            (resolved_key,),
                        ).fetchone()
                        if row:
                            rel_count = row[0]
                except Exception as exc:
                    logger.debug("Failed to query item_relations count for %s: %s", resolved_key, exc)

        toc_str = str(res["toc_content"])
        rel_line = f"**Related Items**: {rel_count}"
        if "## Table of Contents" in toc_str:
            idx = toc_str.find("## Table of Contents")
            prefix = toc_str[:idx].rstrip()
            suffix = toc_str[idx:]
            res["toc_content"] = (prefix + "\n\n" if prefix else "") + rel_line + "\n\n" + suffix
        elif "## Sections" in toc_str:
            idx = toc_str.find("## Sections")
            prefix = toc_str[:idx].rstrip()
            suffix = toc_str[idx:]
            res["toc_content"] = (prefix + "\n\n" if prefix else "") + rel_line + "\n\n" + suffix
        else:
            res["toc_content"] = toc_str.rstrip() + f"\n\n{rel_line}\n"

    if notice and tool_context is not None:
        _ensure_preliminary_report(tool_context, force=True)
        res["budget_notice"] = notice
    return res

get_document_toc_core 

get_document_toc_core(item_key)

Retrieve the human- and agent-readable Table of Contents, Title, Abstract, Figures, and Tables.

Parameters:

Name Type Description Default
item_key str

Unique Zotero item key.

 required

Returns:

Type Description
dict[str, Any]

Dictionary with 'status', 'item_key', and 'toc_content'.
Source code in zocrates/agents/common/toolkit.py 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
def get_document_toc_core(item_key: str) -> dict[str, Any]:
    """Retrieve the human- and agent-readable Table of Contents, Title, Abstract, Figures, and Tables.

    Args:
        item_key: Unique Zotero item key.

    Returns:
        Dictionary with 'status', 'item_key', and 'toc_content'.
    """
    if not _validate_item_key(item_key):
        return {"status": "error", "message": f"Invalid item key: {item_key}"}

    try:
        resolved = _resolve_bundle_dir(item_key)
        if not resolved:
            return {
                "status": "error",
                "message": f"Document bundle for {item_key} not found or not yet converted.",
            }

        bundle_dir, resolved_key, res_notice = resolved
        toc_md_path = bundle_dir / "toc.md"
        if toc_md_path.exists():
            result: dict[str, Any] = {
                "status": "success",
                "item_key": resolved_key,
                "toc_content": toc_md_path.read_text(encoding="utf-8"),
            }
            if res_notice:
                result["notice"] = res_notice
            return result

        # Lazy generation fallback if document.md exists but toc.md was not yet generated
        doc_md_path = bundle_dir / "document.md"
        if doc_md_path.exists():
            syncer = get_syncer()
            doc_text = doc_md_path.read_text(encoding="utf-8")
            meta = syncer.processor.get_metadata(resolved_key)
            title = meta.get("title") if meta else None
            parent_item_key = meta.get("parent_item_key") if meta else None
            from zocrates.rag.toc_generator import generate_toc, save_toc_files

            toc_json, toc_md = generate_toc(
                doc_text,
                title=title,
                item_key=resolved_key,
                parent_item_key=parent_item_key,
            )
            save_toc_files(bundle_dir, toc_json, toc_md)
            result = {
                "status": "success",
                "item_key": resolved_key,
                "toc_content": toc_md,
            }
            if res_notice:
                result["notice"] = res_notice
            return result

        return {
            "status": "error",
            "message": f"Document bundle for {item_key} has no document.md or toc.md.",
        }
    except Exception as exc:
        logger.error("Failed in get_document_toc_core for %s: %s", item_key, exc)
        return {"status": "error", "message": str(exc)}

get_related_items(item_key, tool_context=None)

Retrieve all verified related papers for a given document from the knowledge graph.

Parameters:

Name Type Description Default
item_key str

Unique Zotero item or attachment key.

 required

Returns:

Type Description
dict[str, Any]

Dictionary with status, item_key, total_count, and list of related items.
Source code in zocrates/agents/common/tools.py 
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
def get_related_items(
    item_key: str,
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Retrieve all verified related papers for a given document from the knowledge graph.

    Args:
        item_key: Unique Zotero item or attachment key.

    Returns:
        Dictionary with status, item_key, total_count, and list of related items.
    """
    allowed, err_msg, notice = _check_and_increment_tool_budget(tool_context)
    if not allowed:
        return {"status": "budget_exhausted", "error": err_msg}

    res = get_related_items_core(item_key=item_key, ceiling=25)

    if notice and tool_context is not None:
        _ensure_preliminary_report(tool_context, force=True)
        res["budget_notice"] = notice

    return res

get_related_items_core(item_key, ceiling=25)

Retrieve verified related papers for a given document from the knowledge graph.

Parameters:

Name Type Description Default
item_key str

Unique Zotero item or attachment key.

 required
ceiling int

Maximum number of related items to return (default: 25).

 25

Returns:

Type Description
dict[str, Any]

Dictionary with status, item_key, total_count, and list of related items:
dict[str, Any]

{ "status": "success", "item_key": resolved_key, "total_count": len(related_items), "related_items": [ { "item_key": t_key, "title": title, "relation_type": rel_type, "score": score, "parent_item_key": parent_key, }, ... ],

Optional: "notice": disambiguation notice if res_notice present
dict[str, Any]

}
Source code in zocrates/agents/common/toolkit.py 
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
def get_related_items_core(
    item_key: str,
    ceiling: int = 25,
) -> dict[str, Any]:
    """Retrieve verified related papers for a given document from the knowledge graph.

    Args:
        item_key: Unique Zotero item or attachment key.
        ceiling: Maximum number of related items to return (default: 25).

    Returns:
        Dictionary with status, item_key, total_count, and list of related items:
        {
            "status": "success",
            "item_key": resolved_key,
            "total_count": len(related_items),
            "related_items": [
                {
                    "item_key": t_key,
                    "title": title,
                    "relation_type": rel_type,
                    "score": score,
                    "parent_item_key": parent_key,
                }, ...
            ],
            # Optional: "notice": disambiguation notice if res_notice present
        }
    """
    if not _validate_item_key(item_key):
        return {"status": "error", "message": f"Invalid item key: {item_key}"}

    try:
        resolved = _resolve_bundle_dir(item_key)
        if not resolved:
            return {
                "status": "error",
                "message": f"Document bundle for {item_key} not found or not yet converted.",
            }

        bundle_dir, resolved_key, res_notice = resolved

        cache_dir = bundle_dir.parent if bundle_dir else None
        if not cache_dir:
            try:
                syncer = get_syncer()
                if syncer and getattr(syncer, "processor", None):
                    cache_dir = getattr(syncer.processor, "_cache_dir", None)
            except Exception:
                pass

        related_items: list[dict[str, Any]] = []

        if cache_dir:
            meta_db_path = cache_dir / "conversion_meta.db"
            if meta_db_path.exists():
                try:
                    with contextlib.closing(sqlite3.connect(str(meta_db_path), timeout=5.0)) as conn:
                        conn.row_factory = sqlite3.Row
                        conn.execute("PRAGMA busy_timeout = 5000;")
                        try:
                            rows = conn.execute(
                                """
                                SELECT target_key, score, relation_type
                                FROM item_relations
                                WHERE source_key = ?
                                ORDER BY score DESC, target_key ASC
                                """,
                                (resolved_key,),
                            ).fetchall()
                        except sqlite3.OperationalError:
                            rows = []

                        # Coalesce multiple relation types for the same target_key using strict precedence:
                        # human_curated_and_hybrid_search (1.0) > human_curated (1.0) > hybrid_search > shared_tags (0.70)
                        best_by_target: dict[str, dict[str, Any]] = {}
                        for r in rows:
                            t_key = r["target_key"]
                            if not t_key:
                                continue
                            rel_type = str(r["relation_type"])
                            score = float(r["score"])
                            rank = (RELATION_TYPE_PRECEDENCE.get(rel_type, 0), score)

                            if t_key not in best_by_target:
                                best_by_target[t_key] = {
                                    "target_key": t_key,
                                    "score": score,
                                    "relation_type": rel_type,
                                }
                            else:
                                prev = best_by_target[t_key]
                                prev_rank = (
                                    RELATION_TYPE_PRECEDENCE.get(str(prev["relation_type"]), 0),
                                    float(prev["score"]),
                                )
                                if rank > prev_rank:
                                    best_by_target[t_key] = {
                                        "target_key": t_key,
                                        "score": score,
                                        "relation_type": rel_type,
                                    }

                        # Order by score DESC, target_key ASC with ceiling
                        sorted_targets = sorted(
                            best_by_target.values(),
                            key=lambda x: (-float(x["score"]), str(x["target_key"])),
                        )[:ceiling]

                        target_keys = [t["target_key"] for t in sorted_targets]
                        meta_by_key: dict[str, dict[str, Any]] = {}
                        if target_keys:
                            placeholders = ",".join("?" for _ in target_keys)
                            try:
                                meta_rows = conn.execute(
                                    f"SELECT item_key, title, parent_item_key FROM conversion_meta WHERE item_key IN ({placeholders})",
                                    tuple(target_keys),
                                ).fetchall()
                                for mr in meta_rows:
                                    meta_by_key[mr["item_key"]] = dict(mr)
                            except sqlite3.OperationalError:
                                pass

                        for t in sorted_targets:
                            t_key = t["target_key"]
                            mr = meta_by_key.get(t_key)
                            raw_title = mr.get("title") if mr else None
                            title = raw_title.strip() if raw_title and str(raw_title).strip() else t_key
                            raw_pk = mr.get("parent_item_key") if mr else None
                            parent_key = raw_pk.strip() if raw_pk and str(raw_pk).strip() else None

                            related_items.append(
                                {
                                    "item_key": t_key,
                                    "title": title,
                                    "relation_type": t["relation_type"],
                                    "score": round(float(t["score"]), 4),
                                    "parent_item_key": parent_key,
                                }
                            )
                except Exception as exc:
                    logger.error("Failed to query relations for %s: %s", resolved_key, exc)
                    return {"status": "error", "message": f"Database error querying relations: {exc}"}

        res: dict[str, Any] = {
            "status": "success",
            "item_key": resolved_key,
            "total_count": len(related_items),
            "related_items": related_items,
        }

        if res_notice:
            res["notice"] = res_notice

        return res
    except Exception as exc:
        logger.error("Failed in get_related_items_core for %s: %s", item_key, exc)
        return {"status": "error", "message": str(exc)}

normalize_asset_ids 

normalize_asset_ids(asset_id)

Normalize raw asset_id input from LLM into a canonical list of asset ID strings, or None.

Handles
  • None -> None (full document)
  • JSON array string: '["s1", "s2"]' -> ["s1", "s2"]
  • Bracketed list string: '[s3, s4, s5]' -> ["s3", "s4", "s5"]
  • Single bracketed string: '[s1]' -> ["s1"]
  • Comma-separated string: 's1, s2, s3' -> ["s1", "s2", "s3"]
  • Quoted string: '"s1"' -> ["s1"]
  • Python list / tuple / set of strings: ["s1", "s2"] -> ["s1", "s2"]
  • Preserves order and deduplicates identical IDs.
Source code in zocrates/agents/common/toolkit.py 
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
def normalize_asset_ids(
    asset_id: str | list[str] | None,
) -> list[str] | None:
    """Normalize raw asset_id input from LLM into a canonical list of asset ID strings, or None.

    Handles:
      - None -> None (full document)
      - JSON array string: '["s1", "s2"]' -> ["s1", "s2"]
      - Bracketed list string: '[s3, s4, s5]' -> ["s3", "s4", "s5"]
      - Single bracketed string: '[s1]' -> ["s1"]
      - Comma-separated string: 's1, s2, s3' -> ["s1", "s2", "s3"]
      - Quoted string: '"s1"' -> ["s1"]
      - Python list / tuple / set of strings: ["s1", "s2"] -> ["s1", "s2"]
      - Preserves order and deduplicates identical IDs.
    """
    if asset_id is None:
        return None

    def _strip_wrapping(s: Any) -> str:
        """Strip enclosing brackets, braces, and quotes from a string token.

        Args:
            s: Value or token to strip wrapping delimiters from.

        Returns:
            Unwrapped string.
        """
        t = str(s).strip()
        changed = True
        while changed and len(t) >= 2:
            changed = False
            for open_b, close_b in [("[", "]"), ("{", "}"), ("(", ")"), ('"', '"'), ("'", "'")]:
                if t.startswith(open_b) and t.endswith(close_b):
                    t = t[len(open_b) : -len(close_b)].strip()
                    changed = True
                    break
        return t

    tokens: list[str] = []

    if isinstance(asset_id, str):
        raw_str = asset_id.strip()
        if not raw_str:
            return None

        # 1. Try parsing as JSON array
        if raw_str.startswith("[") and raw_str.endswith("]"):
            try:
                parsed = json.loads(raw_str)
                if isinstance(parsed, list):
                    tokens = [str(item) for item in parsed if item is not None]
            except Exception:
                pass

        # 2. If not parsed as JSON, strip outer brackets/braces/parens (not quotes spanning commas)
        if not tokens:
            unwrapped = raw_str
            while len(unwrapped) >= 2 and (
                (unwrapped[0] == "[" and unwrapped[-1] == "]")
                or (unwrapped[0] == "{" and unwrapped[-1] == "}")
                or (unwrapped[0] == "(" and unwrapped[-1] == ")")
            ):
                unwrapped = unwrapped[1:-1].strip()

            if "," in unwrapped:
                tokens = [part for part in unwrapped.split(",") if part.strip()]
            elif unwrapped:
                tokens = [unwrapped]

    elif isinstance(asset_id, (list, tuple, set)):
        tokens = [str(item) for item in asset_id if item is not None]
    else:
        tokens = [str(asset_id)]

    # Clean individual tokens and deduplicate preserving order
    result: list[str] = []
    seen: set[str] = set()
    for tok in tokens:
        c = _strip_wrapping(tok)
        if c and c not in seen:
            seen.add(c)
            result.append(c)

    return result if result else None

read_document 

read_document(item_key, asset_id=None, page=None, tool_context=None)

Read full document Markdown or targeted sections, figures, or tables from the OKF bundle.

Accepts exact asset IDs from an item's TOC (e.g., 's1', 's2', 'f1', 't1'), 'abstract' (or 'summary') to read the document's abstract, a list of IDs, or fuzzy heading names.

Parameters:

Name Type Description Default
item_key str

Unique Zotero item key.

 required
asset_id str | list[str] | None

Asset ID from TOC, list of asset IDs, or fuzzy heading string. If None (default), reads whole document paginated.

 None
page int | None

1-indexed page number if content exceeds ~3,000 words.

 None

Returns:

Type Description
dict[str, Any]

Dictionary with 'status', 'item_key', and list of 'assets'.
Source code in zocrates/agents/common/tools.py 
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
def read_document(
    item_key: str,
    asset_id: str | list[str] | None = None,
    page: int | None = None,
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Read full document Markdown or targeted sections, figures, or tables from the OKF bundle.

    Accepts exact asset IDs from an item's TOC (e.g., 's1', 's2', 'f1', 't1'), 'abstract' (or 'summary') to read the document's abstract, a list of IDs, or fuzzy heading names.

    Args:
        item_key: Unique Zotero item key.
        asset_id: Asset ID from TOC, list of asset IDs, or fuzzy heading string.
            If None (default), reads whole document paginated.
        page: 1-indexed page number if content exceeds ~3,000 words.

    Returns:
        Dictionary with 'status', 'item_key', and list of 'assets'.
    """
    allowed, err_msg, notice = _check_and_increment_tool_budget(tool_context)
    if not allowed:
        return {"status": "budget_exhausted", "error": err_msg}

    norm_asset_ids = normalize_asset_ids(asset_id)

    # Intra-tool deduplication check using canonical keys only
    if tool_context is not None:
        seen_assets = set(tool_context.state.get(TEMP_SEEN_ASSETS, []))
        if norm_asset_ids is None:
            full_key = f"{item_key}:full"
            if full_key in seen_assets:
                res = {
                    "status": "success",
                    "item_key": item_key,
                    "assets": [],
                    "notice": f"[Full document {item_key} was already retrieved in this session.]",
                }
                if notice:
                    _ensure_preliminary_report(tool_context, force=True)
                    res["budget_notice"] = notice
                return res
        else:
            unseen = [aid for aid in norm_asset_ids if f"{item_key}:{aid}" not in seen_assets]
            if not unseen:
                res = {
                    "status": "success",
                    "item_key": item_key,
                    "assets": [],
                    "notice": f"[All requested assets for {item_key} ({', '.join(norm_asset_ids)}) were already retrieved in this session.]",
                }
                if notice:
                    _ensure_preliminary_report(tool_context, force=True)
                    res["budget_notice"] = notice
                return res
            norm_asset_ids = unseen

    call_aid: str | list[str] | None = (
        norm_asset_ids[0]
        if (norm_asset_ids is not None and len(norm_asset_ids) == 1)
        else norm_asset_ids
    )
    res = read_document_core(item_key=item_key, asset_id=call_aid, page=page)

    if res.get("status") == "success" and tool_context is not None:
        assets = res.get("assets", [])
        intersecting_ids = res.get("intersecting_chunk_ids", [])

        # Update temp:seen_assets with CANONICAL keys (and single requested alias if resolved)
        seen_assets = set(tool_context.state.get(TEMP_SEEN_ASSETS, []))
        if norm_asset_ids is None:
            seen_assets.add(f"{item_key}:full")
        elif len(norm_asset_ids) == 1 and assets:
            seen_assets.add(f"{item_key}:{norm_asset_ids[0]}")

        for asset in assets:
            resolved_aid = asset.get("asset_id", "full")
            seen_assets.add(f"{item_key}:{resolved_aid}")
        tool_context.state[TEMP_SEEN_ASSETS] = list(seen_assets)

        # Range-Aware Pre-Exclusion: Add intersecting chunk IDs to temp:seen_chunk_ids
        seen_chunk_ids = set(tool_context.state.get(TEMP_SEEN_CHUNK_IDS, []))
        seen_chunk_ids.update(intersecting_ids)
        tool_context.state[TEMP_SEEN_CHUNK_IDS] = list(seen_chunk_ids)

        # Convert read assets to synthetic evidence chunks in temp:retrieved_evidence (deduping by chunk_id)
        doc_title = res.get("title") or ""
        doc_authors = res.get("authors")
        doc_year = res.get("year")
        if tool_context is not None and (not doc_title or doc_authors is None or doc_year is None):
            active_items = tool_context.state.get(ACTIVE_CONTEXT_ITEMS, [])
            if isinstance(active_items, list):
                for item in active_items:
                    if isinstance(item, dict) and (item.get("item_key") or item.get("key")) == item_key:
                        if not doc_title:
                            doc_title = item.get("title", "")
                        if doc_authors is None:
                            doc_authors = item.get("authors") or item.get("creators")
                        if doc_year is None:
                            doc_year = item.get("year")
                        break

        # Canonical bundle key: read_document_core resolves scope aliases (e.g.
        # a parent item key whose PDF bundle lives under an attachment key) to
        # the key the bundle is actually stored under.  Evidence chunks must
        # carry the RESOLVED key — the /assets endpoint serves files from
        # cache_dir/<item_key>, so a requested-but-unresolved key would 404
        # the figure image in the sidebar.
        resolved_bundle_key = str(res.get("item_key") or item_key).strip() or item_key

        current_evidence = list(tool_context.state.get(TEMP_RETRIEVED_EVIDENCE, []))
        existing_evidence_ids = {
            e.get("chunk_id") for e in current_evidence if e.get("chunk_id")
        }
        for asset in assets:
            synthetic_chunk_id = f"{resolved_bundle_key}:{asset.get('asset_id', 'full')}"
            if synthetic_chunk_id in existing_evidence_ids:
                continue
            existing_evidence_ids.add(synthetic_chunk_id)
            chunk_dict: dict[str, Any] = {
                "chunk_id": synthetic_chunk_id,
                "item_key": resolved_bundle_key,
                "title": doc_title,
                "chunk_type": asset.get("asset_type", "text"),
                "heading": asset.get("heading", ""),
                "text": asset.get("content", ""),
                "word_count": asset.get("word_count", 0),
                "retrieval_method": "direct_read",
                "cosine_similarity": None,
                "rrf_score": None,
                "rerank_score": None,
                "image_path": asset.get("path") if asset.get("asset_type") == "figure" else None,
                "structured_content": asset.get("structured_content"),
            }
            if doc_authors is not None:
                chunk_dict["authors"] = doc_authors
            if doc_year is not None:
                chunk_dict["year"] = doc_year
            current_evidence.append(chunk_dict)
        tool_context.state[TEMP_RETRIEVED_EVIDENCE] = current_evidence

        total_words = sum(c.get("word_count", 0) for c in current_evidence)
        res["retrieval_status"] = (
            f"[Retrieval status: ~{total_words} words accumulated in evidence set; "
            f"excluded {len(intersecting_ids)} overlapping vector chunks from future searches]"
        )

    if notice and tool_context is not None:
        _ensure_preliminary_report(tool_context, force=True)
        res["budget_notice"] = notice

    return res

read_document_core 

read_document_core(item_key, asset_id=None, page=None)

Read full document Markdown or targeted sections, figures, tables, or abstract from the OKF bundle.

Accepts exact asset IDs from toc.md ('s0', 's1', 'f1', 't1'), 'abstract' (or 'summary') to read the document's abstract, a list of IDs, or fuzzy heading names. Supports Range-Aware Pre-Exclusion by returning all intersecting vector chunk IDs.

Parameters:

Name Type Description Default
item_key str

Unique Zotero item key.

 required
asset_id str | list[str] | None

Asset ID, list of asset IDs, or fuzzy heading string. If None, reads full document paginated.

 None
page int | None

1-indexed page number if content exceeds WORDS_PER_PAGE.

 None

Returns:

Type Description
dict[str, Any]

Dictionary with 'status', 'item_key', 'assets', and 'intersecting_chunk_ids'.
Source code in zocrates/agents/common/toolkit.py 
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
def read_document_core(
    item_key: str,
    asset_id: str | list[str] | None = None,
    page: int | None = None,
) -> dict[str, Any]:
    """Read full document Markdown or targeted sections, figures, tables, or abstract from the OKF bundle.

    Accepts exact asset IDs from toc.md ('s0', 's1', 'f1', 't1'), 'abstract' (or 'summary') to read the
    document's abstract, a list of IDs, or fuzzy heading names.
    Supports Range-Aware Pre-Exclusion by returning all intersecting vector chunk IDs.

    Args:
        item_key: Unique Zotero item key.
        asset_id: Asset ID, list of asset IDs, or fuzzy heading string. If None, reads full document paginated.
        page: 1-indexed page number if content exceeds WORDS_PER_PAGE.

    Returns:
        Dictionary with 'status', 'item_key', 'assets', and 'intersecting_chunk_ids'.
    """
    if not _validate_item_key(item_key):
        return {"status": "error", "message": f"Invalid item key: {item_key}"}

    try:
        resolved = _resolve_bundle_dir(item_key)
        if not resolved:
            return {
                "status": "error",
                "message": f"Document bundle for {item_key} not found or not yet converted.",
            }

        bundle_dir, resolved_key, read_notice = resolved
        doc_md_path = bundle_dir / "document.md"
        if not doc_md_path.exists():
            return {
                "status": "error",
                "message": f"Document markdown for {item_key} not found.",
            }

        syncer = get_syncer()

        # Ensure toc.json exists
        toc_json_path = bundle_dir / "toc.json"
        if not toc_json_path.exists():
            doc_text = doc_md_path.read_text(encoding="utf-8")
            meta = syncer.processor.get_metadata(resolved_key)
            title = meta.get("title") if meta else None
            parent_item_key = meta.get("parent_item_key") if meta else None
            from zocrates.rag.toc_generator import generate_toc, save_toc_files

            toc_json_data, toc_md = generate_toc(
                doc_text,
                title=title,
                item_key=resolved_key,
                parent_item_key=parent_item_key,
            )
            save_toc_files(bundle_dir, toc_json_data, toc_md)
        else:
            toc_json_data = json.loads(toc_json_path.read_text(encoding="utf-8"))

        doc_title = (
            toc_json_data.get("title")
            or (title if "title" in locals() and title else "")
            or ""
        )
        doc_authors = toc_json_data.get("authors")
        doc_year = toc_json_data.get("year")

        if (doc_authors is None or doc_year is None or not doc_title or doc_title.lower() in ("snapshot", "full text", "attachment")) and hasattr(syncer, "processor") and hasattr(syncer.processor, "get_metadata"):
            try:
                meta = syncer.processor.get_metadata(resolved_key)
                if isinstance(meta, dict):
                    if not doc_title and meta.get("title"):
                        doc_title = meta.get("title")
                    if doc_authors is None and meta.get("authors"):
                        doc_authors = meta.get("authors")
                    if doc_year is None and meta.get("year"):
                        doc_year = meta.get("year")
                    parent_key = meta.get("parent_item_key")
                    if parent_key and syncer.db:
                        parent_item = syncer.db.get_item_by_key(parent_key)
                        if parent_item:
                            if not doc_title or doc_title.lower() in ("snapshot", "full text", "attachment"):
                                doc_title = parent_item.title
                            if doc_authors is None and parent_item.creators:
                                doc_authors = parent_item.creators
                            if doc_year is None and parent_item.year:
                                doc_year = parent_item.year
            except Exception:
                pass

        if doc_authors is None or doc_year is None or not doc_title:
            try:
                doc_text_head = doc_md_path.read_text(encoding="utf-8")
                if doc_text_head.startswith("---"):
                    import yaml
                    parts = doc_text_head.split("---", 2)
                    if len(parts) >= 3:
                        fm = yaml.safe_load(parts[1])
                        if isinstance(fm, dict):
                            if (not doc_title or doc_title.lower() in ("snapshot", "full text", "attachment")) and fm.get("title"):
                                doc_title = str(fm["title"]).strip()
                            if doc_authors is None and fm.get("authors"):
                                doc_authors = fm["authors"]
                            if doc_year is None and fm.get("year"):
                                doc_year = str(fm["year"])
                            elif doc_year is None and fm.get("date"):
                                m = re.search(r"\b(19\d\d|20\d\d)\b", str(fm["date"]))
                                if m:
                                    doc_year = m.group(1)
            except Exception:
                pass

        doc_text = doc_md_path.read_text(encoding="utf-8")
        doc_lines = doc_text.splitlines(keepends=True)

        def slice_lines(start_line: int, end_line: int) -> str:
            """Extract a 1-based inclusive line range from document lines.

            Args:
                start_line: 1-based start line index.
                end_line: 1-based end line index.

            Returns:
                Joined line content for the specified range.
            """
            s_idx = max(0, start_line - 1)
            e_idx = max(0, end_line)
            return "".join(doc_lines[s_idx:e_idx])

        # Precompute hierarchical breadcrumb headings for section assets
        def _is_fm_or_title(h_text: str) -> bool:
            clean_h = h_text.strip().lower()
            if not clean_h or clean_h in ("frontmatter", "title", "document body"):
                return True
            if re.match(r"^(?:figure|table)\s+\d", clean_h, re.IGNORECASE):
                return True
            if doc_title:
                norm_h = re.sub(r"[^\w\s]", "", clean_h).replace(" ", "")
                norm_t = re.sub(r"[^\w\s]", "", doc_title.lower()).replace(" ", "")
                if norm_h and norm_h == norm_t:
                    return True
            return False

        sec_breadcrumb_headings: dict[int, str] = {}
        heading_stack: list[dict[str, Any]] = []

        for sec in toc_json_data.get("sections", []):
            s_id = sec.get("id", "")
            raw_h = sec.get("heading", "")
            lvl = int(sec.get("level") or 1)

            if sec.get("level") in (0, "0") or s_id == "s0" or _is_fm_or_title(raw_h):
                sec_breadcrumb_headings[id(sec)] = raw_h
                continue

            while heading_stack and heading_stack[-1]["level"] >= lvl:
                heading_stack.pop()

            if " > " in raw_h:
                computed_h = raw_h
                leaf_h = raw_h.split(" > ")[-1].strip()
            else:
                parents = [item["heading"] for item in heading_stack]
                if parents:
                    computed_h = f"{' > '.join(parents)} > {raw_h}"
                else:
                    computed_h = raw_h
                leaf_h = raw_h

            sec_breadcrumb_headings[id(sec)] = computed_h
            heading_stack.append({"level": lvl, "heading": leaf_h})

        def _get_sec_heading(s_dict: dict[str, Any]) -> str:
            return sec_breadcrumb_headings.get(id(s_dict), s_dict.get("heading", ""))

        # Case 1: Whole document paginated mode (asset_id is None)
        if asset_id is None:
            if not doc_lines:
                doc_lines = [""]

            # Partition lines into pages of up to WORDS_PER_PAGE words
            pages: list[tuple[int, int]] = []
            cur_start_idx = 0
            cur_words = 0
            for idx, line in enumerate(doc_lines):
                lw = len(line.split())
                if cur_words > 0 and cur_words + lw > WORDS_PER_PAGE:
                    pages.append((cur_start_idx + 1, idx))
                    cur_start_idx = idx
                    cur_words = lw
                else:
                    cur_words += lw

            pages.append((cur_start_idx + 1, len(doc_lines)))
            total_pages = len(pages)
            req_page = max(1, min(page or 1, total_pages))
            page_start_line, page_end_line = pages[req_page - 1]
            content = slice_lines(page_start_line, page_end_line)
            page_words = content.split()
            has_more = req_page < total_pages

            inter_chunk_ids = syncer.store.get_intersecting_chunk_ids(
                resolved_key, page_start_line, page_end_line
            )
            asset_obj = {
                "asset_id": "full",
                "asset_type": "document",
                "heading": "Full Document",
                "content": content,
                "word_count": len(page_words),
                "path": None,
                "structured_content": None,
                "page": req_page,
                "total_pages": total_pages,
                "has_more": has_more,
            }
            full_doc_result: dict[str, Any] = {
                "status": "success",
                "item_key": resolved_key,
                "title": doc_title,
                "assets": [asset_obj],
                "intersecting_chunk_ids": inter_chunk_ids,
            }
            if doc_authors is not None:
                full_doc_result["authors"] = doc_authors
            if doc_year is not None:
                full_doc_result["year"] = doc_year
            if read_notice:
                full_doc_result["notice"] = read_notice
            return full_doc_result

        # Helper to resolve a single asset ID or fuzzy heading
        def _resolve_single_asset(aid: str, req_page: int | None = None) -> tuple[dict[str, Any], list[str]]:
            """Resolve a single asset ID or section heading into an asset payload and intersecting chunks.

            Args:
                aid: Asset ID (e.g. 's1', 'f2', 't1') or section heading title.
                req_page: Optional 1-based page number for paginated section reading.

            Returns:
                A tuple containing the asset dictionary and a list of intersecting chunk IDs.

            Raises:
                KeyError: If the requested asset ID or heading is not found in the TOC manifest.
            """
            clean_aid = aid.strip()
            clean_lower = clean_aid.lower()

            # 0. Virtual Abstract / Summary match ("abstract", "summary")
            if clean_lower in ("abstract", "summary"):
                # Delegate to an explicit section if present
                for sec in toc_json_data.get("sections", []):
                    if sec.get("heading", "").lower().strip() in ("abstract", "summary"):
                        start_l = sec.get("start_line", 1)
                        end_l = sec.get("end_line", start_l)
                        raw_content = slice_lines(start_l, end_l)
                        inter_ids = syncer.store.get_intersecting_chunk_ids(resolved_key, start_l, end_l)
                        words = raw_content.split()
                        total_w = len(words)
                        tot_pages = max(1, (total_w + WORDS_PER_PAGE - 1) // WORDS_PER_PAGE)
                        p = max(1, min(req_page or 1, tot_pages))
                        if tot_pages > 1:
                            p_words = words[(p - 1) * WORDS_PER_PAGE : p * WORDS_PER_PAGE]
                            p_content = " ".join(p_words)
                            has_m = p < tot_pages
                        else:
                            p_content = raw_content
                            p_words = words
                            has_m = False

                        return {
                            "asset_id": sec["id"],
                            "asset_type": "section",
                            "heading": _get_sec_heading(sec),
                            "content": p_content,
                            "word_count": len(p_words),
                            "path": None,
                            "structured_content": None,
                            "page": p,
                            "total_pages": tot_pages,
                            "has_more": has_m,
                        }, inter_ids

                # Fallback to metadata abstract from toc.json
                abstract_text = (toc_json_data.get("abstract") or "").strip()
                abs_src = (toc_json_data.get("abstract_source") or "").strip().lower()
                if abstract_text and abstract_text != "No abstract or summary available." and abs_src != "none":
                    words = abstract_text.split()
                    total_w = len(words)
                    tot_pages = max(1, (total_w + WORDS_PER_PAGE - 1) // WORDS_PER_PAGE)
                    p = max(1, min(req_page or 1, tot_pages))
                    if tot_pages > 1:
                        p_words = words[(p - 1) * WORDS_PER_PAGE : p * WORDS_PER_PAGE]
                        p_content = " ".join(p_words)
                        has_m = p < tot_pages
                    else:
                        p_content = abstract_text
                        p_words = words
                        has_m = False

                    return {
                        "asset_id": "abstract",
                        "asset_type": "abstract",
                        "heading": "Abstract",
                        "content": p_content,
                        "word_count": len(p_words),
                        "path": None,
                        "structured_content": None,
                        "page": p,
                        "total_pages": tot_pages,
                        "has_more": has_m,
                    }, []

                raise KeyError(f"No abstract or summary available for document {resolved_key}.")

            # 1. Section match (s0, s1, ...)
            for sec in toc_json_data.get("sections", []):
                if sec.get("id") == clean_aid:
                    start_l = sec.get("start_line", 1)
                    end_l = sec.get("end_line", start_l)
                    raw_content = slice_lines(start_l, end_l)
                    inter_ids = syncer.store.get_intersecting_chunk_ids(resolved_key, start_l, end_l)
                    words = raw_content.split()
                    total_w = len(words)
                    tot_pages = max(1, (total_w + WORDS_PER_PAGE - 1) // WORDS_PER_PAGE)
                    p = max(1, min(req_page or 1, tot_pages))
                    if tot_pages > 1:
                        p_words = words[(p - 1) * WORDS_PER_PAGE : p * WORDS_PER_PAGE]
                        p_content = " ".join(p_words)
                        has_m = p < tot_pages
                    else:
                        p_content = raw_content
                        p_words = words
                        has_m = False

                    return {
                        "asset_id": sec["id"],
                        "asset_type": "section",
                        "heading": _get_sec_heading(sec),
                        "content": p_content,
                        "word_count": len(p_words),
                        "path": None,
                        "structured_content": None,
                        "page": p,
                        "total_pages": tot_pages,
                        "has_more": has_m,
                    }, inter_ids

            # 2. Figure match (f1, f2, ...)
            for fig in toc_json_data.get("figures", []):
                if fig.get("id") == clean_aid:
                    start_l = fig.get("start_line", 1)
                    end_l = fig.get("end_line", start_l)
                    raw_content = slice_lines(start_l, end_l)
                    inter_ids = syncer.store.get_intersecting_chunk_ids(resolved_key, start_l, end_l)
                    return {
                        "asset_id": fig["id"],
                        "asset_type": "figure",
                        "heading": fig.get("caption", "") or "",
                        "content": raw_content,
                        "word_count": len(raw_content.split()),
                        "path": fig.get("path"),
                        "structured_content": None,
                        "page": 1,
                        "total_pages": 1,
                        "has_more": False,
                    }, inter_ids

            # 3. Table match (t1, t2, ...)
            for tab in toc_json_data.get("tables", []):
                if tab.get("id") == clean_aid:
                    start_l = tab.get("start_line", 1)
                    end_l = tab.get("end_line", start_l)
                    raw_content = slice_lines(start_l, end_l)
                    inter_ids = syncer.store.get_intersecting_chunk_ids(resolved_key, start_l, end_l)
                    csv_content = None
                    if tab.get("path"):
                        csv_p = (bundle_dir / tab["path"]).resolve()
                        if csv_p.is_relative_to(bundle_dir.resolve()) and csv_p.exists():
                            csv_content = csv_p.read_text(encoding="utf-8")
                    return {
                        "asset_id": tab["id"],
                        "asset_type": "table",
                        "heading": tab.get("caption", "") or "",
                        "content": raw_content,
                        "word_count": len(raw_content.split()),
                        "path": tab.get("path"),
                        "structured_content": csv_content,
                        "page": 1,
                        "total_pages": 1,
                        "has_more": False,
                    }, inter_ids

            # 4. Fuzzy heading matching fallback
            norm_aid = re.sub(r"[^\w\s]", "", clean_aid).lower().strip()
            best_sec = None
            for sec in toc_json_data.get("sections", []):
                h = sec.get("heading", "")
                norm_h = re.sub(r"[^\w\s]", "", h).lower().strip()
                b_h = _get_sec_heading(sec)
                norm_b = re.sub(r"[^\w\s]", "", b_h).lower().strip()
                if (
                    norm_aid == norm_h
                    or (norm_aid and norm_aid in norm_h)
                    or norm_aid == norm_b
                    or (norm_aid and norm_aid in norm_b)
                ):
                    best_sec = sec
                    break

            if best_sec is not None:
                start_l = best_sec.get("start_line", 1)
                end_l = best_sec.get("end_line", start_l)
                raw_content = slice_lines(start_l, end_l)
                inter_ids = syncer.store.get_intersecting_chunk_ids(resolved_key, start_l, end_l)
                words = raw_content.split()
                total_w = len(words)
                tot_pages = max(1, (total_w + WORDS_PER_PAGE - 1) // WORDS_PER_PAGE)
                p = max(1, min(req_page or 1, tot_pages))
                if tot_pages > 1:
                    p_words = words[(p - 1) * WORDS_PER_PAGE : p * WORDS_PER_PAGE]
                    p_content = " ".join(p_words)
                    has_m = p < tot_pages
                else:
                    p_content = raw_content
                    p_words = words
                    has_m = False
                return {
                    "asset_id": best_sec["id"],
                    "asset_type": "section",
                    "heading": _get_sec_heading(best_sec),
                    "content": p_content,
                    "word_count": len(p_words),
                    "path": None,
                    "structured_content": None,
                    "page": p,
                    "total_pages": tot_pages,
                    "has_more": has_m,
                }, inter_ids

            raise KeyError(f"Asset ID or heading '{clean_aid}' not found in document bundle.")

        norm_asset_ids = normalize_asset_ids(asset_id)

        # Case 1: Whole document paginated mode (norm_asset_ids is None)
        # Note: if asset_id was None, it was already handled above.
        if norm_asset_ids is None:
            return {
                "status": "error",
                "message": f"Invalid asset_id: '{asset_id}'",
            }

        # Case 2: Single asset string
        if len(norm_asset_ids) == 1:
            aid = norm_asset_ids[0]
            try:
                single_asset, inter_ids = _resolve_single_asset(aid, req_page=page)
                single_result: dict[str, Any] = {
                    "status": "success",
                    "item_key": resolved_key,
                    "title": doc_title,
                    "assets": [single_asset],
                    "intersecting_chunk_ids": inter_ids,
                }
                if doc_authors is not None:
                    single_result["authors"] = doc_authors
                if doc_year is not None:
                    single_result["year"] = doc_year
                if read_notice:
                    single_result["notice"] = read_notice
                return single_result
            except KeyError as err:
                return {"status": "error", "message": str(err)}

        # Case 3: List of asset IDs
        assets: list[dict[str, Any]] = []
        all_inter: list[str] = []
        total_words_accum = 0
        unresolved_ids: list[str] = []

        for aid in norm_asset_ids:
            try:
                asset_obj, inter_ids = _resolve_single_asset(aid, req_page=1)
                all_inter.extend(inter_ids)
                if total_words_accum + asset_obj["word_count"] > MULTI_ASSET_WORD_CAP:
                    remaining = max(0, MULTI_ASSET_WORD_CAP - total_words_accum)
                    if remaining > 0:
                        w_slice = asset_obj["content"].split()[:remaining]
                        asset_obj["content"] = " ".join(w_slice)
                        asset_obj["word_count"] = len(w_slice)
                        asset_obj["has_more"] = True
                        assets.append(asset_obj)
                    break
                else:
                    total_words_accum += asset_obj["word_count"]
                    assets.append(asset_obj)
            except KeyError:
                unresolved_ids.append(aid)

        # Deduplicate intersecting IDs preserving order
        seen: set[str] = set()
        deduped_inter = [cid for cid in all_inter if not (cid in seen or seen.add(cid))]

        status = "success" if assets else "error"
        result_payload: dict[str, Any] = {
            "status": status,
            "item_key": resolved_key,
            "title": doc_title,
            "assets": assets,
            "intersecting_chunk_ids": deduped_inter,
        }
        if doc_authors is not None:
            result_payload["authors"] = doc_authors
        if doc_year is not None:
            result_payload["year"] = doc_year
        if unresolved_ids:
            result_payload["unresolved_asset_ids"] = unresolved_ids
            result_payload["notice"] = (
                f"[Warning: Asset IDs {unresolved_ids} could not be resolved in document {resolved_key}. "
                "Verify asset IDs against get_document_toc output.]"
            )
        elif read_notice:
            result_payload["notice"] = read_notice

        return result_payload
    except Exception as exc:
        logger.error("Failed in read_document_core for %s: %s", item_key, exc)
        return {"status": "error", "message": str(exc)}

request_search_extension 

request_search_extension(justification, covered_summary, missing_aspects, tool_context=None)

Request permission from the user to extend the retrieval budget beyond the current limit.

Use this tool when evidence volume or retrieval tool budget is approaching its ceiling and crucial literature aspects or candidate papers remain unaddressed.

Parameters:

Name Type Description Default
justification str

Clear reason why additional retrieval calls or evidence capacity are needed.

 required
covered_summary str

Brief summary of the findings and evidence collected so far.

 required
missing_aspects list[str]

List of specific unaddressed aspects, papers, or topics still needed.

 required

Returns:

Type Description
dict[str, Any]

Dictionary confirming extension request registration and turn conclusion.
Source code in zocrates/agents/common/tools.py 
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
def request_search_extension(
    justification: str,
    covered_summary: str,
    missing_aspects: list[str],
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Request permission from the user to extend the retrieval budget beyond the current limit.

    Use this tool when evidence volume or retrieval tool budget is approaching its ceiling
    and crucial literature aspects or candidate papers remain unaddressed.

    Args:
        justification: Clear reason why additional retrieval calls or evidence capacity are needed.
        covered_summary: Brief summary of the findings and evidence collected so far.
        missing_aspects: List of specific unaddressed aspects, papers, or topics still needed.

    Returns:
        Dictionary confirming extension request registration and turn conclusion.
    """
    if tool_context is not None:
        tool_context.state[TEMP_SEARCH_REPORT] = {
            "summary": covered_summary,
            "gaps": missing_aspects if isinstance(missing_aspects, list) else [str(missing_aspects)],
            "notes": justification,
            "source": "subagent_request",
        }
        tool_context.state[TEMP_EXTENSION_REQUESTED] = True
        ext_num = int(tool_context.state.get(TEMP_EXTENSION_COUNT, 0)) + 1
        tool_context.state[TEMP_EXTENSION_REQUEST_PASS] = ext_num
        tool_context.actions.skip_summarization = True

    return {
        "status": "extension_requested",
        "message": "Extension request registered. Concluding retrieval turn for user approval.",
    }

search_corpus 

search_corpus(query, item_key='active', chunk_type=None, limit=10, tool_context=None)

Hybrid (dense vector + FTS5 BM25) search over indexed library chunks with compound pre-filtering.

Parameters:

Name Type Description Default
query str

The search query string. If empty, returns immediately.

 required
item_key str | list[str] | None

Scope of the search: - "active" (default): searches active context items selected in Zotero (or entire library if none selected). - None, "all", or "*": searches across the ENTIRE library (unfiltered). - Specific item key(s) (e.g. "HCZLU22L" or ["HCZLU22L", "M5Z7SFWM"]): filters strictly to those items.

 'active'
chunk_type str | list[str] | None

Chunk type filter ('frontmatter', 'figure', 'table', 'equation', 'code', 'text'). Omit, None, or "all" to search across all chunk types.

 None
limit int

Maximum number of chunks to return (default 10).

 10

Returns:

Type Description
dict[str, Any]

Dictionary with 'status' and list of matching 'results'.
Source code in zocrates/agents/common/tools.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def search_corpus(
    query: str,
    item_key: str | list[str] | None = "active",
    chunk_type: str | list[str] | None = None,
    limit: int = 10,
    tool_context: ToolContext | None = None,
) -> dict[str, Any]:
    """Hybrid (dense vector + FTS5 BM25) search over indexed library chunks with compound pre-filtering.

    Args:
        query: The search query string. If empty, returns immediately.
        item_key: Scope of the search:
            - "active" (default): searches active context items selected in Zotero (or entire library if none selected).
            - None, "all", or "*": searches across the ENTIRE library (unfiltered).
            - Specific item key(s) (e.g. "HCZLU22L" or ["HCZLU22L", "M5Z7SFWM"]): filters strictly to those items.
        chunk_type: Chunk type filter ('frontmatter', 'figure', 'table', 'equation', 'code', 'text').
            Omit, None, or "all" to search across all chunk types.
        limit: Maximum number of chunks to return (default 10).

    Returns:
        Dictionary with 'status' and list of matching 'results'.
    """
    if not query or not query.strip():
        return {
            "status": "error",
            "message": "Query string cannot be empty.",
            "results": [],
        }

    allowed, err_msg, notice = _check_and_increment_tool_budget(tool_context)
    if not allowed:
        return {"status": "budget_exhausted", "error": err_msg}

    filter_keys = item_key
    seen_chunk_ids: list[str] | None = None

    if tool_context is not None:
        # Resolve active selection scope only when explicitly requested via "active", "selected", or "context"
        if isinstance(filter_keys, str) and filter_keys.strip().lower() in ("active", "selected", "context"):
            active_items = tool_context.state.get(ACTIVE_CONTEXT_ITEMS)
            if active_items and isinstance(active_items, list):
                extracted = [
                    item.get("item_key") or item.get("key")
                    for item in active_items
                    if isinstance(item, dict) and (item.get("item_key") or item.get("key"))
                ]
                filter_keys = extracted if extracted else None
            else:
                filter_keys = None

        raw_seen = tool_context.state.get(TEMP_SEEN_CHUNK_IDS, [])
        if isinstance(raw_seen, (list, set)):
            seen_chunk_ids = list(raw_seen)

    res = search_corpus_core(
        query=query,
        item_key=filter_keys,
        chunk_type=chunk_type,
        limit=limit,
        exclude_chunk_ids=seen_chunk_ids,
    )

    if res.get("status") == "success" and tool_context is not None:
        results = res.get("results", [])
        new_ids = [r["chunk_id"] for r in results if "chunk_id" in r]

        # Update temp:seen_chunk_ids
        current_seen = list(tool_context.state.get(TEMP_SEEN_CHUNK_IDS, []))
        seen_set = set(current_seen)
        seen_set.update(new_ids)
        tool_context.state[TEMP_SEEN_CHUNK_IDS] = list(seen_set)

        # Update temp:retrieved_evidence
        current_evidence = list(tool_context.state.get(TEMP_RETRIEVED_EVIDENCE, []))
        for r in results:
            chunk_entry: dict[str, Any] = {
                "chunk_id": r["chunk_id"],
                "item_key": r["item_key"],
                "title": r.get("title", ""),
                "chunk_type": r.get("chunk_type", "text"),
                "heading": r.get("heading", ""),
                "text": r.get("text", ""),
                "word_count": r.get("word_count", 0),
                "retrieval_method": "hybrid",
                "cosine_similarity": r.get("cosine_similarity"),
                "rrf_score": r.get("rrf_score"),
                "rerank_score": None,
                "image_path": r.get("image_path"),
                "structured_content": None,
            }
            if "authors" in r:
                chunk_entry["authors"] = r["authors"]
            if "year" in r:
                chunk_entry["year"] = r["year"]
            current_evidence.append(chunk_entry)
        tool_context.state[TEMP_RETRIEVED_EVIDENCE] = current_evidence

        total_words = sum(c.get("word_count", 0) for c in current_evidence)
        res["retrieval_status"] = (
            f"[Retrieval status: {len(new_ids)} new chunks retrieved, "
            f"~{total_words} words accumulated in evidence set]"
        )

    if notice and tool_context is not None:
        _ensure_preliminary_report(tool_context, force=True)
        res["budget_notice"] = notice

    return res

search_corpus_core 

search_corpus_core(query, item_key=None, chunk_type=None, limit=10, exclude_chunk_ids=None)

Hybrid (dense vector + FTS5 BM25) search over indexed library chunks with compound pre-filtering.

Parameters:

Name Type Description Default
query str

The search query string.

 required
item_key str | list[str] | None

Single item key or list of item keys to filter by. Defaults to searching all items.

 None
chunk_type str | list[str] | None

Chunk type filter ('frontmatter', 'figure', 'table', 'equation', 'code', 'text').

 None
limit int

Maximum number of results to return.

 10
exclude_chunk_ids set[str] | list[str] | None

Set or list of chunk IDs to exclude from search results.

 None

Returns:

Type Description
dict[str, Any]

Dictionary with 'status' and list of clean agent 'results'.
Source code in zocrates/agents/common/toolkit.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def search_corpus_core(
    query: str,
    item_key: str | list[str] | None = None,
    chunk_type: str | list[str] | None = None,
    limit: int = 10,
    exclude_chunk_ids: set[str] | list[str] | None = None,
) -> dict[str, Any]:
    """Hybrid (dense vector + FTS5 BM25) search over indexed library chunks with compound pre-filtering.

    Args:
        query: The search query string.
        item_key: Single item key or list of item keys to filter by. Defaults to searching all items.
        chunk_type: Chunk type filter ('frontmatter', 'figure', 'table', 'equation', 'code', 'text').
        limit: Maximum number of results to return.
        exclude_chunk_ids: Set or list of chunk IDs to exclude from search results.

    Returns:
        Dictionary with 'status' and list of clean agent 'results'.
    """
    if not query or not query.strip():
        return {"status": "success", "results": []}

    try:
        syncer = get_syncer()

        # Normalize item_key filter
        norm_item_keys: list[str] | None = None
        if isinstance(item_key, str):
            if item_key.strip().lower() in ("all", "*", "library", "entire_library"):
                norm_item_keys = None
            elif _validate_item_key(item_key):
                norm_item_keys = [item_key]
        elif isinstance(item_key, list):
            norm_item_keys = [
                k
                for k in item_key
                if isinstance(k, str)
                and k.strip().lower() not in ("all", "*", "library", "entire_library")
                and _validate_item_key(k)
            ]
            if not norm_item_keys:
                norm_item_keys = None

        # Normalize chunk_type filter
        norm_chunk_types: list[str] | None = None
        if isinstance(chunk_type, str):
            ct = chunk_type.strip().lower()
            if ct in ("none", "null", "all", "*", ""):
                norm_chunk_types = None
            else:
                norm_chunk_types = [ct]
        elif isinstance(chunk_type, list):
            filtered = [
                t.strip().lower()
                for t in chunk_type
                if isinstance(t, str) and t.strip().lower() not in ("none", "null", "all", "*", "")
            ]
            norm_chunk_types = filtered if filtered else None

        # Normalize exclude_chunk_ids
        norm_exclude: set[str] | None = None
        if exclude_chunk_ids:
            norm_exclude = set(exclude_chunk_ids)

        search_results = syncer.store.search(
            query=query,
            final_k=limit,
            item_key_filter=norm_item_keys,
            chunk_type_filter=norm_chunk_types,
            exclude_chunk_ids=norm_exclude,
        )

        cand_keys = {r.item_key for r in search_results if r.item_key}
        if syncer.db and cand_keys:
            active_keys = syncer.db.get_active_item_keys(list(cand_keys))
            if isinstance(active_keys, (set, list, tuple, frozenset)):
                search_results = [r for r in search_results if r.item_key in active_keys]

        # Cache item titles and metadata
        title_cache: dict[str, str] = {}
        authors_cache: dict[str, list[str]] = {}
        year_cache: dict[str, str] = {}
        serialized: list[dict[str, Any]] = []

        for r in search_results:
            if r.item_key not in title_cache:
                item = None
                if syncer.db and r.item_id:
                    cand = syncer.db.get_item(r.item_id)
                    cand_key = (
                        getattr(cand, "item_key", getattr(cand, "key", None))
                        if cand
                        else None
                    )
                    if cand and (
                        cand_key == r.item_key or not isinstance(cand_key, str)
                    ):
                        item = cand
                if not item and syncer.db and r.item_key:
                    item = syncer.db.get_item_by_key(r.item_key)

                parent_key = getattr(r, "parent_item_key", None)
                if not parent_key and syncer.processor:
                    meta = syncer.processor.get_metadata(r.item_key)
                    if isinstance(meta, dict):
                        parent_key = meta.get("parent_item_key")

                parent_item = None
                if parent_key and syncer.db:
                    parent_item = syncer.db.get_item_by_key(parent_key)

                raw_title = item.title if item else ""
                if (not raw_title or raw_title.lower() in ("snapshot", "full text", "attachment")) and parent_item:
                    raw_title = parent_item.title
                if not raw_title and syncer.processor:
                    meta = syncer.processor.get_metadata(r.item_key)
                    if isinstance(meta, dict) and meta.get("title"):
                        raw_title = meta["title"]
                title_cache[r.item_key] = raw_title or "Untitled Document"

                extracted_authors = getattr(item, "creators", []) or [] if item else []
                if not extracted_authors and parent_item:
                    extracted_authors = getattr(parent_item, "creators", []) or []
                if not isinstance(extracted_authors, (list, tuple)):
                    extracted_authors = []
                elif not extracted_authors and getattr(r, "creators", None):
                    extracted_authors = list(r.creators)

                # Fallback to document.md frontmatter for authors
                if not extracted_authors:
                    bundle_res = _resolve_bundle_dir(r.item_key)
                    if bundle_res:
                        doc_md = bundle_res[0] / "document.md"
                        if doc_md.exists():
                            try:
                                import yaml
                                lines = doc_md.read_text(encoding="utf-8").split("---", 2)
                                if len(lines) >= 3:
                                    fm = yaml.safe_load(lines[1])
                                    if isinstance(fm, dict) and fm.get("authors"):
                                        extracted_authors = fm["authors"]
                            except Exception:
                                pass
                authors_cache[r.item_key] = list(extracted_authors)

                extracted_year = getattr(item, "year", "") or "" if item else ""
                if not extracted_year and parent_item:
                    extracted_year = getattr(parent_item, "year", "") or ""
                if not isinstance(extracted_year, str):
                    extracted_year = ""
                elif not extracted_year and getattr(r, "year", None):
                    extracted_year = str(r.year)

                # Fallback to document.md frontmatter for year
                if not extracted_year:
                    bundle_res = _resolve_bundle_dir(r.item_key)
                    if bundle_res:
                        doc_md = bundle_res[0] / "document.md"
                        if doc_md.exists():
                            try:
                                import yaml
                                lines = doc_md.read_text(encoding="utf-8").split("---", 2)
                                if len(lines) >= 3:
                                    fm = yaml.safe_load(lines[1])
                                    if isinstance(fm, dict):
                                        if fm.get("year"):
                                            extracted_year = str(fm["year"])
                                        elif fm.get("date"):
                                            m = re.search(r"\b(19\d\d|20\d\d)\b", str(fm["date"]))
                                            if m:
                                                extracted_year = m.group(1)
                            except Exception:
                                pass
                year_cache[r.item_key] = str(extracted_year) if extracted_year else ""

            heading_str = " > ".join(r.heading_path) if r.heading_path else ""
            word_count = len(r.chunk_text.split()) if r.chunk_text else 0

            serialized.append(
                {
                    "chunk_id": r.chunk_id or f"{r.item_key}::{r.chunk_index}",
                    "item_key": r.item_key,
                    "title": title_cache[r.item_key],
                    "authors": authors_cache[r.item_key],
                    "year": year_cache[r.item_key],
                    "chunk_type": r.chunk_type,
                    "heading": heading_str,
                    "text": r.chunk_text,
                    "word_count": word_count,
                    "image_path": r.image_path,
                    "cosine_similarity": r.raw_semantic_score,
                    "rrf_score": r.score,
                    "line_start": r.line_start,
                    "line_end": r.line_end,
                }
            )

        return {"status": "success", "results": serialized}
    except Exception as exc:
        logger.error("Failed in search_corpus_core: %s", exc)
        return {"status": "error", "message": str(exc)}