Skip to content

Feat/eric/dev #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
EricNGOntos merged 5 commits into from
May 8, 2026
Merged

Feat/eric/dev #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
84d2266
feat: enhance graph deletion, sync performance, and zip manifest
EricNGOntos May 7, 2026
77b6122
feat(agentic): implement LLMPolicy and unify agentic RAG ranking logic
EricNGOntos May 8, 2026
4568334
refactor: implement recursive bottom-up doc_nav summary with virtual …
EricNGOntos May 8, 2026
7618520
fix: resolve ruff lint errors (unused imports and variables)
EricNGOntos May 8, 2026
2dd0f1c
test: update expected top_summary format in unit tests
EricNGOntos May 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
181 changes: 180 additions & 1 deletion apps/api/tests/contract/test_documents_contract.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,144 @@ async def _fetch_document(document_id: str) -> dict[str, object]:
await engine.dispose()


async def _fetch_graph_counts(
*,
document_id: str,
peer_document_id: str | None = None,
) -> dict[str, int]:
engine = await _create_contract_engine()
document_node_id = f"doc:{document_id}"
try:
async with engine.begin() as connection:
node_count = (
await connection.execute(
text("""
SELECT COUNT(*)
FROM graph_nodes
WHERE owner_document_id = :document_id
"""),
{"document_id": document_id},
)
).scalar_one()
related_edge_count = (
await connection.execute(
text("""
SELECT COUNT(*)
FROM graph_edges
WHERE owner_document_id = :document_id
OR source_node_id = :document_node_id
OR target_node_id = :document_node_id
"""),
{
"document_id": document_id,
"document_node_id": document_node_id,
},
)
).scalar_one()
peer_node_count = 0
if peer_document_id is not None:
peer_node_count = (
await connection.execute(
text("""
SELECT COUNT(*)
FROM graph_nodes
WHERE owner_document_id = :peer_document_id
"""),
{"peer_document_id": peer_document_id},
)
).scalar_one()
return {
"nodes": int(node_count),
"related_edges": int(related_edge_count),
"peer_nodes": int(peer_node_count),
}
finally:
await engine.dispose()


async def _insert_document_graph_fixture(
*,
document_id: str,
job_result_id: str,
peer_document_id: str,
peer_job_result_id: str,
user_id: str = "local-dev-user",
namespace: str = "contract-documents",
) -> None:
engine = await _create_contract_engine()
timestamp = datetime.now(timezone.utc).replace(tzinfo=None)
try:
async with engine.begin() as connection:
await connection.execute(
text("""
INSERT INTO graph_nodes (
node_id,
user_id,
namespace,
node_kind,
owner_document_id,
job_result_id,
ref_document_id,
ref_section_id,
properties,
created_at,
updated_at
) VALUES
(:doc_node_id, :user_id, :namespace, 'document', :document_id, :job_result_id, :document_id, NULL, CAST('{}' AS JSON), :created_at, :updated_at),
(:peer_node_id, :user_id, :namespace, 'document', :peer_document_id, :peer_job_result_id, :peer_document_id, NULL, CAST('{}' AS JSON), :created_at, :updated_at)
"""),
{
"doc_node_id": f"doc:{document_id}",
"peer_node_id": f"doc:{peer_document_id}",
"user_id": user_id,
"namespace": namespace,
"document_id": document_id,
"peer_document_id": peer_document_id,
"job_result_id": job_result_id,
"peer_job_result_id": peer_job_result_id,
"created_at": timestamp,
"updated_at": timestamp,
},
)
await connection.execute(
text("""
INSERT INTO graph_edges (
edge_id,
user_id,
namespace,
edge_kind,
source_node_id,
target_node_id,
owner_document_id,
job_result_id,
is_directed,
weight,
properties,
created_at,
updated_at
) VALUES
(:owned_edge_id, :user_id, :namespace, 'related', :doc_node_id, :peer_node_id, :document_id, :job_result_id, FALSE, 1.0, CAST('{}' AS JSON), :created_at, :updated_at),
(:incoming_edge_id, :user_id, :namespace, 'related', :peer_node_id, :doc_node_id, :peer_document_id, :peer_job_result_id, FALSE, 1.0, CAST('{}' AS JSON), :created_at, :updated_at)
"""),
{
"owned_edge_id": f"edge_{uuid4().hex[:12]}",
"incoming_edge_id": f"edge_{uuid4().hex[:12]}",
"user_id": user_id,
"namespace": namespace,
"doc_node_id": f"doc:{document_id}",
"peer_node_id": f"doc:{peer_document_id}",
"document_id": document_id,
"peer_document_id": peer_document_id,
"job_result_id": job_result_id,
"peer_job_result_id": peer_job_result_id,
"created_at": timestamp,
"updated_at": timestamp,
},
)
finally:
await engine.dispose()


async def _insert_document_revision_with_chunks(
*,
document_id: str,
Expand Down Expand Up @@ -662,21 +800,62 @@ async def test_should_archive_a_document_via_the_canonical_archive_route(
],
) -> None:
document_id = f"doc_{uuid4().hex[:12]}"
peer_document_id = f"doc_{uuid4().hex[:12]}"

