Skip to content
Merged
Show file tree
Hide file tree
Changes from 21 commits
Commits
Show all changes
33 commits
Select commit Hold shift + click to select a range
0545560
fix(ask): release pool before embedding provider work
seonghobae Aug 25, 2026
45b0a44
style: keep load evidence reviewable
seonghobae Aug 25, 2026
24d8fa1
Merge remote-tracking branch 'origin/main' into fix/global-ask-embedd…
seonghobae Aug 25, 2026
3d69ea4
docs(gaps): refresh protected delivery evidence
seonghobae Aug 25, 2026
8797605
fix(ask): reject blank embedding requests
seonghobae Aug 25, 2026
2e4fbc5
fix(ask): preserve unavailable embedding short circuit
seonghobae Aug 25, 2026
fee4d76
Merge remote-tracking branch 'origin/fix/global-ask-embedding-pool-re…
seonghobae Aug 25, 2026
99f6b81
fix(ask): honor validated precomputed embeddings
seonghobae Aug 25, 2026
23e3fde
fix(ask): honor precomputed embedding envelope
seonghobae Aug 25, 2026
09ce91b
Merge remote-tracking branch 'origin/fix/global-ask-embedding-pool-re…
seonghobae Aug 25, 2026
445571a
fix(k6): reject unitless request timeouts
seonghobae Aug 25, 2026
ca5d304
fix(migrations): replay global ask queue safely
seonghobae Aug 25, 2026
08e8705
Merge remote-tracking branch 'origin/fix/global-ask-embedding-pool-re…
seonghobae Aug 25, 2026
71c10dd
fix(ask): reject nonfinite embeddings
seonghobae Aug 25, 2026
4967528
Merge remote-tracking branch 'origin/fix/global-ask-embedding-pool-re…
seonghobae Aug 25, 2026
ccb4bb9
perf: keep authenticated web reads responsive (#633)
seonghobae Aug 25, 2026
143a6a3
fix(verification): persist completed provider results
seonghobae Aug 25, 2026
238a6cd
fix: count only claimed relation verifications
seonghobae Aug 25, 2026
883d1ff
fix: fence concurrent relation verification and migration drift
seonghobae Aug 25, 2026
ac38c65
fix: avoid evaluating absent migration indexes
seonghobae Aug 25, 2026
0f4665b
Merge branch 'main' into fix/global-ask-embedding-pool-release
opencode-agent[bot] Aug 25, 2026
74823e9
fix(ci): document trusted eligibility SQL
seonghobae Aug 25, 2026
4b4d670
test(ask): activate embedding diagnostics path
Aug 25, 2026
6d2fb7b
Merge remote-tracking branch 'origin/main' into repair/pr629-semgrep
Aug 25, 2026
0138db5
Merge remote-tracking branch 'origin/main' into repair/pr629-semgrep
Aug 25, 2026
c95f931
fix(verification): fence deleted evidence rows
Aug 25, 2026
967ba24
fix(lineage): preserve landing tie order
Aug 25, 2026
48496ff
fix(lineage): retain string tie-break contract
Aug 25, 2026
134e8f0
Merge remote-tracking branch 'origin/main' into restack/pr-629
seonghobae Aug 26, 2026
b2bc72c
fix(verification): restore incremental relation persistence
Aug 26, 2026
c00b571
test(db): execute Global Ask migration replay
Aug 26, 2026
b721b0f
fix(verification): fence deleted evidence
Aug 26, 2026
fcb933b
Merge origin/main into fix/global-ask-embedding-pool-release
Aug 27, 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
3 changes: 2 additions & 1 deletion Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -35,4 +35,5 @@ seed:
load-http:
@test -n "$${LINEAGEWEAVE_VUS:-}" || { echo "LINEAGEWEAVE_VUS is required" >&2; exit 1; }
@test -n "$${LINEAGEWEAVE_DURATION:-}" || { echo "LINEAGEWEAVE_DURATION is required" >&2; exit 1; }
k6 run --vus "$${LINEAGEWEAVE_VUS}" --duration "$${LINEAGEWEAVE_DURATION}" scripts/k6_http_e2e.js
@test -n "$${LINEAGEWEAVE_REQUEST_TIMEOUT:-}" || { echo "LINEAGEWEAVE_REQUEST_TIMEOUT is required" >&2; exit 1; }
k6 run -e REQUEST_TIMEOUT="$${LINEAGEWEAVE_REQUEST_TIMEOUT}" --vus "$${LINEAGEWEAVE_VUS}" --duration "$${LINEAGEWEAVE_DURATION}" scripts/k6_http_e2e.js
2 changes: 2 additions & 0 deletions backend/app/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ class Settings:
valkey_url: str
searxng_base_url: str
tepp_transport_url: str
tepp_api_key: str
Comment thread
seonghobae marked this conversation as resolved.
Resolved
caldav_base_url: str
naruon_calendar_base_url: str
naruon_calendar_service_token: str
Expand Down Expand Up @@ -171,6 +172,7 @@ def load_settings() -> Settings:
valkey_url=os.environ.get("VALKEY_URL", "redis://localhost:16379/0"),
searxng_base_url=os.environ.get("SEARXNG_BASE_URL", ""),
tepp_transport_url=os.environ.get("TEPP_TRANSPORT_URL", ""),
tepp_api_key=os.environ.get("TEPP_API_KEY", ""),
caldav_base_url=os.environ.get("CALDAV_BASE_URL", "").strip(),
naruon_calendar_base_url=os.environ.get("NARUON_CALENDAR_BASE_URL", "").strip(),
naruon_calendar_service_token=os.environ.get(
Expand Down
28 changes: 18 additions & 10 deletions backend/app/global_ask_queue.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
_seoul_today,
cited_post_images,
gather_global_chat_sources,
prepare_global_question_embedding,
)

GLOBAL_ASK_STREAM_KEY = "global_ask_request_stream"
Expand Down Expand Up @@ -237,16 +238,23 @@ def can_see(row: asyncpg.Record) -> bool:

today = _seoul_today()
try:
async with pool.acquire() as conn:
sources = await gather_global_chat_sources(
conn,
can_see,
corporate_entity_ids,
process_unit_ids,
question=question_text,
today=today,
embedding_client=embedding_client,
)
question_embedding = await prepare_global_question_embedding(
question_text, embedding_client or NullEmbeddingClient()
)
if question_embedding is None:
sources = []
else:
async with pool.acquire() as conn:
sources = await gather_global_chat_sources(
conn,
can_see,
corporate_entity_ids,
process_unit_ids,
question=question_text,
question_embedding=question_embedding,
today=today,
embedding_client=embedding_client,
)
Comment thread
seonghobae marked this conversation as resolved.
Resolved
Comment thread
seonghobae marked this conversation as resolved.
Resolved
Comment thread
seonghobae marked this conversation as resolved.
Resolved
except Exception as exc:
log_internal_fault("global_ask", exc)
record_server_failure("global_ask", exc, outcome="internal_error")
Expand Down
50 changes: 46 additions & 4 deletions backend/app/lineage_ingestion.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
import math
import re
from collections import defaultdict
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from datetime import datetime
from typing import Any

Expand Down Expand Up @@ -525,6 +525,40 @@
return visible_all, edge_rows


async def _fetch_lineage_landing_rows(
conn: asyncpg.Connection,
corporate_entity_ids: Sequence[str],
process_unit_ids: Sequence[str],
limit: int,
):
"""Fetch only the authorized, bounded landing projection in PostgreSQL."""
posts = await conn.fetch(
"select post_id, post_title, voc_type_code, visibility_code, "
"corporate_entity_id, process_unit_id, thread_group_key, created_at "
"from source_post where "
f"{SOURCE_POST_ELIGIBILITY_SQL.format(alias='source_post')} and "
"(visibility_code = 'public' or (corporate_entity_id::text = any($1::text[]) "
"and (cardinality($2::text[]) = 0 or process_unit_id::text = any($2::text[])))) "
"order by created_at desc, post_id desc limit $3",
list(corporate_entity_ids),
Comment thread
seonghobae marked this conversation as resolved.
Resolved
list(process_unit_ids),
limit + 1,
)
Comment thread
seonghobae marked this conversation as resolved.
Resolved
Comment thread
github-advanced-security[bot] marked this conversation as resolved.
Fixed
Comment thread
github-advanced-security[bot] marked this conversation as resolved.
Fixed
visible = list(posts[:limit])
visible_ids = [str(row["post_id"]) for row in visible]
edge_rows = (
await conn.fetch(
"select parent_post_id, child_post_id, fused_score, interval_relation_code "
"from post_lineage_edge where parent_post_id = any($1::uuid[]) "
"and child_post_id = any($1::uuid[])",
visible_ids,
)
if visible_ids
else []
)
return visible, edge_rows, len(posts) > limit


def _undirected_neighbors(edge_rows) -> dict[str, set[str]]:
neighbors: dict[str, set[str]] = {}
for edge in edge_rows:
Expand Down Expand Up @@ -658,23 +692,31 @@
limit: int = _LINEAGE_GRAPH_NODE_LIMIT,
focus_post_id: str | None = None,
include_isolated: bool = False,
corporate_entity_ids: Sequence[str] | None = None,
process_unit_ids: Sequence[str] = (),
) -> dict[str, Any]:
"""ABAC-filtered graph bounded for the browser's initial viewport.

The persisted graph can contain tens of thousands of posts. The UI opens
individual posts for complete lineage, while this landing projection keeps
only the newest ``limit`` visible nodes and edges between them.
"""
visible_all, edge_rows = await _fetch_visible_lineage_rows(conn, can_see_post)
if focus_post_id is None and corporate_entity_ids is not None:
visible, edge_rows, truncated = await _fetch_lineage_landing_rows(
conn, corporate_entity_ids, process_unit_ids, limit
)
visible_all = visible
else:
visible_all, edge_rows = await _fetch_visible_lineage_rows(conn, can_see_post)

