mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-11-01 13:25:18 +08:00
fix(pipeline): deliver explicit attachments without scanning Box
Co-authored-by: multica-agent <[email protected]>
This commit is contained in:
RockChinQ
co-authored by
multica-agent
multica-agent
parent
709e5e7a1d
commit
f9b8883729
@@ -0,0 +1,223 @@
|
||||
# LANG-17 post-merge integration
|
||||
|
||||
## Fixed inputs and scope
|
||||
|
||||
Core base: `7974a430680f813eae58e60d85c58ef935a144f3` (PR #2706 merge).
|
||||
Plugins base: `31a05080b58f2cda8b1acecee668b2cda5503c4b`.
|
||||
SDK base: `7ac774f72eadb104ffb2d5b0c2c7850b667f65ed` (published version 0.7.13).
|
||||
All three integration branches are `agent/langbot-local-qa/602fe0ce0551`.
|
||||
The previous core branch at `3794a7ee` has a separate preserved worktree;
|
||||
the pre-merge migration stash and LANG-16 branches remain intact.
|
||||
|
||||
This is a local integration combination, not a released distribution. The SDK
|
||||
package still reports 0.7.13 after a local build, so the Git revision and patched
|
||||
manager file hash must accompany that version. The core dependency lock still
|
||||
points at the published SDK; deployment through that lock alone omits this SDK fix.
|
||||
No new PR, merge, release, deployment, or default-branch write is performed.
|
||||
|
||||
## Reconciliation against LANG-16
|
||||
|
||||
| Earlier change | Merge/upstream disposition | Integration action |
|
||||
| --- | --- | --- |
|
||||
| Core catch-all group gating (`e212f9ea`) | Covered differently: upstream checks ordinary routed group messages at the response stage, including explicit filters; interaction/Dify callback exceptions remain | Preserve upstream policy; do not restore the old filtered-binding bypass |
|
||||
| Core busy-session steering gate / stable random draw (`e212f9ea`) | Absent; new tests reproduce unrelated input being claimed and prefixes not being stripped | Port only controller/random changes; preserve original queued message via a copy |
|
||||
| Original monitoring row on stage exception (`e212f9ea`) | Absent; direct failure test shows no status update before error-detail failure | Port the original-row error transition before optional detail recording |
|
||||
| Groq outbound metadata (`fefe4d85`) | Upstream independently covers message, tool-call and function envelopes, separated from reasoning sanitization | No port; rerun requester regressions |
|
||||
| Migration install diagnostics (`079e945b`) | Absent | Port allowlisted error kind/HTTP status, UI translations, MCP and skill documentation together |
|
||||
| LocalAgent lazy Box (`f33146a0`) | Absent from current plugins master | Port lock-protected attachment/tool preparation and tests; no ordinary-answer acquisition |
|
||||
| SDK no-listener event admission (`11b52589`) | Absent from SDK 0.7.13/current master | Port component-kind admission and subscriber-contract tests; retain ordered listener dispatch |
|
||||
| Old LANG-17 missing-table migration patch | Consolidated 0039 already skips absent tables | Do not apply the old 0039/0040 patch |
|
||||
| Missing columns in partial historical tables | Still fails in consolidated 0039: all eight new cases fail before repair | Check available columns before creating each index; preserve data and unrelated indexes |
|
||||
| Historical inline TTL / reference-only writes / CAS | Upstream additions, not earlier task fixes | Keep implementation; verify cleanup, cursor, ownership, failure/rollback and read paths |
|
||||
| SQLite 3.53 flattened expiry query plan | Previous task test adjustment still necessary | Retain indexed owner lookup assertions for flattened EXISTS plans |
|
||||
|
||||
## Automated regression
|
||||
|
||||
- Core migration/storage/response/install group: **176 passed** (includes SQLite
|
||||
fresh/baseline/upgrade paths and eight new partial-schema tests).
|
||||
- Core controller, queue, error handling and LiteLLM requester group: **153 passed**.
|
||||
- Response rules, pending interaction migration and migration runtime group:
|
||||
**73 passed**. Some tests overlap the preceding groups; totals are not unique.
|
||||
- Real PostgreSQL 16 + pgvector: **49 passed**, including 28 media contracts,
|
||||
13 migration/runtime tests, and eight partial-schema cases. Each run uses one
|
||||
owned, resource-limited container. The first image lacked pgvector (35 passed,
|
||||
six environment failures); it was removed and replaced, not counted as success.
|
||||
- SDK manager: before **7 failed / 61 passed**, after **68 passed**.
|
||||
- LocalAgent Box/shared tests: before **2 failed / 32 passed**; current full
|
||||
LocalAgent suite **270 passed**, excluding the obsolete exact-version release
|
||||
assertion in `test_shared_release.py`. This is not a full plugin release gate.
|
||||
- TypeScript typecheck and production build pass. Ruff and whitespace checks pass.
|
||||
|
||||
Missing-column tests deliberately construct partial schemas; they prove migration
|
||||
tolerance, not that every such schema is a supported complete application schema.
|
||||
The migration does not fabricate missing business columns or recover missing data.
|
||||
|
||||
## Actual services and operations
|
||||
|
||||
All probes start real Core, WebSocket RT, plugin workers and frontend in task-owned
|
||||
directories. Ports: 16399, 16400/16401, 14000, 13280/13281 and optional model fixture
|
||||
16420. Runtime config, SQLite, plugins, cache, logs and tokens are isolated. Box is
|
||||
disabled and DOCKER_HOST points at a nonexistent task socket. No real QQ traffic.
|
||||
The supervisor enforces 2 GiB aggregate RSS and five consecutive >200% CPU samples,
|
||||
records processes, and cleans only owned PIDs after checking the Multica daemon PID.
|
||||
|
||||
- `live-20261018-163651`: ordinary/other-mention/unrelated burst ignored; keyword,
|
||||
bot mention, private and image controls respond. Busy Runner consumes the keyword
|
||||
follow-up but not ordinary group chatter. Image response and real browser monitoring page pass; CAS read/write
|
||||
behavior is covered separately by storage regressions. Database has 38 successful + two failed user records,
|
||||
five successful assistant records, five completed + two failed runs, and no
|
||||
nonhistorical pending/running rows. All owned processes stopped.
|
||||
- `live-20261018-163855`: deliberately invalid response-rule configuration raises
|
||||
a stage exception. Original user row is error, optional error detail is recorded;
|
||||
two positive controls still succeed. No nonhistorical pending rows remain.
|
||||
- `live-20261018-164140`: real current LocalAgent package with `box-enabled=true`
|
||||
and `exec` allowed, while Core Box is disabled. Unrelated message makes no model
|
||||
call; ordinary answer succeeds via a loopback OpenAI streaming protocol fixture.
|
||||
This verifies lazy acquisition for plain answers, not a real LLM or sandbox tool.
|
||||
- `live-20261018-164300`: copy of the task's actual v4.10.10 seed database boots
|
||||
through 0039; original account authenticates, bot and legacy pipeline survive,
|
||||
and migration preview remains available. Original seed is unchanged. The first
|
||||
probe mistakenly queried `pipelines` instead of `legacy_pipelines`; corrected
|
||||
probe passed. This does not claim real Dify/DeerFlow backend conversion coverage.
|
||||
|
||||
The 64 deliberately seeded raw historical pending rows remain historical. In-flight
|
||||
SSE excludes them, but dashboard aggregate still counts raw status. Changing that
|
||||
user-visible statistic awaits the separately identified product decision; it does
|
||||
not block independent fixes or tests above.
|
||||
|
||||
## Reproduction and evidence boundaries
|
||||
|
||||
The `tests/manual/lang17_*` probes are reusable and bounded. `lang17_live.py` runs
|
||||
the ordinary ingress test by default; LANG17_FAULT=1 selects the stage-failure case,
|
||||
LANG17_LOCALAGENT=1 selects the loopback-model LocalAgent case, and LANG17_SEED_DB
|
||||
plus LANG17_UPGRADE_ACCOUNT selects a copy of a task-owned old installation.
|
||||
Never point these at a real user's data or credentials. Build LocalAgent's package
|
||||
from the integration plugins checkout into the task evidence directory first.
|
||||
|
||||
Synthetic model/protocol tests are identified explicitly. Real Groq remains deferred.
|
||||
Process-scoped mainland egress, actual Dify/interaction provider setup and securely
|
||||
supplied real-model credentials are not available in this run. Box isolation was
|
||||
subsequently provisioned as documented below; shared Docker is never sufficient. No claim of
|
||||
overnight model/history leak freedom, mainland network success, or complete nine-item
|
||||
LANG-16 acceptance follows from the bounded tests here.
|
||||
|
||||
## Bounded multi-plugin load
|
||||
|
||||
Same integration Core with published SDK 0.7.13 versus the locally patched SDK:
|
||||
healthy reply after another plugin's blocking Runner improved from 3.162 s to
|
||||
0.389 s (control input was sent at +0.3 s). RAISE/TIMEOUT/EXIT controls stayed around
|
||||
0.39–0.41 s. This comparison targets the no-listener RPC stall, not whole-system
|
||||
throughput or all listener failures.
|
||||
|
||||
Patched combination ran 749.9 s, including 60 ten-second soak rounds and recovery,
|
||||
with 3,232 synthetic inputs. Final queue/cache/workspace counts were zero; all owned
|
||||
processes stopped. Peak RSS: Core 609.20 MiB, RT 72.95 MiB, combined RT children
|
||||
185.83 MiB, production-preview frontend 122.75 MiB. Sampled average CPU: Core
|
||||
28.33%, RT 1.44%, RT children 0.46%, frontend 0.05%. These averages include startup,
|
||||
faults and load; the 70.85 s baseline has different duration/load and is not a
|
||||
valid CPU percentage improvement comparison. Plugin-list API max was 24 ms.
|
||||
|
||||
The 160-message burst produced 100 inbound monitoring records. QueryPool's existing
|
||||
100-per-Workspace cap discards queued oldest work under overload; the 60-record
|
||||
difference is consistent with that policy. There is no per-rejection log to prove
|
||||
each individual discarded input. Do not describe all 3,232 inputs as successful
|
||||
replies. The 3,000 unrelated messages and 60 soak controls are separately present.
|
||||
Final persisted states: user success 3,168 / error 4; assistant success 168 / error 1.
|
||||
No pending monitoring rows remain at recovery. This load has no real LLM history.
|
||||
|
||||
## Box isolation and an additional wrapper regression
|
||||
|
||||
Created task machine `lang17-qa-602fe0ce` with OrbStack isolated mode, networking
|
||||
isolation, no SSH-agent forwarding, 1 GiB memory / 1 CPU / 3 GiB disk, and exactly
|
||||
one selective mount: this task's Box directory. Installed a new Docker daemon;
|
||||
its ID differs from the existing daemon and its initial container inventory was
|
||||
empty. A task-only `docker` wrapper pins every SDK Docker CLI call to that machine
|
||||
and its explicit Unix socket. Existing host Docker is not used by Box.
|
||||
|
||||
Alpine initially did not enable delegated cgroup controllers; the first resource-
|
||||
limited container failed on missing `pids.max`. Moved only this machine's init
|
||||
processes into a child cgroup, enabled cpu/memory/pids controllers, and restarted
|
||||
only its newly installed Docker service. A bounded Docker smoke command passed.
|
||||
No host cgroup, Docker configuration or existing service was changed.
|
||||
|
||||
`live-20261018-170317`: real LocalAgent + Core + RT + Box, loopback model protocol.
|
||||
Unrelated input -> 0 sessions, plain answer -> 0 sessions, exec tool -> 1 session.
|
||||
The model fixture receives actual `LANG17_TOOL_OK` stdout from the container. Box
|
||||
health is live. Container sample: 1.078 MiB RSS of a 512 MiB limit, 2 PIDs, 0.00%
|
||||
CPU at sampling; this is an idle sample, not peak container usage.
|
||||
|
||||
This run also exposed a pre-existing wrapper outbox scan on ordinary replies,
|
||||
contrary to the documented explicit attachment contract. Removed that implicit
|
||||
RPC/scan and preserved explicit file-only Message/MessageChunk output. Six new
|
||||
regressions fail before repair (four unexpected scans, two missing file-only
|
||||
outputs); all 27 wrapper tests pass afterward. Final real Box rerun is recorded
|
||||
in the attached run summary. No implicit outbox discovery is promised.
|
||||
|
||||
Final Box rerun `live-20261018-170523` passes with session counts 0/0/1, real
|
||||
container stdout, healthy Box, no implicit collection warning, and no surviving
|
||||
Core/RT/Box/frontend/probe processes. Shared Docker container IDs are unchanged.
|
||||
The task-owned isolated machine is stopped (not a persistent service handoff);
|
||||
its bounded disk is retained for audit. No daemon/container cleanup crossed into
|
||||
the shared engine. The initial cgroup failure is excluded from acceptance.
|
||||
|
||||
## Installation failure paths
|
||||
|
||||
`live-20261018-170848` exercises both actual migration-install and extension-install
|
||||
APIs against a loopback marketplace fault server. HTTP 503, disconnect, truncated
|
||||
body, timeout and offline cases all terminate (10/10). Timeout takes about 15.2 s;
|
||||
other cases take about 0.02–1.04 s. All five migration results contain allowlisted
|
||||
diagnostics, preserving the failure reason without raw request secrets. These are
|
||||
fault simulations, not China-mainland network tests.
|
||||
|
||||
The first EventListener matrix mixed LBLOCK with a Runner substring match for
|
||||
BLOCK, causing two intentional stalls. It is retained as exploratory evidence;
|
||||
the corrected fixture matches Runner fault tokens exactly and is rerun separately.
|
||||
Subscriber-order and prevent-postorder controls passed in that initial run.
|
||||
|
||||
Actual public-marketplace migration install `live-20261018-171007`: downloaded
|
||||
and initialized LocalAgent 0.2.1, task completed in 7.11 s, and the synthetic
|
||||
4.10.10 pipeline reported migrated. This checks current Core/RT installation,
|
||||
not whether that published plugin includes this task's lazy-Box source patch.
|
||||
|
||||
Actual extension marketplace install `live-20261018-171058` also succeeds. Both
|
||||
success paths use fresh task data and the real public marketplace; no existing
|
||||
user installation is reused. See the sanitized installation summary for timing.
|
||||
|
||||
## Source checkpoint caveat
|
||||
|
||||
The probes were run on the fixed merge checkout with an uncommitted task diff,
|
||||
so their raw metrics record the merge SHA. Core implementation during the long
|
||||
load corresponds to checkpoint `709e5e7a` (reconstructed and committed afterward).
|
||||
The final commit additionally removes wrapper scans and fixes file-only output;
|
||||
that change has its own 27-test and real Box/installation regressions. The long
|
||||
load was not repeated after that small wrapper change. Exact final three-repo
|
||||
SHAs and installed-code hashes accompany the evidence manifest.
|
||||
|
||||
Corrected listener matrix `live-20261018-171152` passes subscription mutation order
|
||||
and prevent-postorder controls. Healthy response latency is 3.156 s for LBLOCK
|
||||
(listener-only fault) and 3.161 s for BLOCK in a plugin that also has listeners;
|
||||
RAISE/TIMEOUT/EXIT controls are about 0.39–0.41 s. Thus skipping no-listener workers
|
||||
does not isolate a synchronous blocking EventListener. Queues recover to zero,
|
||||
with no owned process survivors. Changing listener failure/timeout propagation
|
||||
requires a separate contract decision; no parallel dispatch was introduced.
|
||||
|
||||
At long-load recovery, Core retains 200 completed + 9 running task wrappers and
|
||||
two RT worker children; Core RSS is 565.36 MiB versus 540.44 MiB before plugin
|
||||
installation. The health field `id_index=3189` is a monotonically increasing ID
|
||||
counter, not a collection of retained tasks. This is not evidence of 3,189 live
|
||||
tasks or an overnight memory-leak diagnosis.
|
||||
|
||||
## Outstanding acceptance inputs
|
||||
|
||||
- Real model credentials through a secure mechanism for growing-history/image
|
||||
workloads; the loopback model test is not a substitute. Real Groq stays deferred.
|
||||
- A process-scoped mainland exit without changing the user's VPN/routing.
|
||||
- A real Dify/interactive callback test environment. Callback unit/contracts pass,
|
||||
but live third-party callback acceptance is not claimed.
|
||||
- Product decision on raw historical Running totals versus effective leases;
|
||||
preserve current behavior until decided. Blocking EventListener behavior remains
|
||||
an explicit serial-contract limit, not a claim of complete fault isolation.
|
||||
|
||||
All independently executable work in this integration pass is delivered. These
|
||||
gaps remain on LANG-16's nine-item acceptance ledger; neither the merged PR nor
|
||||
this task-branch push constitutes complete project acceptance.
|
||||
@@ -10,7 +10,6 @@ from .. import stage
|
||||
|
||||
import langbot_plugin.api.entities.builtin.platform.message as platform_message
|
||||
import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
|
||||
import langbot_plugin.api.entities.builtin.provider.message as provider_message
|
||||
import langbot_plugin.api.entities.events as events
|
||||
|
||||
|
||||
@@ -27,50 +26,6 @@ class ResponseWrapper(stage.PipelineStage):
|
||||
async def initialize(self, pipeline_config: dict):
|
||||
pass
|
||||
|
||||
def _is_final_assistant_message(self, result) -> bool:
|
||||
"""Whether *result* is the agent's final, tool-call-free answer.
|
||||
|
||||
Intermediate streaming chunks and tool-call rounds must NOT trigger
|
||||
outbound attachment collection — only the terminal assistant message.
|
||||
"""
|
||||
if getattr(result, 'role', None) != 'assistant':
|
||||
return False
|
||||
if result.tool_calls:
|
||||
return False
|
||||
if isinstance(result, provider_message.MessageChunk):
|
||||
return bool(result.is_final)
|
||||
return True
|
||||
|
||||
async def _append_outbound_attachments(
|
||||
self,
|
||||
query: pipeline_query.Query,
|
||||
message_chain: platform_message.MessageChain,
|
||||
) -> None:
|
||||
"""Collect sandbox outbox files and append them to *message_chain*.
|
||||
|
||||
Runs at most once per query (guarded by a query variable) and never
|
||||
raises into the pipeline — attachment delivery is best-effort.
|
||||
"""
|
||||
if query.variables.get('_sandbox_outbound_collected'):
|
||||
return
|
||||
box_service = getattr(self.ap, 'box_service', None)
|
||||
if box_service is None or not getattr(box_service, 'available', False):
|
||||
return
|
||||
query.variables['_sandbox_outbound_collected'] = True
|
||||
try:
|
||||
attachments = await box_service.collect_outbound_attachments(query)
|
||||
except Exception as e:
|
||||
self.ap.logger.warning(f'Outbound attachment collection failed: {e}')
|
||||
return
|
||||
for att in attachments:
|
||||
att_type = att.get('type')
|
||||
if att_type == 'Image':
|
||||
message_chain.append(platform_message.Image(base64=att['base64']))
|
||||
elif att_type == 'Voice':
|
||||
message_chain.append(platform_message.Voice(base64=att['base64']))
|
||||
else:
|
||||
message_chain.append(platform_message.File(name=att.get('name', 'file'), base64=att['base64']))
|
||||
|
||||
async def process(
|
||||
self,
|
||||
query: pipeline_query.Query,
|
||||
@@ -107,7 +62,8 @@ class ResponseWrapper(stage.PipelineStage):
|
||||
|
||||
reply_text = ''
|
||||
|
||||
if result.content: # 有内容
|
||||
# Runners submit attachments explicitly; the wrapper never scans a Box.
|
||||
if result.content or getattr(result, 'attachments', None):
|
||||
reply_text = str(result.get_content_platform_message_chain())
|
||||
|
||||
# ============= 触发插件事件 ===============
|
||||
@@ -142,11 +98,6 @@ class ResponseWrapper(stage.PipelineStage):
|
||||
reply_chain = result.get_content_platform_message_chain()
|
||||
is_plugin_reply = False
|
||||
|
||||
# Only collect Box outbox files for a terminal assistant
|
||||
# response; also accept provider final chunks here.
|
||||
if self._is_final_assistant_message(result):
|
||||
await self._append_outbound_attachments(query, reply_chain)
|
||||
|
||||
query.resp_message_chain.append(reply_chain)
|
||||
if is_plugin_reply:
|
||||
plugin_diagnostics.record_last_plugin_response_source(
|
||||
@@ -160,20 +111,6 @@ class ResponseWrapper(stage.PipelineStage):
|
||||
result_type=entities.ResultType.CONTINUE,
|
||||
new_query=query,
|
||||
)
|
||||
elif (
|
||||
isinstance(result, provider_message.MessageChunk) and result.is_final and not result.tool_calls
|
||||
):
|
||||
# A blank final chunk may carry Box outbox files, but do not
|
||||
# send an empty chain: streaming adapters treat it as final.
|
||||
reply_chain = platform_message.MessageChain([])
|
||||
await self._append_outbound_attachments(query, reply_chain)
|
||||
if len(reply_chain) > 0:
|
||||
query.resp_message_chain.append(reply_chain)
|
||||
yield entities.StageProcessResult(
|
||||
result_type=entities.ResultType.CONTINUE,
|
||||
new_query=query,
|
||||
)
|
||||
|
||||
if result.tool_calls is not None and len(result.tool_calls) > 0: # 有函数调用
|
||||
function_names = [tc.function.name for tc in result.tool_calls]
|
||||
|
||||
|
||||
@@ -4,6 +4,12 @@ from langbot_plugin.api.entities.builtin.platform.message import MessageChain, F
|
||||
from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk
|
||||
import pytest
|
||||
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from tests.factories import FakeApp, text_query
|
||||
from langbot_plugin.api.entities.builtin.provider.session import Session, LauncherTypes
|
||||
|
||||
|
||||
@pytest.mark.parametrize('cls', [Message, MessageChunk])
|
||||
def test_explicit_output_attachments_survive_platform_conversion_only(cls):
|
||||
@@ -16,3 +22,47 @@ def test_explicit_output_attachments_survive_platform_conversion_only(cls):
|
||||
def test_file_only_output(cls):
|
||||
message = cls(role='assistant', attachments=MessageChain([File(name='result.txt', base64='YQ==')]))
|
||||
assert len(message.get_content_platform_message_chain()) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize('cls', [Message, MessageChunk])
|
||||
@pytest.mark.parametrize('content,with_file', [('answer', False), ('answer', True), ('', True)])
|
||||
async def test_wrapper_delivers_explicit_files_without_scanning_box(cls, content, with_file):
|
||||
from langbot.pkg.pipeline import pipelinemgr # noqa: F401
|
||||
from langbot.pkg.pipeline.wrapper.wrapper import ResponseWrapper
|
||||
|
||||
app = FakeApp()
|
||||
app.box_service = SimpleNamespace(available=True, collect_outbound_attachments=AsyncMock(return_value=[]))
|
||||
event = SimpleNamespace(reply_message_chain=None)
|
||||
app.plugin_connector.emit_event = AsyncMock(
|
||||
return_value=SimpleNamespace(event=event, is_prevented_default=lambda: False)
|
||||
)
|
||||
query = text_query('hello')
|
||||
query.pipeline_config = {
|
||||
'output': {'misc': {'at-sender': False, 'quote-origin': False, 'track-function-calls': False}}
|
||||
}
|
||||
message = cls(
|
||||
role='assistant',
|
||||
content=content,
|
||||
attachments=MessageChain([File(name='result.txt', base64='YQ==')]) if with_file else None,
|
||||
)
|
||||
if isinstance(message, MessageChunk):
|
||||
message.is_final = True
|
||||
query.resp_messages = [message]
|
||||
query.resp_message_chain = []
|
||||
app.sess_mgr.get_session = AsyncMock(
|
||||
return_value=Session(
|
||||
launcher_type=LauncherTypes.PERSON,
|
||||
launcher_id=12345,
|
||||
sender_id=12345,
|
||||
use_prompt_name='default',
|
||||
using_conversation=None,
|
||||
conversations=[],
|
||||
)
|
||||
)
|
||||
stage = ResponseWrapper(app)
|
||||
results = [result async for result in stage.process(query, 'ResponseWrapper')]
|
||||
assert results
|
||||
if with_file:
|
||||
assert query.resp_message_chain[-1][-1].name == 'result.txt'
|
||||
app.box_service.collect_outbound_attachments.assert_not_awaited()
|
||||
|
||||
Reference in New Issue
Block a user