merge: resolve conflicts with master, add inbound attachment materialization

- Delete localagent.py and test_difysvapi_runner.py (replaced by plugins) - Keep master's tool loader enhancements (byte_offset, encoding params) - Remove feature branch's artifact reference code (use sandbox paths instead) - Add _materialize_inbound_attachments in orchestrator for sandbox file staging - Keep master's test formatting and new tests - Keep master's frontend refactoring
2026-07-21 03:46:11 +00:00 · 2026-06-19 10:51:00 +08:00
parent 79a5fba06b a020ca680f
commit 03ad58ec4d
149 changed files with 6721 additions and 3197 deletions
@@ -6,6 +6,7 @@ import os
 import shutil
 import shlex
 import threading
+from contextlib import suppress
 from typing import TYPE_CHECKING, Any

 import pydantic
@@ -285,11 +286,9 @@ class BoxStdioSessionRuntime:
                if os.path.isdir(path) and not os.path.islink(path):
                    shutil.rmtree(path, ignore_errors=True)
                else:
-                    try:
+                    # The entry may disappear between listdir and unlink if cleanup races us.
+                    with suppress(FileNotFoundError):
                        os.unlink(path)
-                    except FileNotFoundError:
-                        # The entry may disappear between listdir and unlink if cleanup races us.
-                        pass
            shutil.copytree(
                source_path,
                process_host_workspace,
@@ -1,7 +1,7 @@
 from __future__ import annotations

+import base64
 import json
-import mimetypes
 import os

 import langbot_plugin.api.entities.builtin.resource.tool as resource_tool
@@ -230,6 +230,7 @@ class NativeToolLoader(loader.ToolLoader):

    async def _read_workspace_via_box(self, path: str, parameters: dict, query: pipeline_query.Query) -> dict:
        offset = self._positive_int(parameters.get('offset'), default=1)
+        byte_offset = self._non_negative_int(parameters.get('byte_offset'), default=0)
        max_lines = self._positive_int(
            parameters.get('limit'),
            default=_DEFAULT_READ_MAX_LINES,
@@ -241,12 +242,15 @@ class NativeToolLoader(loader.ToolLoader):
            self._positive_int(parameters.get('max_bytes'), default=_DEFAULT_TOOL_RESULT_MAX_BYTES),
            _BOX_FILE_SCRIPT_MAX_BYTES,
        )
+        encoding = self._read_encoding(parameters)
        script = f"""
-import json, os
+import base64, json, os
 path = {json.dumps(path)}
 offset = {offset}
+byte_offset = {byte_offset}
 max_lines = {max_lines}
 max_bytes = {max_bytes}
+encoding = {json.dumps(encoding)}
 if not path.startswith('/workspace'):
    print(json.dumps({{'ok': False, 'error': 'Path must be under /workspace.'}}))
 elif not os.path.exists(path):
@@ -255,6 +259,24 @@ elif os.path.isdir(path):
    entries = sorted(os.listdir(path))
    content = '\\n'.join(entries)
    print(json.dumps({{'ok': True, 'content': content, 'is_directory': True, 'total': len(entries), 'truncated': False}}))
+elif encoding == 'base64':
+    size_bytes = os.path.getsize(path)
+    with open(path, 'rb') as f:
+        f.seek(byte_offset)
+        data = f.read(max_bytes + 1)
+    chunk = data[:max_bytes]
+    has_more = len(data) > max_bytes
+    print(json.dumps({{
+        'ok': True,
+        'content': base64.b64encode(chunk).decode('ascii'),
+        'encoding': 'base64',
+        'byte_offset': byte_offset,
+        'length': len(chunk),
+        'size_bytes': size_bytes,
+        'has_more': has_more,
+        'next_byte_offset': byte_offset + len(chunk) if has_more else None,
+        'max_bytes': max_bytes,
+    }}))
 else:
    lines = []
    output_bytes = 0
@@ -290,18 +312,37 @@ else:
 """.strip()
        return await self._run_workspace_file_script(script, query)

-    async def _write_workspace_via_box(self, path: str, content: str, query: pipeline_query.Query) -> dict:
+    async def _write_workspace_via_box(
+        self,
+        path: str,
+        content: str,
+        parameters: dict,
+        query: pipeline_query.Query,
+    ) -> dict:
+        encoding, mode = self._write_options(parameters)
        script = f"""
-import json, os
+import base64, json, os
 path = {json.dumps(path)}
 content = {json.dumps(content)}
+encoding = {json.dumps(encoding)}
+mode = {json.dumps(mode)}
 if not path.startswith('/workspace'):
    print(json.dumps({{'ok': False, 'error': 'Path must be under /workspace.'}}))
 else:
    os.makedirs(os.path.dirname(path) or '/workspace', exist_ok=True)
-    with open(path, 'w', encoding='utf-8') as f:
-        f.write(content)
-    print(json.dumps({{'ok': True, 'path': path}}))
+    if encoding == 'base64':
+        try:
+            data = base64.b64decode(content, validate=True)
+        except Exception as exc:
+            print(json.dumps({{'ok': False, 'error': f'invalid base64 content: {{exc}}'}}))
+        else:
+            with open(path, 'ab' if mode == 'append' else 'wb') as f:
+                f.write(data)
+            print(json.dumps({{'ok': True, 'path': path}}))
+    else:
+        with open(path, 'a' if mode == 'append' else 'w', encoding='utf-8') as f:
+            f.write(content)
+        print(json.dumps({{'ok': True, 'path': path}}))
 """.strip()
        return await self._run_workspace_file_script(script, query)

@@ -478,9 +519,7 @@ else:
            if host_path and os.path.exists(host_path):
                if os.path.isdir(host_path):
                    return self._build_directory_result(os.listdir(host_path))
-                result = self._read_text_file_preview(host_path, parameters)
-                host_root = str(selected_skill.get('package_root', '') or '')
-                return await self._attach_file_artifact_ref(result, host_path, host_root, path, query)
+                return self._read_text_file_preview(host_path, parameters)

            try:
                result = await self.ap.box_service.read_skill_file(selected_skill['name'], relative)
@@ -506,14 +545,13 @@ else:
        if os.path.isdir(host_path):
            entries = os.listdir(host_path)
            return self._build_directory_result(entries)
-        result = self._read_text_file_preview(host_path, parameters)
-        host_root = self._get_host_root(selected_skill)
-        return await self._attach_file_artifact_ref(result, host_path, host_root, path, query)
+        return self._read_text_file_preview(host_path, parameters)

    async def _invoke_write(self, parameters: dict, query: pipeline_query.Query) -> dict:
        path = parameters['path']
        content = parameters['content']
        self.ap.logger.info(f'write tool invoked: query_id={query.query_id} path={path} length={len(content)}')
+        encoding, _mode = self._write_options(parameters)
        skill_request = self._resolve_skill_relative_path(
            query,
            path,
@@ -521,6 +559,8 @@ else:
            include_activated=True,
        )
        if skill_request is not None and hasattr(self.ap.box_service, 'write_skill_file'):
+            if encoding != 'text':
+                return {'ok': False, 'error': 'base64 writes to skill packages are not supported.'}
            selected_skill, relative = skill_request
            await self.ap.box_service.write_skill_file(selected_skill['name'], relative, content)
            await self.ap.skill_mgr.reload_skills()
@@ -533,10 +573,12 @@ else:
            include_activated=True,
        )
        if self._should_use_box_workspace_files(selected_skill):
