diff --git a/README.md b/README.md index 5fc18f17ce..d74fe4484a 100644 --- a/README.md +++ b/README.md @@ -2677,6 +2677,9 @@ Automatic and manual compaction process all selected history in chronological chunks before replacing it. `summarization.trim_tokens_to_summarize` bounds raw input per step; generated summaries use at most half to leave room for source. Oversized summaries retain marked head/tail text; raw history is never clipped. +Native document titles, context and direct plain-text bodies are included in the +summary input. Binary documents and remote media are not decoded or fetched; +nested document blocks are not traversed. A pass is limited to 64 chunks and leaves history unchanged on failure or limit exhaustion. A limit breach starts a five-minute per-process cooldown for the same thread/history window and policy, including rebuilt middleware. A larger diff --git a/backend/docs/summarization.md b/backend/docs/summarization.md index 73bd603b37..a5d937e392 100644 --- a/backend/docs/summarization.md +++ b/backend/docs/summarization.md @@ -25,6 +25,17 @@ PII-redacted before splitting, then escaped before insertion into prompt blocks. No human-anchor filter or successful "too long" placeholder can discard selected source. Recent-message retention and AI/Tool partition boundaries stay unchanged. +State system messages, tagged dynamic-context reminders and the latest genuine +user request remain outside compression. Summary source conversion uses message +copies to include native document `text`/`content` plaintext, title and context +in their original block order, before redaction and chunking. Nonempty string +titles and context are retained even when the source is opaque, absent or +malformed; unsupported source bodies are not stringified. A `content` source +contributes only direct strings and `text` blocks. Nested `document` blocks are +not traversed, including their metadata; recursive extraction is outside this +fix's scope. Compaction does not decode binary documents, fetch remote media or +change checkpoint content. + Only after all chunks succeed are hooks, the compaction event, history capture and message replacement committed once. Any failed chunk leaves active history unchanged; manual `/compact` reports generation failure. A pass processes at most diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index 30ac852bde..265a2d3b1b 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -19,14 +19,11 @@ Preserve the recognition sample and 5,000,000 UTF-8-byte guard. Count incrementa on parsing or field-limit errors, report an undetermined count rather than a physical-line or sample total. -Keep state `SystemMessage`s; inject transient instructions via request wrappers. -Skip fully rescued partitions. Fold all raw source in order without human -anchoring. PII-redact before splitting; budget before escaping. Generated carry: -marked head/tail within half the raw budget; at most 64 chunks/pass. Commit -hooks/history/removal only on success. Cap breaches use a bounded process-local -5m cooldown across builds for the same user/thread/executor/window/policy; -smaller source or changed policy retries immediately. See the linked contract. -`trim_tokens_to_summarize=None` uses one unbounded step. +Fold selected source in order; preserve required context. Copy document metadata +and direct plaintext; no media decoding/fetching or nested-document traversal. +PII before splitting, budgets before escaping; hooks/history/removal on success. +Chunk/carry/cooldown invariants and tests: +[contract](../../../../../docs/summarization.md). Todo compaction: [contract](../../../../../docs/summarization.md#todo-reminders). diff --git a/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py index 75113a22aa..34e5ef62a5 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py @@ -49,6 +49,46 @@ _SUMMARY_CHUNK_RETRY_LOCK = threading.Lock() _SHORTENED_SUMMARY_MARKER = "\n[summary shortened]\n" +def _document_summary_text(block: object) -> str | None: + """Expose document metadata and direct plaintext, without fetching media.""" + if not isinstance(block, dict) or block.get("type") != "document": + return None + parts = [] + for key in ("title", "context"): + value = block.get(key) + if isinstance(value, str) and value: + parts.append(f"{key}: {value}") + source = block.get("source") + body = None + if isinstance(source, dict): + if source.get("type") == "text": + body = source.get("data") + elif source.get("type") == "content": + content = source.get("content") + if isinstance(content, str): + body = content + elif isinstance(content, (list, tuple)): + body = "\n".join(part if isinstance(part, str) else part["text"] for part in content if isinstance(part, str) or (isinstance(part, dict) and part.get("type") == "text" and isinstance(part.get("text"), str))) + if isinstance(body, str) and body: + parts.append(body) + if not parts: + return None + return "\n" + "\n".join(["Document:", *parts]) + "\n" + + +def _summary_message(message: AnyMessage) -> AnyMessage: + """Use request-local text blocks; keep checkpoint content and call metadata.""" + if not isinstance(message.content, list): + return message + content = [] + changed = False + for block in message.content: + text = _document_summary_text(block) + content.append({"type": "text", "text": text} if text is not None else block) + changed = changed or text is not None + return message.model_copy(update={"content": content}) if changed else message + + class _ReusedSummary(str): """Marks a summary reused from an identical, previously no-op prompt (no LLM call).""" @@ -327,7 +367,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): # Redact the complete source before splitting: an identifier spanning a # chunk boundary must not bypass PII screening. Input remains plain text, # so the chat-message human-anchor filter is neither needed nor safe. - return redact_text(get_buffer_string(messages), getattr(self._app_config, "pii_redaction", None)) + return redact_text(get_buffer_string([_summary_message(message) for message in messages]), getattr(self._app_config, "pii_redaction", None)) def _summarize_with(self, messages_to_summarize: list[AnyMessage], previous_summary: str | None = None, *, runtime: Runtime | None = None) -> str | None: """Fold every selected source fragment before committing any removal. diff --git a/backend/tests/test_summarization_source_coverage.py b/backend/tests/test_summarization_source_coverage.py index 8aa2354b9b..ce941f84df 100644 --- a/backend/tests/test_summarization_source_coverage.py +++ b/backend/tests/test_summarization_source_coverage.py @@ -12,9 +12,11 @@ import pytest from langchain_core.language_models import BaseChatModel from langchain_core.messages import AIMessage, HumanMessage, ToolMessage, get_buffer_string from langchain_core.outputs import ChatGeneration, ChatResult +from langgraph.graph.message import add_messages from pydantic import Field from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware, SummaryGenerationError +from deerflow.config.pii_redaction_config import PiiRedactionConfig from deerflow.config.task_continuity_config import TaskContinuityConfig from deerflow.extensions import ExtensionRegistry @@ -302,3 +304,236 @@ async def test_unbounded_budget_preserves_generated_summary_bytes(asynchronous, assert dict(sections[0])["existing_summary"] == previous captured = "".join(body for step in sections for name, body in step if name == "new_messages") assert captured == get_buffer_string(selected) + + +def _document_history(source): + return { + "messages": [ + HumanMessage( + content=[ + {"type": "text", "text": "BEFORE_DOCUMENT"}, + {"type": "document", "title": "DOCUMENT_TITLE", "context": "DOCUMENT_CONTEXT", "source": source}, + {"type": "text", "text": "AFTER_DOCUMENT"}, + ], + id="old-document", + ), + AIMessage(content="I received the document.", id="old-answer"), + HumanMessage(content="CURRENT_REQUEST_STAYS_RAW", id="current-user"), + AIMessage(content="RECENT_ANSWER_STAYS_RAW", id="recent-answer"), + ] + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("source_kind", ["text", "content-string", "content-blocks"]) +@pytest.mark.parametrize("long_document", [False, True], ids=["short", "chunked"]) +async def test_native_document_text_reaches_summary_before_message_removal(asynchronous, source_kind, long_document): + paragraphs = [f"SECTION_{index:03d}: content &" for index in range(80 if long_document else 2)] + text = "DOCUMENT_BEGIN\n" + "\n".join(paragraphs) + "\nDOCUMENT_END" + expected_body = text + if source_kind == "text": + source = {"type": "text", "media_type": "text/plain", "data": text} + elif source_kind == "content-string": + source = {"type": "content", "content": text} + else: + source = {"type": "content", "content": [text[:20], {"type": "text", "text": text[20:]}, {"type": "image", "source": {"type": "base64", "data": "OPAQUE_IMAGE_SECRET"}}]} + # Native content blocks form separate text sections. The image is opaque. + expected_body = text[:20] + "\n" + text[20:] + state = _document_history(source) + original = copy.deepcopy(state) + model = RecordingSummaryModel() + middleware = _middleware(model) + runtime = SimpleNamespace(context={"thread_id": "document-coverage"}) + + update = await middleware.abefore_model(state, runtime) if asynchronous else middleware.before_model(state, runtime) + + assert update is not None + retained = add_messages(state["messages"], update["messages"]) + assert {message.id for message in retained} == {"current-user", "recent-answer"} + captured = "".join(body for prompt in model.prompts for name, body in _raw_sections(prompt) if name == "new_messages") + assert captured.count(expected_body) == 1, "the complete document must reach the summary model exactly once, including its middle" + assert captured.index("BEFORE_DOCUMENT") < captured.index("DOCUMENT_TITLE") < captured.index("DOCUMENT_CONTEXT") < captured.index("DOCUMENT_BEGIN") + assert captured.index("DOCUMENT_END") < captured.index("AFTER_DOCUMENT") + assert "OPAQUE_IMAGE_SECRET" not in captured + assert state == original, "summary source conversion must not rewrite checkpoint messages" + assert all(sum(len(body) for _, body in _raw_sections(prompt)) <= _RAW_BUDGET for prompt in model.prompts) + if long_document: + assert len(model.prompts) > 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("opaque", [False, True], ids=["plaintext", "opaque"]) +async def test_native_document_metadata_and_body_are_redacted_before_summary(asynchronous, opaque): + source = {"type": "base64", "data": "OPAQUE_DOCUMENT_SECRET"} if opaque else {"type": "text", "media_type": "text/plain", "data": "Contact document@example.com. BODY_UNTRUSTED"} + state = _document_history(source) + state["messages"][0].content[1].update(title="title@example.com ", context="context@example.com UNTRUSTED") + original = copy.deepcopy(state) + model = RecordingSummaryModel() + middleware = _middleware(model, app_config=SimpleNamespace(pii_redaction=PiiRedactionConfig(enabled=True, token_secret="offline-document-test-secret"))) + runtime = SimpleNamespace(context={"thread_id": "document-redaction"}) + + update = await middleware.abefore_model(state, runtime) if asynchronous else middleware.before_model(state, runtime) + + assert update is not None + prompt = "\n".join(model.prompts) + assert "UNTRUSTED" in prompt + assert "<system>" in prompt and "</new_messages>" in prompt + assert "@example.com" not in prompt + assert "OPAQUE_DOCUMENT_SECRET" not in prompt + if not opaque: + assert "<system>BODY_UNTRUSTED</system>" in prompt + assert state == original + + +@pytest.mark.asyncio +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("opaque", [False, True], ids=["plaintext", "metadata-only"]) +async def test_document_midway_summary_failure_keeps_history_and_skips_hooks(asynchronous, opaque): + source = {"type": "base64", "data": "OPAQUE_DOCUMENT_SECRET"} if opaque else {"type": "content", "content": [{"type": "text", "text": "DOCUMENT_SOURCE " * 300}]} + state = _document_history(source) + if opaque: + state["messages"][0].content[1]["context"] = "DOCUMENT_CONTEXT " * 300 + original = copy.deepcopy(state) + model = RecordingSummaryModel(fail_after=1) + hooks = [] + middleware = _middleware(model, before_summarization=[hooks.append]) + runtime = SimpleNamespace(context={"thread_id": "document-failure"}) + + update = await middleware.abefore_model(state, runtime) if asynchronous else middleware.before_model(state, runtime) + + assert len(model.prompts) > 1, "failure must happen after a successful document chunk" + assert update is None + assert hooks == [] + assert state == original + + +@pytest.mark.asyncio +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize( + "source", + [ + {"type": "base64", "media_type": "application/pdf", "data": "OPAQUE_DOCUMENT_SECRET"}, + {"type": "url", "url": "https://example.invalid/OPAQUE_DOCUMENT_SECRET"}, + {"type": "file", "file_id": "OPAQUE_DOCUMENT_SECRET"}, + {"type": "text", "data": {"unexpected": "OPAQUE_DOCUMENT_SECRET"}}, + {"type": "content", "content": [{"type": "reasoning", "text": "OPAQUE_DOCUMENT_SECRET"}]}, + {"type": "text", "data": ""}, + {"type": "content", "content": None}, + {"type": "unknown", "data": "OPAQUE_DOCUMENT_SECRET"}, + None, + "OPAQUE_DOCUMENT_SECRET", + ], + ids=["base64", "url", "file-id", "malformed-text", "reasoning", "empty-text", "empty-content", "unknown", "missing", "malformed-source"], +) +async def test_document_metadata_reaches_summary_without_stringifying_opaque_sources(source, asynchronous): + state = _document_history(source) + original = copy.deepcopy(state) + model = RecordingSummaryModel() + middleware = _middleware(model) + runtime = SimpleNamespace(context={"thread_id": "opaque-document"}) + update = await middleware.abefore_model(state, runtime) if asynchronous else middleware.before_model(state, runtime) + assert update is not None + retained = add_messages(state["messages"], update["messages"]) + assert {message.id for message in retained} == {"current-user", "recent-answer"} + captured = "".join(body for prompt in model.prompts for name, body in _raw_sections(prompt) if name == "new_messages") + assert captured.count("Document:\ntitle: DOCUMENT_TITLE\ncontext: DOCUMENT_CONTEXT") == 1 + assert captured.index("BEFORE_DOCUMENT") < captured.index("DOCUMENT_TITLE") < captured.index("DOCUMENT_CONTEXT") < captured.index("AFTER_DOCUMENT") + assert "OPAQUE_DOCUMENT_SECRET" not in captured + assert state == original + assert all(sum(len(body) for _, body in _raw_sections(prompt)) <= _RAW_BUDGET for prompt in model.prompts) + + +@pytest.mark.parametrize(("title", "context", "expected"), [("TITLE_ONLY", "", "title: TITLE_ONLY"), (None, "CONTEXT_ONLY", "context: CONTEXT_ONLY"), ("", "", None), ({"unexpected": "SECRET"}, ["SECRET"], None)]) +def test_opaque_document_metadata_accepts_only_nonempty_text(title, context, expected): + state = _document_history({"type": "base64", "data": "OPAQUE_DOCUMENT_SECRET"}) + state["messages"][0].content[1].update(title=title, context=context) + original = copy.deepcopy(state) + middleware = _middleware(RecordingSummaryModel()) + + captured = middleware._summary_source_text(state["messages"]) + + if expected is None: + assert captured == get_buffer_string(state["messages"]) + else: + assert f"Document:\n{expected}\n" in captured + assert "SECRET" not in captured + assert state == original + + +@pytest.mark.asyncio +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +async def test_opaque_document_metadata_is_fully_chunked_before_message_removal(asynchronous): + state = _document_history({"type": "base64", "data": "OPAQUE_DOCUMENT_SECRET"}) + context = "\n".join(f"CONTEXT_SECTION_{index:03d}: preserved &" for index in range(80)) + state["messages"][0].content[1]["context"] = context + original = copy.deepcopy(state) + model = RecordingSummaryModel() + middleware = _middleware(model) + runtime = SimpleNamespace(context={"thread_id": "opaque-document-chunking"}) + + update = await middleware.abefore_model(state, runtime) if asynchronous else middleware.before_model(state, runtime) + + assert update is not None + retained = add_messages(state["messages"], update["messages"]) + assert {message.id for message in retained} == {"current-user", "recent-answer"} + captured = "".join(body for prompt in model.prompts for name, body in _raw_sections(prompt) if name == "new_messages") + assert captured.count(context) == 1 + assert captured.index("DOCUMENT_TITLE") < captured.index(context) < captured.index("AFTER_DOCUMENT") + assert "OPAQUE_DOCUMENT_SECRET" not in captured + assert len(model.prompts) > 1 + assert all(sum(len(body) for _, body in _raw_sections(prompt)) <= _RAW_BUDGET for prompt in model.prompts) + assert state == original + + +@pytest.mark.asyncio +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +async def test_document_content_keeps_direct_text_without_recursing_into_nested_documents(asynchronous): + state = _document_history( + { + "type": "content", + "content": [ + "DIRECT_BEFORE", + {"type": "document", "title": "NESTED_DOCUMENT_TITLE", "context": "NESTED_DOCUMENT_CONTEXT", "source": {"type": "text", "data": "NESTED_DOCUMENT_BODY"}}, + {"type": "text", "text": "DIRECT_AFTER"}, + ], + } + ) + original = copy.deepcopy(state) + model = RecordingSummaryModel() + middleware = _middleware(model) + runtime = SimpleNamespace(context={"thread_id": "nested-document-scope"}) + + update = await middleware.abefore_model(state, runtime) if asynchronous else middleware.before_model(state, runtime) + + assert update is not None + captured = "".join(body for prompt in model.prompts for name, body in _raw_sections(prompt) if name == "new_messages") + assert "DIRECT_BEFORE\nDIRECT_AFTER" in captured + assert "DOCUMENT_TITLE" in captured and "DOCUMENT_CONTEXT" in captured + assert "NESTED_DOCUMENT" not in captured + assert state == original + + +@pytest.mark.asyncio +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +async def test_native_document_compaction_through_real_agent_graph(asynchronous): + from langchain.agents import create_agent + + from deerflow.agents.thread_state import ThreadState + + summary_model = RecordingSummaryModel() + lead_model = RecordingSummaryModel(output_text="Finished.") + middleware = _middleware(summary_model) + graph = create_agent(model=lead_model, tools=[], middleware=[middleware], state_schema=ThreadState) + state = _document_history({"type": "text", "media_type": "text/plain", "data": "REAL_GRAPH_DOCUMENT_REQUIREMENT"}) + original = copy.deepcopy(state) + + result = await graph.ainvoke(state, context={"thread_id": "document-graph"}) if asynchronous else graph.invoke(state, context={"thread_id": "document-graph"}) + + assert "old-document" not in {message.id for message in result["messages"]} + assert "current-user" in {message.id for message in result["messages"]} + assert "REAL_GRAPH_DOCUMENT_REQUIREMENT" in "\n".join(summary_model.prompts) + assert result["summary_text"] + assert lead_model.prompts + assert state == original