if focus_post_id is None:
if focus_post_id is None and corporate_entity_ids is None:
visible = sorted(
visible_all,
key=lambda row: (row["created_at"], str(row["post_id"])),
reverse=True,
)[:limit]
truncated = len(visible_all) > len(visible)
else:
elif focus_post_id is not None:
focus_id = str(focus_post_id)
neighbors = _undirected_neighbors(edge_rows)
allowed = {str(row["post_id"]) for row in visible_all}
Comment thread
seonghobae marked this conversation as resolved.
Resolved
Expand Down
55 changes: 30 additions & 25 deletions backend/app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@
require_summary_source_body,
)
from backend.app.ranking_ingestion import load_visible_ranking_posts
from backend.app.relation_verification_ingestion import verify_post_relations
from backend.app.relation_verification_ingestion import verify_post_relations_from_pool
from backend.app.report_ingestion import (
GROUPING_KINDS,
fetch_period_comparison,
Expand Down Expand Up @@ -1251,6 +1251,8 @@ async def read_lineage_graph(
lambda row: _can_see_post(account, row),
limit=limit,
focus_post_id=post_id,
corporate_entity_ids=account.corporate_entity_ids,
process_unit_ids=account.process_unit_ids,
)


Expand Down Expand Up @@ -2310,28 +2312,25 @@ async def verify_post_entity_relationships(
status.HTTP_503_SERVICE_UNAVAILABLE,
"Relation verification is unavailable: set SEARXNG_BASE_URL",
)
async with pool.acquire() as conn:
try:
verified = await verify_post_relations(
conn,
client,
post_id,
visible_corporate_entity_ids=account.corporate_entity_ids,
)
except (HttpClientError, OSError) as exc:
# verify_post_relations() deliberately raises on a failed search
# (a failed search is not "searched and found nothing" -- see
# its docstring); this is the one caller, so it is the right
# place to turn that into a clean 503 instead of a raw 500.
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
"Relation verification is unavailable: the search provider did not respond",
) from exc
except Exception as exc: # noqa: BLE001 - provider boundary is fail-closed.
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
"Relation verification is unavailable: the search provider did not respond",
) from exc
try:
verified = await verify_post_relations_from_pool(
pool,
client,
post_id,
visible_corporate_entity_ids=account.corporate_entity_ids,
)
except (HttpClientError, OSError) as exc:
# A failed search is not "searched and found nothing"; turn the
# provider failure into a clean 503 rather than persisting a miss.
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
"Relation verification is unavailable: the search provider did not respond",
) from exc
except Exception as exc: # noqa: BLE001 - provider boundary is fail-closed.
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
"Relation verification is unavailable: the search provider did not respond",
) from exc
await publish_activity_event(
valkey,
post_id,
Expand Down Expand Up @@ -3386,7 +3385,12 @@ async def derive_post_commitment(
# Friday" in a January post must resolve to that January, not to the
# Friday after the operator clicked Derive.
reference_date = post["created_at"].date().isoformat()
commitment = client.extract(post["post_title"], normalized_body, reference_date)
commitment = await asyncio.to_thread(
client.extract,
post["post_title"],
normalized_body,
reference_date,
)
Comment thread
seonghobae marked this conversation as resolved.
Resolved
except (HttpClientError, KeyError, OSError, TypeError, ValueError, RuntimeError) as exc:
raise HTTPException(
status.HTTP_503_SERVICE_UNAVAILABLE,
Expand Down Expand Up @@ -3600,7 +3604,8 @@ async def read_calendar(
settings = load_settings()
if window_start is None or window_end is None:
window_start, window_end = default_calendar_window(datetime.now(timezone.utc))
naruon = load_observed_calendar_events(
naruon = await asyncio.to_thread(
load_observed_calendar_events,
build_workspace_naruon_client(
settings.naruon_calendar_base_url,
settings.naruon_calendar_service_token,
Expand Down
58 changes: 47 additions & 11 deletions backend/app/post_chat_ingestion.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from __future__ import annotations

import asyncio
import math
from dataclasses import dataclass
from datetime import date, datetime
from typing import Any, Callable, Iterable
Expand Down Expand Up @@ -385,6 +386,38 @@ async def gather_chat_sources(
return sources


async def prepare_global_question_embedding(
question: str,
embedding_client: EmbeddingClient,
) -> tuple[list[float], str, float] | None:
"""Resolve one question embedding without holding a database connection."""
if not question.strip() or not embedding_client.available:
return None
try:
question_vector = await asyncio.to_thread(embedding_client.embed, question)
except (OSError, RuntimeError, ValueError):
return None
return _validated_question_embedding(
question_vector, embedding_client.resolved_model
)
Comment thread
seonghobae marked this conversation as resolved.
Resolved


def _validated_question_embedding(
question_vector: list[float], embedding_model_code: str | None
) -> tuple[list[float], str, float] | None:
"""Return a finite, non-zero embedding envelope or fail closed."""
if (
not question_vector
or not embedding_model_code
or any(not math.isfinite(value) for value in question_vector)
):
return None
question_norm = math.sqrt(sum(value * value for value in question_vector))
if not math.isfinite(question_norm) or question_norm == 0.0:
return None
return question_vector, embedding_model_code, question_norm
Comment thread
seonghobae marked this conversation as resolved.
Resolved


async def gather_global_chat_sources(
conn: asyncpg.Connection,
can_see_post: Callable[[asyncpg.Record], bool],
Expand All @@ -394,6 +427,7 @@ async def gather_global_chat_sources(
embedding_client: EmbeddingClient | None = None,
*,
question: str | None = None,
question_embedding: tuple[list[float], str, float] | None = None,
limit: int = 4,
today: date | None = None,
) -> list[ChatSourceDocument]:
Expand Down Expand Up @@ -427,20 +461,22 @@ async def gather_global_chat_sources(
resolved_time_range = resolve_korean_relative_time(
question or "", today=today or _seoul_today()
)
if not (question and question.strip() and embedding_client.available):
return []
try:
question_vector = await asyncio.to_thread(embedding_client.embed, question)
except (OSError, RuntimeError, ValueError):
return []
if not question_vector:
if not (question and question.strip()):
return []
embedding_model_code = embedding_client.resolved_model
if not embedding_model_code:
if question_embedding is None:
if not embedding_client.available:
return []
question_embedding = await prepare_global_question_embedding(
question, embedding_client
)
if question_embedding is None:
return []
question_norm = sum(value * value for value in question_vector) ** 0.5
if question_norm == 0.0:
validated_embedding = _validated_question_embedding(
question_embedding[0], question_embedding[1]
)
if validated_embedding is None:
return []
question_vector, embedding_model_code, question_norm = validated_embedding
Comment thread
seonghobae marked this conversation as resolved.
Resolved
# Safe SQL: the only interpolation is the repository-owned eligibility
# expression; all request and model values remain asyncpg parameters.
candidate_rows = await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli
Expand Down
Loading
Loading