-            return await self._write_workspace_via_box(path, content, query)
+            return await self._write_workspace_via_box(path, content, parameters, query)
        os.makedirs(os.path.dirname(host_path), exist_ok=True)
-        with open(host_path, 'w', encoding='utf-8') as f:
-            f.write(content)
+        try:
+            self._write_host_file(host_path, content, parameters)
+        except ValueError as exc:
+            return {'ok': False, 'error': str(exc)}
        self._refresh_skill_from_disk(selected_skill)
        return {'ok': True, 'path': path}

@@ -695,13 +737,24 @@ else:
                    'max_bytes': {
                        'type': 'integer',
                        'description': (
-                            'Maximum bytes of file content to return. '
-                            f'Defaults to {_DEFAULT_TOOL_RESULT_MAX_BYTES}.'
+                            f'Maximum bytes of file content to return. Defaults to {_DEFAULT_TOOL_RESULT_MAX_BYTES}.'
                        ),
                        'default': _DEFAULT_TOOL_RESULT_MAX_BYTES,
                        'minimum': 1,
                        'maximum': _DEFAULT_TOOL_RESULT_MAX_BYTES,
                    },
+                    'encoding': {
+                        'type': 'string',
+                        'description': 'Return text by default, or base64 for binary byte-range reads.',
+                        'enum': ['text', 'base64'],
+                        'default': 'text',
+                    },
+                    'byte_offset': {
+                        'type': 'integer',
+                        'description': '0-indexed byte offset used when encoding is base64. Defaults to 0.',
+                        'default': 0,
+                        'minimum': 0,
+                    },
                },
                'required': ['path'],
                'additionalProperties': False,