async with developer_api_client_factory() as api_client:
await _insert_document(document_id=document_id)
document_revision = await _insert_document_revision_with_chunks(
document_id=document_id,
chunks=[
{
"id": f"dchk_{uuid4().hex[:12]}",
"chunk_id": "archive-chunk-1",
"chunk_type": "text",
"content": "Archived graph chunk",
"source_chunk_path": "Chapter 1/Archive",
"metadata": {"keywords": ["archive"]},
}
],
)
peer_revision = await _insert_document_revision_with_chunks(
document_id=peer_document_id,
chunks=[
{
"id": f"dchk_{uuid4().hex[:12]}",
"chunk_id": "peer-chunk-1",
"chunk_type": "text",
"content": "Peer graph chunk",
"source_chunk_path": "Chapter 1/Peer",
"metadata": {"keywords": ["peer"]},
}
],
)
await _insert_document_graph_fixture(
document_id=document_id,
job_result_id=document_revision["job_result_id"],
peer_document_id=peer_document_id,
peer_job_result_id=peer_revision["job_result_id"],
)
response = await api_client.post(f"/api/v1/documents/{document_id}/archive")

assert response.status_code == 200

response_json = cast(dict[str, object], response.json())
persisted_document = await _fetch_document(document_id)
graph_counts = await _fetch_graph_counts(
document_id=document_id,
peer_document_id=peer_document_id,
)

assert response_json["document_id"] == document_id
assert response_json["status"] == "archived"
assert response_json["archived_at"]
assert persisted_document["status"] == "archived"
assert persisted_document["archived_at"] is not None
assert graph_counts == {
"nodes": 0,
"related_edges": 0,
"peer_nodes": 1,
}


@pytest.mark.asyncio
Expand Down
46 changes: 2 additions & 44 deletions apps/worker/app/core/tasks/kb_tasks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
All I/O operations use sync services that yield cooperatively under gevent.
"""

import json
import os
from datetime import datetime, timezone

Expand Down Expand Up @@ -74,39 +73,6 @@
celery_app = get_celery_app()


def _extract_nav_sections_for_publish(add_dir: str) -> list:
"""Extract top-level nav sections from doc_nav.json for chunk metadata injection.

These sections will be read by graph_service.py when creating GraphNode,
enabling hierarchical navigation at query time.
"""
nav_path = os.path.join(add_dir, "doc_nav.json")
if not os.path.exists(nav_path):
return []
try:
from shared.utils.text_utils import truncate_content_preview
with open(nav_path, "r", encoding="utf-8") as f:
doc_nav = json.load(f)
sections = []
for section in doc_nav.get("sections", []):
title = section.get("title", "")
if title.lower() in ("root", "__root__"):
continue
sections.append({
"title": title,
"path": section.get("path", ""),
"summary": truncate_content_preview(
section.get("summary") or "", head=80, tail=0
),
"chunk_count": section.get("chunk_count", 0),
"children_count": len(section.get("children", [])),
})
return sections
except Exception as _e:
logger.warning(
f'doc_nav extraction failed (non-fatal): add_dir={add_dir!r}, error={_e}'
)
return []

@celery_app.task(
bind=True,
Expand Down Expand Up @@ -626,7 +592,6 @@ def _parse(job_id: str, user_id: str | None):
source_file_name = os.path.basename(source_file_name)

document_top_summary = ""
document_nav_sections = []
section_summaries: dict[str, str] = {}
if add_dir and source_file_name:
if add_contents_df is not None and "path" in add_contents_df.columns:
Expand All @@ -636,8 +601,6 @@ def _parse(job_id: str, user_id: str | None):
source_file_name=str(source_file_name),
)
# Enrich non-leaf section summaries (bottom-up aggregation)
# Must run before _extract_nav_sections_for_publish so that
# GraphNode.nav_sections gets populated summaries, not empty strings.
try:
kb_dir_for_enrich = os.path.dirname(str(add_dir))
summary_use_llm = JobMetadataHelper.get_parsing_param(
Expand All @@ -652,18 +615,13 @@ def _parse(job_id: str, user_id: str | None):
except Exception as _e:
logger.warning(f"doc_nav enrichment failed (non-fatal): {_e}")
document_top_summary = load_nav_top_summary(str(add_dir), str(source_file_name))
# Extract nav_sections from doc_nav.json for GraphNode persistence
document_nav_sections = _extract_nav_sections_for_publish(str(add_dir))
if document_top_summary or document_nav_sections:
if document_top_summary:
for chunk in chunks:
metadata = chunk.get("metadata")
if not isinstance(metadata, dict):
metadata = {}
chunk["metadata"] = metadata
if document_top_summary:
metadata["document_top_summary"] = document_top_summary
if document_nav_sections:
metadata["document_nav_sections"] = document_nav_sections
metadata["document_top_summary"] = document_top_summary

data_id = JobMetadataHelper.get_field(job_metadata, "data_id")

Expand Down
Loading
Loading