692 lines
31 KiB
Python
692 lines
31 KiB
Python
"""Unified context pack and budgeter for Space PMFlow AI.
|
|
|
|
This module is intentionally domain-agnostic at selection time: it ranks context
|
|
sections by product runtime priority, explicit relevance and size budget rather
|
|
than by business-scene keyword lists.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from .pmflow_ai_context_evidence_service import build_context_evidence_packet
|
|
from .pmflow_ai_thread_context_service import slim_thread_context_pack
|
|
|
|
|
|
DEFAULT_CONTEXT_BUDGET_CHARS = 32000
|
|
|
|
|
|
class PMFlowAIContextPackBuilder:
|
|
"""Build the single context contract consumed by Agent OS stages."""
|
|
|
|
def __init__(self, *, budget_chars: int = DEFAULT_CONTEXT_BUDGET_CHARS) -> None:
|
|
self.budgeter = PMFlowAIContextBudgeter(max_chars=budget_chars)
|
|
|
|
def build(
|
|
self,
|
|
*,
|
|
user_question: str,
|
|
thread_context_pack: dict[str, Any] | None,
|
|
conversation_context: dict[str, Any] | None,
|
|
tenant_project_context: dict[str, Any] | None,
|
|
memory_context_pack: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
question = _clean_text(user_question, limit=1600)
|
|
base = {
|
|
"packet_version": "pmflow_ai_context_pack.v1",
|
|
"built_at": datetime.utcnow().isoformat(),
|
|
"current_user_message": {
|
|
"role": "user",
|
|
"content": question,
|
|
},
|
|
"thread_context": slim_thread_context_pack(thread_context_pack, recent_limit=12, relevant_limit=10),
|
|
"memory_context": _memory_context(memory_context_pack),
|
|
"topic_context": _topic_context(conversation_context=conversation_context),
|
|
"object_context": _object_context(conversation_context=conversation_context, tenant_project_context=tenant_project_context),
|
|
"workspace_context": _workspace_context(conversation_context=conversation_context),
|
|
"surface_context": _surface_context(tenant_project_context),
|
|
"runtime_context": _runtime_context(conversation_context=conversation_context, tenant_project_context=tenant_project_context),
|
|
"context_policy": {
|
|
"current_user_message_is_primary": True,
|
|
"same_thread_history_is_primary_for_multi_turn_recall": True,
|
|
"project_truth_must_not_replace_thread_transcript": True,
|
|
"budgeting_is_section_priority_based_not_business_keyword_based": True,
|
|
},
|
|
}
|
|
packed = self.budgeter.apply(base)
|
|
packed["retrieval_critic_packet"] = context_pack_retrieval_critic_packet(
|
|
packed,
|
|
user_question=question,
|
|
)
|
|
packed["context_evidence_packet"] = build_context_evidence_packet(
|
|
context_pack=packed,
|
|
user_question=question,
|
|
)
|
|
return packed
|
|
|
|
|
|
class PMFlowAIContextBudgeter:
|
|
"""Apply a deterministic context budget with section-level priorities."""
|
|
|
|
def __init__(self, *, max_chars: int = DEFAULT_CONTEXT_BUDGET_CHARS) -> None:
|
|
self.max_chars = max(8000, int(max_chars or DEFAULT_CONTEXT_BUDGET_CHARS))
|
|
|
|
def apply(self, pack: dict[str, Any]) -> dict[str, Any]:
|
|
out = {
|
|
"packet_version": pack.get("packet_version") or "pmflow_ai_context_pack.v1",
|
|
"built_at": pack.get("built_at"),
|
|
"current_user_message": pack.get("current_user_message") if isinstance(pack.get("current_user_message"), dict) else {},
|
|
"context_policy": pack.get("context_policy") if isinstance(pack.get("context_policy"), dict) else {},
|
|
}
|
|
used = _estimated_size(out)
|
|
budget_rows: list[dict[str, Any]] = []
|
|
for section_name, priority, value, shrinker in _section_specs(pack):
|
|
slim_value = shrinker(value, self._remaining(used))
|
|
if slim_value in (None, {}, [], ""):
|
|
budget_rows.append(_budget_row(section_name, priority, "dropped_empty", value, slim_value))
|
|
continue
|
|
size = _estimated_size(slim_value)
|
|
if used + size <= self.max_chars:
|
|
out[section_name] = slim_value
|
|
used += size
|
|
budget_rows.append(_budget_row(section_name, priority, "included", value, slim_value))
|
|
continue
|
|
smaller = shrinker(value, max(1200, self._remaining(used)))
|
|
smaller_size = _estimated_size(smaller)
|
|
if smaller not in (None, {}, [], "") and used + smaller_size <= self.max_chars:
|
|
out[section_name] = smaller
|
|
used += smaller_size
|
|
budget_rows.append(_budget_row(section_name, priority, "compressed", value, smaller))
|
|
else:
|
|
budget_rows.append(_budget_row(section_name, priority, "dropped_budget", value, smaller))
|
|
out["budget_report"] = {
|
|
"packet_version": "pmflow_ai_context_budget_report.v1",
|
|
"max_chars": self.max_chars,
|
|
"used_chars": _estimated_size(out),
|
|
"sections": budget_rows,
|
|
}
|
|
return out
|
|
|
|
def _remaining(self, used: int) -> int:
|
|
return max(0, self.max_chars - int(used or 0))
|
|
|
|
|
|
def slim_pmflow_context_pack(value: Any, *, limit_chars: int = 18000) -> dict[str, Any]:
|
|
"""Return a prompt-safe projection of the unified context pack."""
|
|
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
budgeter = PMFlowAIContextBudgeter(max_chars=max(8000, int(limit_chars or 18000)))
|
|
return budgeter.apply(value)
|
|
|
|
|
|
def context_pack_thread_claim_rows(value: Any, *, limit: int = 8) -> list[dict[str, str]]:
|
|
if not isinstance(value, dict):
|
|
return []
|
|
thread_context = value.get("thread_context") if isinstance(value.get("thread_context"), dict) else {}
|
|
rows: list[dict[str, str]] = []
|
|
seen: set[str] = set()
|
|
for item in [
|
|
*list(thread_context.get("relevant_messages") or []),
|
|
*list(thread_context.get("recent_messages") or [])[-4:],
|
|
*list((thread_context.get("rolling_summary") or {}).get("durable_points") or [])[-6:],
|
|
]:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
content = str(item.get("content") or item.get("summary") or "").strip()
|
|
if not content:
|
|
continue
|
|
key = f"{item.get('message_id') or item.get('source_message_indexes')}:{content[:80]}"
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
rows.append(
|
|
{
|
|
"message_id": str(item.get("message_id") or "")[:80],
|
|
"role": str(item.get("role") or item.get("kind") or "thread_context")[:40],
|
|
"message_index": str(item.get("message_index") or "")[:20],
|
|
"content": content[:900],
|
|
}
|
|
)
|
|
if len(rows) >= limit:
|
|
break
|
|
return rows
|
|
|
|
|
|
def context_pack_memory_claim_rows(value: Any, *, limit: int = 8) -> list[dict[str, str]]:
|
|
if not isinstance(value, dict):
|
|
return []
|
|
memory_context = value.get("memory_context") if isinstance(value.get("memory_context"), dict) else {}
|
|
evidence = value.get("context_evidence_packet") if isinstance(value.get("context_evidence_packet"), dict) else {}
|
|
rows: list[dict[str, str]] = []
|
|
seen: set[str] = set()
|
|
for item in list(memory_context.get("memories") or []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
content = str(item.get("summary") or "").strip()
|
|
if not content:
|
|
continue
|
|
key = f"{item.get('memory_scope')}:{item.get('scope_ref')}:{content[:80]}"
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
rows.append(
|
|
{
|
|
"memory_id": str(item.get("memory_id") or "")[:80],
|
|
"memory_scope": str(item.get("memory_scope") or "")[:40],
|
|
"scope_ref": str(item.get("scope_ref") or "")[:160],
|
|
"content": content[:1000],
|
|
}
|
|
)
|
|
if len(rows) >= limit:
|
|
break
|
|
return rows
|
|
|
|
|
|
def context_pack_observability_packet(value: Any) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
sections = list((value.get("budget_report") or {}).get("sections") or []) if isinstance(value.get("budget_report"), dict) else []
|
|
thread_context = value.get("thread_context") if isinstance(value.get("thread_context"), dict) else {}
|
|
memory_context = value.get("memory_context") if isinstance(value.get("memory_context"), dict) else {}
|
|
evidence = value.get("context_evidence_packet") if isinstance(value.get("context_evidence_packet"), dict) else {}
|
|
return {
|
|
"packet_version": "pmflow_ai_context_observability.v1",
|
|
"context_packet_version": str(value.get("packet_version") or ""),
|
|
"budget": {
|
|
"max_chars": (value.get("budget_report") or {}).get("max_chars") if isinstance(value.get("budget_report"), dict) else None,
|
|
"used_chars": (value.get("budget_report") or {}).get("used_chars") if isinstance(value.get("budget_report"), dict) else None,
|
|
"sections": [
|
|
{
|
|
"section": str(item.get("section") or ""),
|
|
"status": str(item.get("status") or ""),
|
|
"priority": item.get("priority"),
|
|
"included_chars": item.get("included_chars"),
|
|
}
|
|
for item in sections
|
|
if isinstance(item, dict)
|
|
],
|
|
},
|
|
"thread": {
|
|
"recent_count": len(list(thread_context.get("recent_messages") or [])),
|
|
"relevant_count": len(list(thread_context.get("relevant_messages") or [])),
|
|
"has_rolling_summary": isinstance(thread_context.get("rolling_summary"), dict),
|
|
},
|
|
"memory": {
|
|
"memory_count": len(list(memory_context.get("memories") or [])),
|
|
"scopes": sorted(
|
|
{
|
|
str(item.get("memory_scope") or "")
|
|
for item in list(memory_context.get("memories") or [])
|
|
if isinstance(item, dict) and str(item.get("memory_scope") or "").strip()
|
|
}
|
|
),
|
|
},
|
|
"context_evidence": {
|
|
"claim_count": len(list(evidence.get("claims") or [])),
|
|
"obligation_count": len(list(evidence.get("continuation_obligations") or [])),
|
|
"topic_signature": str((evidence.get("topic_workspace") or {}).get("topic_signature") or "")[:80]
|
|
if isinstance(evidence.get("topic_workspace"), dict)
|
|
else "",
|
|
},
|
|
"dropped_sections": [
|
|
str(item.get("section") or "")
|
|
for item in sections
|
|
if isinstance(item, dict) and str(item.get("status") or "").startswith("dropped")
|
|
],
|
|
"retrieval_critic": value.get("retrieval_critic_packet") if isinstance(value.get("retrieval_critic_packet"), dict) else {},
|
|
}
|
|
|
|
|
|
def context_pack_used_context_packet(value: Any) -> dict[str, Any]:
|
|
"""Customer-safe grounding packet for the context supplied to answer runtime."""
|
|
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
thread_context = value.get("thread_context") if isinstance(value.get("thread_context"), dict) else {}
|
|
memory_context = value.get("memory_context") if isinstance(value.get("memory_context"), dict) else {}
|
|
evidence = value.get("context_evidence_packet") if isinstance(value.get("context_evidence_packet"), dict) else {}
|
|
critic = value.get("retrieval_critic_packet") if isinstance(value.get("retrieval_critic_packet"), dict) else {}
|
|
sections = list((value.get("budget_report") or {}).get("sections") or []) if isinstance(value.get("budget_report"), dict) else []
|
|
thread_rows: list[dict[str, Any]] = []
|
|
seen_messages: set[str] = set()
|
|
for item in [*list(thread_context.get("relevant_messages") or []), *list(thread_context.get("recent_messages") or [])]:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
message_id = str(item.get("message_id") or "")[:80]
|
|
dedupe = message_id or f"{item.get('role')}:{item.get('message_index')}:{str(item.get('content') or '')[:80]}"
|
|
if dedupe in seen_messages:
|
|
continue
|
|
seen_messages.add(dedupe)
|
|
thread_rows.append(
|
|
{
|
|
"message_id": message_id,
|
|
"role": str(item.get("role") or "")[:32],
|
|
"message_index": item.get("message_index"),
|
|
"content_preview": _clean_text(item.get("content") or item.get("summary") or "", limit=220),
|
|
}
|
|
)
|
|
if len(thread_rows) >= 12:
|
|
break
|
|
memory_rows: list[dict[str, Any]] = []
|
|
seen_memories: set[str] = set()
|
|
for item in list(memory_context.get("memories") or []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
memory_id = str(item.get("memory_id") or "")[:80]
|
|
dedupe = memory_id or f"{item.get('memory_scope')}:{item.get('scope_ref')}:{str(item.get('summary') or '')[:80]}"
|
|
if dedupe in seen_memories:
|
|
continue
|
|
seen_memories.add(dedupe)
|
|
memory_rows.append(
|
|
{
|
|
"memory_id": memory_id,
|
|
"memory_scope": str(item.get("memory_scope") or "")[:40],
|
|
"confidence_score": item.get("confidence_score"),
|
|
"summary_preview": _clean_text(item.get("summary") or "", limit=260),
|
|
}
|
|
)
|
|
if len(memory_rows) >= 12:
|
|
break
|
|
dropped_risks = []
|
|
for item in sections:
|
|
if not isinstance(item, dict) or not str(item.get("status") or "").startswith("dropped"):
|
|
continue
|
|
dropped_risks.append(
|
|
{
|
|
"kind": "context_section_dropped",
|
|
"section": str(item.get("section") or "")[:80],
|
|
"status": str(item.get("status") or "")[:40],
|
|
}
|
|
)
|
|
for risk in list(critic.get("risks") or []):
|
|
if not isinstance(risk, dict):
|
|
continue
|
|
dropped_risks.append(
|
|
{
|
|
"kind": str(risk.get("kind") or "")[:120],
|
|
"severity": str(risk.get("severity") or "")[:40],
|
|
}
|
|
)
|
|
return {
|
|
"packet_version": "pmflow_ai_used_context_packet.v1",
|
|
"usage_status": "provided_to_answer_runtime",
|
|
"usage_policy": {
|
|
"packet_is_grounding_observability_not_a_claim_that_every_item_was_quoted": True,
|
|
"customer_safe_previews_only": True,
|
|
},
|
|
"thread_message_refs": thread_rows,
|
|
"memory_refs": memory_rows,
|
|
"dropped_risks": dropped_risks[:16],
|
|
"coverage": {
|
|
"thread_message_count": len(thread_rows),
|
|
"memory_count": len(memory_rows),
|
|
"context_evidence_claim_count": len(list(evidence.get("claims") or [])),
|
|
"context_obligation_count": len(list(evidence.get("continuation_obligations") or [])),
|
|
"dropped_risk_count": len(dropped_risks),
|
|
},
|
|
"context_evidence": {
|
|
"claims": [
|
|
{
|
|
"claim_id": str(item.get("claim_id") or "")[:80],
|
|
"semantic_role": str(item.get("semantic_role") or "")[:80],
|
|
"source_kind": str(item.get("source_kind") or "")[:80],
|
|
"source_ref": str(item.get("source_ref") or "")[:160],
|
|
"claim_preview": _clean_text(item.get("claim_text"), limit=260),
|
|
"required_for_answer": bool(item.get("required_for_answer")),
|
|
}
|
|
for item in list(evidence.get("claims") or [])[:12]
|
|
if isinstance(item, dict)
|
|
],
|
|
"continuation_obligations": [
|
|
{
|
|
"obligation_id": str(item.get("obligation_id") or "")[:80],
|
|
"claim_id": str(item.get("claim_id") or "")[:80],
|
|
"semantic_role": str(item.get("semantic_role") or "")[:80],
|
|
}
|
|
for item in list(evidence.get("continuation_obligations") or [])[:12]
|
|
if isinstance(item, dict)
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def context_pack_retrieval_critic_packet(value: Any, *, user_question: str = "") -> dict[str, Any]:
|
|
"""Assess context retrieval quality without business-scene routing."""
|
|
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
question_terms = _text_terms(user_question or (value.get("current_user_message") or {}).get("content") if isinstance(value.get("current_user_message"), dict) else "")
|
|
thread_context = value.get("thread_context") if isinstance(value.get("thread_context"), dict) else {}
|
|
memory_context = value.get("memory_context") if isinstance(value.get("memory_context"), dict) else {}
|
|
sections = list((value.get("budget_report") or {}).get("sections") or []) if isinstance(value.get("budget_report"), dict) else []
|
|
dropped = {
|
|
str(item.get("section") or "")
|
|
for item in sections
|
|
if isinstance(item, dict) and str(item.get("status") or "").startswith("dropped")
|
|
}
|
|
memory_scores = []
|
|
for item in list(memory_context.get("memories") or []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
text = " ".join(
|
|
[
|
|
str(item.get("summary") or ""),
|
|
str(item.get("durable_facts") or ""),
|
|
str(item.get("open_loops") or ""),
|
|
str(item.get("decisions") or ""),
|
|
str(item.get("preferences") or ""),
|
|
]
|
|
)
|
|
terms = _text_terms(text)
|
|
overlap = len(question_terms & terms)
|
|
memory_scores.append(
|
|
{
|
|
"memory_scope": str(item.get("memory_scope") or "")[:40],
|
|
"scope_ref": str(item.get("scope_ref") or "")[:160],
|
|
"overlap": overlap,
|
|
"confidence_score": item.get("confidence_score"),
|
|
}
|
|
)
|
|
risks: list[dict[str, Any]] = []
|
|
gaps: list[dict[str, Any]] = []
|
|
if "thread_context" in dropped:
|
|
risks.append({"kind": "budget_dropped_thread_context", "severity": "high"})
|
|
if "memory_context" in dropped:
|
|
risks.append({"kind": "budget_dropped_memory_context", "severity": "medium"})
|
|
if not list(thread_context.get("recent_messages") or []) and not list(thread_context.get("relevant_messages") or []):
|
|
gaps.append({"kind": "no_same_thread_context", "source": "thread_context"})
|
|
if not list(memory_context.get("memories") or []):
|
|
gaps.append({"kind": "no_layered_memory", "source": "memory_context"})
|
|
low_overlap_count = len([item for item in memory_scores if int(item.get("overlap") or 0) <= 0])
|
|
if memory_scores and low_overlap_count == len(memory_scores) and question_terms:
|
|
risks.append({"kind": "memory_present_but_low_question_overlap", "severity": "medium", "memory_count": len(memory_scores)})
|
|
budget = value.get("budget_report") if isinstance(value.get("budget_report"), dict) else {}
|
|
max_chars = int(budget.get("max_chars") or 0)
|
|
used_chars = int(budget.get("used_chars") or 0)
|
|
if max_chars and used_chars / max_chars > 0.86:
|
|
risks.append({"kind": "context_budget_pressure", "severity": "medium", "used_ratio": round(used_chars / max_chars, 4)})
|
|
return {
|
|
"packet_version": "pmflow_ai_context_retrieval_critic.v1",
|
|
"input_signal": {
|
|
"question_term_count": len(question_terms),
|
|
},
|
|
"coverage": {
|
|
"thread_recent_count": len(list(thread_context.get("recent_messages") or [])),
|
|
"thread_relevant_count": len(list(thread_context.get("relevant_messages") or [])),
|
|
"memory_count": len(list(memory_context.get("memories") or [])),
|
|
"memory_scores": memory_scores[:12],
|
|
},
|
|
"gaps": gaps,
|
|
"risks": risks,
|
|
"retrieval_policy": {
|
|
"same_thread_context_has_priority": True,
|
|
"memory_relevance_is_generic_text_overlap_plus_scope_anchor": True,
|
|
"critic_does_not_route_by_business_scene": True,
|
|
},
|
|
}
|
|
|
|
|
|
def evaluate_context_replay_case(
|
|
*,
|
|
context_pack: dict[str, Any],
|
|
expected_evidence: list[str],
|
|
) -> dict[str, Any]:
|
|
"""Deterministic replay check for context retention regressions."""
|
|
|
|
haystack = str(context_pack or "")
|
|
checks = []
|
|
for item in expected_evidence:
|
|
needle = str(item or "").strip()
|
|
if not needle:
|
|
continue
|
|
checks.append({"expected": needle[:160], "present": needle in haystack})
|
|
return {
|
|
"packet_version": "pmflow_ai_context_replay_evaluation.v1",
|
|
"passed": all(item["present"] for item in checks) if checks else True,
|
|
"checks": checks,
|
|
"observability": context_pack_observability_packet(context_pack),
|
|
}
|
|
|
|
|
|
def _section_specs(pack: dict[str, Any]) -> list[tuple[str, int, Any, Any]]:
|
|
return [
|
|
("thread_context", 10, pack.get("thread_context"), _shrink_thread_context),
|
|
("memory_context", 15, pack.get("memory_context"), _shrink_mapping),
|
|
("topic_context", 18, pack.get("topic_context"), _shrink_mapping),
|
|
("object_context", 24, pack.get("object_context"), _shrink_mapping),
|
|
("workspace_context", 30, pack.get("workspace_context"), _shrink_mapping),
|
|
("runtime_context", 40, pack.get("runtime_context"), _shrink_mapping),
|
|
("surface_context", 50, pack.get("surface_context"), _shrink_mapping),
|
|
]
|
|
|
|
|
|
def _memory_context(value: Any) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
memories = []
|
|
for item in list(value.get("memories") or [])[:16]:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
memories.append(
|
|
{
|
|
"memory_id": str(item.get("memory_id") or "")[:80],
|
|
"memory_scope": str(item.get("memory_scope") or "")[:40],
|
|
"scope_ref": str(item.get("scope_ref") or "")[:160],
|
|
"summary": _clean_text(item.get("summary"), limit=1600),
|
|
"durable_facts": _shrink_mapping(list(item.get("durable_facts") or [])[:8], 3000),
|
|
"open_loops": _shrink_mapping(list(item.get("open_loops") or [])[:6], 2000),
|
|
"decisions": _shrink_mapping(list(item.get("decisions") or [])[:6], 2000),
|
|
"preferences": _shrink_mapping(list(item.get("preferences") or [])[:6], 1600),
|
|
"object_refs": _shrink_mapping(list(item.get("object_refs") or [])[:6], 1200),
|
|
"confidence": str(item.get("confidence") or "")[:32],
|
|
"confidence_score": item.get("confidence_score"),
|
|
}
|
|
)
|
|
if not memories:
|
|
return {}
|
|
return {
|
|
"packet_version": value.get("packet_version") or "pmflow_ai_layered_memory_pack.v1",
|
|
"scope_refs": value.get("scope_refs") if isinstance(value.get("scope_refs"), dict) else {},
|
|
"memories": [item for item in memories if item.get("summary") or item.get("durable_facts")],
|
|
"context_policy": {
|
|
"long_term_memory_is_semantic_compaction_not_raw_transcript": True,
|
|
"same_thread_recent_messages_override_memory_when_conflicting": True,
|
|
},
|
|
}
|
|
|
|
|
|
def _topic_context(*, conversation_context: dict[str, Any] | None) -> dict[str, Any]:
|
|
context = conversation_context if isinstance(conversation_context, dict) else {}
|
|
pack = context.get("topic_workspace_pack") if isinstance(context.get("topic_workspace_pack"), dict) else {}
|
|
rows = []
|
|
for item in list(pack.get("topic_workspaces") or [])[:8]:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
rows.append(
|
|
{
|
|
"topic_id": str(item.get("topic_id") or "")[:80],
|
|
"topic_signature": str(item.get("topic_signature") or "")[:80],
|
|
"topic_summary": _clean_text(item.get("topic_summary"), limit=1200),
|
|
"current_objective": _clean_text(item.get("current_objective"), limit=900),
|
|
"known_assumptions": _shrink_mapping(list(item.get("known_assumptions") or [])[:8], 2400),
|
|
"prior_proposals": _shrink_mapping(list(item.get("prior_proposals") or [])[:8], 3000),
|
|
"open_questions": _shrink_mapping(list(item.get("open_questions") or [])[:6], 1600),
|
|
"monetization_hypothesis": _clean_text(item.get("monetization_hypothesis"), limit=800),
|
|
"next_actionable_package": _clean_text(item.get("next_actionable_package"), limit=1200),
|
|
"linked_thread_ids": list(item.get("linked_thread_ids") or [])[:12],
|
|
"linked_run_ids": list(item.get("linked_run_ids") or [])[:12],
|
|
"linked_memory_ids": list(item.get("linked_memory_ids") or [])[:12],
|
|
"confidence_score": item.get("confidence_score"),
|
|
}
|
|
)
|
|
if not rows:
|
|
return {}
|
|
return {
|
|
"packet_version": pack.get("packet_version") or "pmflow_ai_topic_context_pack.v1",
|
|
"topic_workspaces": rows,
|
|
"context_policy": {
|
|
"topic_workspace_supports_cross_thread_continuity": True,
|
|
"matching_is_semantic_not_title_or_business_keyword": True,
|
|
},
|
|
}
|
|
|
|
|
|
def _object_context(*, conversation_context: dict[str, Any] | None, tenant_project_context: dict[str, Any] | None) -> dict[str, Any]:
|
|
context = conversation_context if isinstance(conversation_context, dict) else {}
|
|
tenant = tenant_project_context if isinstance(tenant_project_context, dict) else {}
|
|
surface = tenant.get("surface_context") if isinstance(tenant.get("surface_context"), dict) else {}
|
|
out: dict[str, Any] = {}
|
|
for key in ("object_ledger", "current_project_id", "last_subject", "last_entity_ids"):
|
|
value = context.get(key)
|
|
if value not in (None, "", [], {}):
|
|
out[key] = value
|
|
for key in ("current_focus_object", "visible_objects"):
|
|
value = surface.get(key)
|
|
if value not in (None, "", [], {}):
|
|
out[key] = value
|
|
capsules = []
|
|
for item in list(context.get("context_capsules") or [])[:12]:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
if str(item.get("capsule_type") or item.get("type") or "").strip() in {
|
|
"pmflow_ai_object_ledger",
|
|
"object_ledger",
|
|
"session_object_ledger",
|
|
}:
|
|
capsules.append(_shrink_mapping(item, 2500))
|
|
if capsules:
|
|
out["object_capsules"] = capsules
|
|
return out
|
|
|
|
|
|
def _workspace_context(*, conversation_context: dict[str, Any] | None) -> dict[str, Any]:
|
|
context = conversation_context if isinstance(conversation_context, dict) else {}
|
|
out: dict[str, Any] = {}
|
|
for key in ("active_workspace", "thread_workspace", "agent_workspace"):
|
|
value = context.get(key)
|
|
if value not in (None, "", [], {}):
|
|
out[key] = _shrink_mapping(value, 5000)
|
|
return out
|
|
|
|
|
|
def _surface_context(tenant_project_context: dict[str, Any] | None) -> dict[str, Any]:
|
|
tenant = tenant_project_context if isinstance(tenant_project_context, dict) else {}
|
|
surface = tenant.get("surface_context") if isinstance(tenant.get("surface_context"), dict) else {}
|
|
runtime = surface.get("space_runtime_context") if isinstance(surface.get("space_runtime_context"), dict) else {}
|
|
workbench = runtime.get("workbench") if isinstance(runtime.get("workbench"), dict) else surface.get("workbench")
|
|
workbench = workbench if isinstance(workbench, dict) else {}
|
|
return {
|
|
"workbench_counts": {
|
|
key: len(value) if isinstance(value, list) else 0
|
|
for key, value in workbench.items()
|
|
if isinstance(value, list)
|
|
},
|
|
"visible_objects": list(surface.get("visible_objects") or [])[:12] if isinstance(surface.get("visible_objects"), list) else [],
|
|
"space_operating_packet": _shrink_mapping(runtime.get("space_operating_packet"), 5000)
|
|
if isinstance(runtime.get("space_operating_packet"), dict)
|
|
else {},
|
|
}
|
|
|
|
|
|
def _runtime_context(*, conversation_context: dict[str, Any] | None, tenant_project_context: dict[str, Any] | None) -> dict[str, Any]:
|
|
context = conversation_context if isinstance(conversation_context, dict) else {}
|
|
tenant = tenant_project_context if isinstance(tenant_project_context, dict) else {}
|
|
out: dict[str, Any] = {}
|
|
for key in ("session_id", "conversation_scope", "surface", "source_channel"):
|
|
value = context.get(key)
|
|
if value not in (None, "", [], {}):
|
|
out[key] = value
|
|
for key in ("current_project_id", "tenant_id", "user_id"):
|
|
value = tenant.get(key)
|
|
if value not in (None, "", [], {}):
|
|
out[key] = value
|
|
return out
|
|
|
|
|
|
def _shrink_thread_context(value: Any, limit_chars: int) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
remaining = max(1600, int(limit_chars or 0))
|
|
recent_limit = 10 if remaining > 9000 else 6 if remaining > 4500 else 3
|
|
relevant_limit = 10 if remaining > 9000 else 6 if remaining > 4500 else 3
|
|
out = slim_thread_context_pack(value, recent_limit=recent_limit, relevant_limit=relevant_limit)
|
|
if _estimated_size(out) <= remaining:
|
|
return out
|
|
if isinstance(out.get("recent_messages"), list):
|
|
out["recent_messages"] = out["recent_messages"][-max(1, recent_limit // 2):]
|
|
if isinstance(out.get("relevant_messages"), list):
|
|
out["relevant_messages"] = out["relevant_messages"][: max(1, relevant_limit // 2)]
|
|
return out
|
|
|
|
|
|
def _shrink_mapping(value: Any, limit_chars: int) -> Any:
|
|
if value in (None, "", [], {}):
|
|
return {} if isinstance(value, dict) else value
|
|
if isinstance(value, dict):
|
|
out: dict[str, Any] = {}
|
|
used = 0
|
|
for key, item in value.items():
|
|
slim = _shrink_mapping(item, max(600, int(limit_chars / 3) if limit_chars else 1200))
|
|
size = _estimated_size({key: slim})
|
|
if used + size > max(800, int(limit_chars or 1200)):
|
|
continue
|
|
out[str(key)[:80]] = slim
|
|
used += size
|
|
return out
|
|
if isinstance(value, list):
|
|
rows = []
|
|
used = 0
|
|
for item in value[:24]:
|
|
slim = _shrink_mapping(item, max(500, int(limit_chars / 4) if limit_chars else 1000))
|
|
size = _estimated_size(slim)
|
|
if used + size > max(800, int(limit_chars or 1200)):
|
|
break
|
|
rows.append(slim)
|
|
used += size
|
|
return rows
|
|
if isinstance(value, str):
|
|
return _clean_text(value, limit=max(200, min(1600, int(limit_chars or 1200))))
|
|
return value
|
|
|
|
|
|
def _budget_row(section_name: str, priority: int, status: str, original: Any, included: Any) -> dict[str, Any]:
|
|
return {
|
|
"section": section_name,
|
|
"priority": priority,
|
|
"status": status,
|
|
"original_chars": _estimated_size(original),
|
|
"included_chars": _estimated_size(included),
|
|
}
|
|
|
|
|
|
def _estimated_size(value: Any) -> int:
|
|
return len(str(value or ""))
|
|
|
|
|
|
def _text_terms(value: Any) -> set[str]:
|
|
text = str(value or "").lower()
|
|
alnum = {item for item in re.findall(r"[a-z0-9_]{2,}", text)}
|
|
cjk = re.findall(r"[\u4e00-\u9fff]", text)
|
|
cjk_terms = {"".join(cjk[index : index + 2]) for index in range(max(0, len(cjk) - 1))}
|
|
return {item for item in [*alnum, *cjk_terms] if item}
|
|
|
|
|
|
def _clean_text(value: Any, *, limit: int) -> str:
|
|
return " ".join(str(value or "").replace("\x00", " ").split())[: max(0, limit)]
|
|
|
|
|
|
__all__ = [
|
|
"PMFlowAIContextPackBuilder",
|
|
"PMFlowAIContextBudgeter",
|
|
"context_pack_memory_claim_rows",
|
|
"context_pack_observability_packet",
|
|
"context_pack_retrieval_critic_packet",
|
|
"context_pack_used_context_packet",
|
|
"context_pack_thread_claim_rows",
|
|
"evaluate_context_replay_case",
|
|
"slim_pmflow_context_pack",
|
|
]
|