@@ -727,7 +780,19 @@ else:
                    },
                    'content': {
                        'type': 'string',
-                        'description': 'Content to write to the file.',
+                        'description': 'Text content, or base64 content when encoding is base64.',
+                    },
+                    'encoding': {
+                        'type': 'string',
+                        'description': 'Write content as text by default, or decode it from base64 for binary files.',
+                        'enum': ['text', 'base64'],
+                        'default': 'text',
+                    },
+                    'mode': {
+                        'type': 'string',
+                        'description': 'Overwrite the file by default, or append to it.',
+                        'enum': ['overwrite', 'append'],
+                        'default': 'overwrite',
                    },
                },
                'required': ['path', 'content'],
@@ -984,84 +1049,6 @@ else:
            raise ValueError('Path escapes the skill package boundary.')
        return host_path

-    def _get_host_root(self, selected_skill: dict | None) -> str:
-        if selected_skill is not None:
-            return str(selected_skill.get('package_root', '') or '')
-        return str(getattr(self.ap.box_service, 'default_workspace', '') or '')
-
-    async def _attach_file_artifact_ref(
-        self,
-        result: dict,
-        host_path: str,
-        host_root: str,
-        sandbox_path: str,
-        query: pipeline_query.Query,
-    ) -> dict:
-        if not result.get('ok') or not result.get('truncated') or result.get('artifact_refs'):
-            return result
-        if not host_root or not os.path.isfile(host_path):
-            return result
-
-        run_session = self._get_agent_run_session(query)
-        if not run_session:
-            return result
-
-        persistence_mgr = getattr(self.ap, 'persistence_mgr', None)
-        get_db_engine = getattr(persistence_mgr, 'get_db_engine', None)
-        if not callable(get_db_engine):
-            return result
-
-        try:
-            from langbot.pkg.agent.runner.artifact_store import ArtifactStore
-
-            authorization = run_session.get('authorization', {}) if isinstance(run_session, dict) else {}
-            mime_type = mimetypes.guess_type(host_path)[0] or 'text/plain'
-            size_bytes = os.path.getsize(host_path)
-            metadata = {
-                'tool_name': READ_TOOL_NAME,
-                'sandbox_path': sandbox_path,
-                'truncated_by': result.get('truncated_by'),
-                'start_line': result.get('start_line'),
-                'end_line': result.get('end_line'),
-                'next_offset': result.get('next_offset'),
-            }
-            artifact_id = await ArtifactStore(get_db_engine()).register_file_artifact(
-                artifact_id=None,
-                host_path=host_path,
-                host_root=host_root,
-                artifact_type='file',
-                source='tool',
-                mime_type=mime_type,
-                name=os.path.basename(host_path),
-                size_bytes=size_bytes,
-                conversation_id=authorization.get('conversation_id'),
-                run_id=run_session.get('run_id') if isinstance(run_session, dict) else None,
-                runner_id=run_session.get('runner_id') if isinstance(run_session, dict) else None,
-                bot_id=getattr(query, 'bot_uuid', None),
-                workspace_id=authorization.get('workspace_id'),
-                thread_id=authorization.get('thread_id'),
-                metadata=metadata,
-            )
-            artifact_ref = {
-                'artifact_id': artifact_id,
-                'artifact_type': 'file',
-                'mime_type': mime_type,
-                'name': os.path.basename(host_path),
-                'size_bytes': size_bytes,
-            }
-            enriched = dict(result)
-            enriched['preview'] = str(result.get('content') or '')
-            enriched['artifact_refs'] = [artifact_ref]
-            return enriched
-        except Exception as exc:
-            self.ap.logger.warning(f'Failed to register read artifact for {sandbox_path}: {exc}')
-            return result
-
-    @staticmethod
-    def _get_agent_run_session(query: pipeline_query.Query) -> dict | None:
-        session = getattr(query, '_agent_run_session', None)
-        return session if isinstance(session, dict) else None
-
    def _normalize_exec_result(self, result: dict) -> dict:
        normalized = dict(result)
        stdout = str(normalized.get('stdout') or '')
@@ -1101,6 +1088,9 @@ else:
        }

    def _read_text_file_preview(self, host_path: str, parameters: dict) -> dict:
+        if self._read_encoding(parameters) == 'base64':
+            return self._read_binary_file_chunk(host_path, parameters)
+
        offset = self._positive_int(parameters.get('offset'), default=1)
        max_lines = self._positive_int(
            parameters.get('limit'),
@@ -1160,6 +1150,54 @@ else:
            'max_bytes': max_bytes,
        }

+    def _read_binary_file_chunk(self, host_path: str, parameters: dict) -> dict:
+        byte_offset = self._non_negative_int(parameters.get('byte_offset'), default=0)
+        max_bytes = self._positive_int(
+            parameters.get('max_bytes'),
+            default=_DEFAULT_TOOL_RESULT_MAX_BYTES,
+            max_value=_DEFAULT_TOOL_RESULT_MAX_BYTES,
+        )
+        size_bytes = os.path.getsize(host_path)
+        with open(host_path, 'rb') as f:
+            f.seek(byte_offset)
+            data = f.read(max_bytes + 1)
+        chunk = data[:max_bytes]
+        has_more = len(data) > max_bytes
+        return {
+            'ok': True,
+            'content': base64.b64encode(chunk).decode('ascii'),
+            'encoding': 'base64',
+            'byte_offset': byte_offset,
+            'length': len(chunk),
+            'size_bytes': size_bytes,
+            'has_more': has_more,
+            'next_byte_offset': byte_offset + len(chunk) if has_more else None,
+            'max_bytes': max_bytes,
+        }
+
+    def _write_host_file(self, host_path: str, content: str, parameters: dict) -> None:
+        encoding, mode = self._write_options(parameters)
+        if encoding == 'base64':
+            try:
+                data = base64.b64decode(content, validate=True)
+            except Exception as exc:
+                raise ValueError(f'invalid base64 content: {exc}') from exc
+            with open(host_path, 'ab' if mode == 'append' else 'wb') as f:
+                f.write(data)
+            return
+        with open(host_path, 'a' if mode == 'append' else 'w', encoding='utf-8') as f:
+            f.write(content)
+
+    @staticmethod
+    def _read_encoding(parameters: dict) -> str:
+        return 'base64' if parameters.get('encoding') == 'base64' else 'text'
+
+    @staticmethod
+    def _write_options(parameters: dict) -> tuple[str, str]:
+        encoding = 'base64' if parameters.get('encoding') == 'base64' else 'text'
+        mode = 'append' if parameters.get('mode') == 'append' else 'overwrite'
+        return encoding, mode
+
    def _build_read_result_from_text(self, content: str, parameters: dict) -> dict:
        offset = self._positive_int(parameters.get('offset'), default=1)
        max_lines = self._positive_int(
@@ -1221,6 +1259,14 @@ else:
            parsed = min(parsed, max_value)
        return parsed

+    @staticmethod
+    def _non_negative_int(value, *, default: int) -> int:
+        try:
+            parsed = int(value)
+        except (TypeError, ValueError):
+            parsed = default
+        return parsed if parsed >= 0 else default
+
    @staticmethod
    def _truncate_grep_line(line: str) -> tuple[str, bool]:
        if len(line) <= _GREP_MAX_LINE_CHARS:
@@ -10,7 +10,6 @@ if typing.TYPE_CHECKING:
    from langbot_plugin.api.entities.events import pipeline_query

 ACTIVATED_SKILLS_KEY = '_activated_skills'
-ACTIVATED_SKILL_NAMES_STATE_KEY = 'host.activated_skills'
 PIPELINE_BOUND_SKILLS_KEY = '_pipeline_bound_skills'
 SKILL_MOUNT_PREFIX = '/workspace/.skills'
 _SKILL_MOUNT_PATTERN = re.compile(r'/workspace/\.skills/([A-Za-z0-9_-]+)')
@@ -73,7 +72,8 @@ def register_activated_skill(query: pipeline_query.Query, skill_data: dict) -> N
        activated[skill_name] = skill_data


-def _normalize_skill_names(value: typing.Any) -> list[str]:
+def normalize_skill_names(value: typing.Any) -> list[str]:
+    """Return a de-duplicated list of non-empty skill names."""
    if not isinstance(value, list):
        return []

@@ -85,21 +85,24 @@ def _normalize_skill_names(value: typing.Any) -> list[str]:
    return names


-def restore_activated_skills_from_state(
+def get_activated_skill_names(query: pipeline_query.Query) -> list[str]:
+    """Return activated skill names for callers that own persistence policy."""
+    return normalize_skill_names(list(get_activated_skills(query).keys()))
+
+
+def restore_activated_skills(
    ap: app.Application,
    query: pipeline_query.Query,
-    state: dict[str, dict[str, typing.Any]],
+    skill_names: typing.Any,
 ) -> list[str]:
-    """Restore persisted activated skill names into Query variables.
+    """Restore caller-provided activated skill names into Query variables.

-    The state value stores names only. Full skill metadata is rebuilt from the
-    current pipeline-visible skill cache so removed or unbound skills remain
-    unavailable to native exec/write/edit.
+    Persistence and state scope ownership belong to higher-level flows. This
+    helper only rebuilds current Query state from pipeline-visible skills, so
+    removed or unbound skills stay unavailable to native exec/write/edit.
    """
-    conversation_state = state.get('conversation', {}) if isinstance(state, dict) else {}
-    skill_names = _normalize_skill_names(conversation_state.get(ACTIVATED_SKILL_NAMES_STATE_KEY))
    restored: list[str] = []
-    for skill_name in skill_names:
+    for skill_name in normalize_skill_names(skill_names):
        skill_data = get_visible_skill(ap, query, skill_name)
        if skill_data is None:
            continue
@@ -108,81 +111,6 @@ def restore_activated_skills_from_state(
    return restored


-def _get_agent_run_authorization(query: pipeline_query.Query) -> dict[str, typing.Any] | None:
-    session = getattr(query, '_agent_run_session', None)
-    if not isinstance(session, dict):
-        return None
-    authorization = session.get('authorization')
-    return authorization if isinstance(authorization, dict) else None
-
-
-def _get_conversation_state_target(query: pipeline_query.Query) -> tuple[str, str, str, dict[str, typing.Any]] | None:
-    session = getattr(query, '_agent_run_session', None)
-    if not isinstance(session, dict):
-        return None
-
-    authorization = _get_agent_run_authorization(query)
-    if authorization is None:
-        return None
-
-    state_policy = authorization.get('state_policy') or {}
-    if not state_policy.get('enable_state', True):
-        return None
-    state_scopes = state_policy.get('state_scopes', ['conversation', 'actor'])
-    if 'conversation' not in state_scopes:
-        return None
-
-    state_context = authorization.get('state_context') or {}
-    scope_keys = state_context.get('scope_keys') or {}
-    scope_key = scope_keys.get('conversation')
-    if not scope_key:
-        return None
-
-    runner_id = str(session.get('runner_id') or 'unknown')
-    binding_identity = str(state_context.get('binding_identity') or 'unknown')
-    return scope_key, runner_id, binding_identity, state_context
-
-
-async def persist_activated_skill(ap: app.Application, query: pipeline_query.Query, skill_name: str) -> bool:
-    """Persist activated skill names for the current AgentRunner conversation.
-
-    Returns False when the call is outside an AgentRunner run or state policy
-    does not expose a conversation scope. The in-memory Query activation still
-    remains valid for the current turn.
-    """
-    target = _get_conversation_state_target(query)
-    if target is None:
-        return False
-
-    persistence_mgr = getattr(ap, 'persistence_mgr', None)
-    if persistence_mgr is None or not hasattr(persistence_mgr, 'get_db_engine'):
-        return False
-
-    from ....agent.runner.persistent_state_store import get_persistent_state_store
-
-    scope_key, runner_id, binding_identity, state_context = target
-    store = get_persistent_state_store(persistence_mgr.get_db_engine())
-    existing_names = _normalize_skill_names(await store.state_get(scope_key, ACTIVATED_SKILL_NAMES_STATE_KEY))
-    if skill_name not in existing_names:
-        existing_names.append(skill_name)
-
-    success, error = await store.state_set(
-        scope_key=scope_key,
-        state_key=ACTIVATED_SKILL_NAMES_STATE_KEY,
-        value=existing_names,
-        runner_id=runner_id,
-        binding_identity=binding_identity,
-        scope='conversation',
-        context=state_context,
-        logger=getattr(ap, 'logger', None),
-    )
-    if not success:
-        logger = getattr(ap, 'logger', None)
-        if logger is not None:
-            logger.warning(f'Failed to persist activated skill "{skill_name}": {error}')
-    return success
-
-
 def parse_skill_mount_path(sandbox_path: str) -> tuple[str | None, str]:
    normalized_path = str(sandbox_path or '/workspace').strip() or '/workspace'
    if normalized_path == SKILL_MOUNT_PREFIX: