Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 33 additions & 15 deletions scripts/sync_agent_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
同步 openclaw.json 中的 agent 配置 → data/agent_config.json
支持自动发现 agent workspace 下的 Skills 目录
"""
import json, pathlib, datetime, logging
import json, os, pathlib, datetime, logging
from file_lock import atomic_json_write

log = logging.getLogger('sync_agent_config')
Expand Down Expand Up @@ -210,8 +210,35 @@ def main():
'zaochao': 'zaochao',
}

def _sync_script_symlink(src_file: pathlib.Path, dst_file: pathlib.Path) -> bool:
"""Create a symlink dst_file → src_file (resolved).

Using symlinks instead of physical copies ensures that ``__file__`` in
each script always resolves back to the project ``scripts/`` directory,
so relative-path computations like ``Path(__file__).resolve().parent.parent``
point to the correct project root regardless of which workspace runs the
script. (Fixes #56 — kanban data-path split)

Returns True if the link was (re-)created, False if already up-to-date.
"""
src_resolved = src_file.resolve()
# Already a correct symlink?
if dst_file.is_symlink() and dst_file.resolve() == src_resolved:
return False
# Remove stale file / old physical copy / broken symlink
if dst_file.exists() or dst_file.is_symlink():
dst_file.unlink()
os.symlink(src_resolved, dst_file)
return True


def sync_scripts_to_workspaces():
"""将项目 scripts/ 目录同步到各 agent workspace(保持 kanban_update.py 等最新)"""
"""将项目 scripts/ 目录同步到各 agent workspace(保持 kanban_update.py 等最新)

Uses symlinks so that ``__file__`` in workspace copies resolves to the
project ``scripts/`` directory, keeping path-derived constants like
``TASKS_FILE`` pointing to the canonical ``data/`` folder.
"""
scripts_src = BASE / 'scripts'
if not scripts_src.is_dir():
return
Expand All @@ -224,16 +251,10 @@ def sync_scripts_to_workspaces():
continue
dst_file = ws_scripts / src_file.name
try:
src_text = src_file.read_bytes()
if _sync_script_symlink(src_file, dst_file):
synced += 1
except Exception:
continue
try:
dst_text = dst_file.read_bytes() if dst_file.exists() else b''
except Exception:
dst_text = b''
if src_text != dst_text:
dst_file.write_bytes(src_text)
synced += 1
# also sync to workspace-main for legacy compatibility
ws_main_scripts = pathlib.Path.home() / '.openclaw/workspace-main/scripts'
ws_main_scripts.mkdir(parents=True, exist_ok=True)
Expand All @@ -242,15 +263,12 @@ def sync_scripts_to_workspaces():
continue
dst_file = ws_main_scripts / src_file.name
try:
src_text = src_file.read_bytes()
dst_text = dst_file.read_bytes() if dst_file.exists() else b''
if src_text != dst_text:
dst_file.write_bytes(src_text)
if _sync_script_symlink(src_file, dst_file):
synced += 1
except Exception:
pass
if synced:
log.info(f'{synced} script files synced to workspaces')
log.info(f'{synced} script symlinks synced to workspaces')


def deploy_soul_files():
Expand Down
171 changes: 171 additions & 0 deletions tests/test_sync_symlinks.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,171 @@
"""Tests for symlink-based script sync (fix for issue #56).

Verifies that ``sync_scripts_to_workspaces()`` creates symlinks instead of
physical copies, so ``__file__`` in workspace scripts resolves back to the
project ``scripts/`` directory.
"""
import os
import pathlib
import sys
import types

import pytest

# ── Bootstrap: make scripts/ importable ──────────────────────────
SCRIPTS = pathlib.Path(__file__).resolve().parent.parent / 'scripts'
sys.path.insert(0, str(SCRIPTS))

import sync_agent_config as sac # noqa: E402


# ────────────────────────────────────────────────────────────────
# Helper: patch BASE, _SOUL_DEPLOY_MAP, HOME to isolate tests
# ────────────────────────────────────────────────────────────────

@pytest.fixture
def project(tmp_path, monkeypatch):
"""Set up a minimal fake project tree and home dir."""
proj = tmp_path / 'project'
scripts = proj / 'scripts'
scripts.mkdir(parents=True)
data = proj / 'data'
data.mkdir()
(data / 'tasks_source.json').write_text('[]')

home = tmp_path / 'home'
home.mkdir()

# Create a couple of dummy scripts
(scripts / 'kanban_update.py').write_text('# kanban\n')
(scripts / 'refresh_live_data.py').write_text('# refresh\n')
(scripts / '__init__.py').write_text('') # should be skipped

# Patch module-level state
monkeypatch.setattr(sac, 'BASE', proj)
monkeypatch.setattr(sac, '_SOUL_DEPLOY_MAP', {'agent-a': 'aaa'})
monkeypatch.setattr(pathlib.Path, 'home', staticmethod(lambda: home))

return types.SimpleNamespace(root=proj, scripts=scripts, data=data, home=home)


# ── Tests ────────────────────────────────────────────────────────

class TestSyncScriptSymlink:
"""Unit tests for the helper ``_sync_script_symlink``."""

def test_creates_symlink(self, tmp_path):
src = tmp_path / 'src.py'
src.write_text('hello')
dst = tmp_path / 'dst.py'

created = sac._sync_script_symlink(src, dst)

assert created is True
assert dst.is_symlink()
assert dst.resolve() == src.resolve()
assert dst.read_text() == 'hello'

def test_idempotent_when_up_to_date(self, tmp_path):
src = tmp_path / 'src.py'
src.write_text('hello')
dst = tmp_path / 'dst.py'

sac._sync_script_symlink(src, dst)
created = sac._sync_script_symlink(src, dst)

assert created is False # already correct

def test_replaces_physical_copy(self, tmp_path):
"""A pre-existing physical copy should be replaced with a symlink."""
src = tmp_path / 'src.py'
src.write_text('v2')
dst = tmp_path / 'dst.py'
dst.write_text('v1') # physical copy, not a symlink

created = sac._sync_script_symlink(src, dst)

assert created is True
assert dst.is_symlink()
assert dst.resolve() == src.resolve()

def test_replaces_broken_symlink(self, tmp_path):
src = tmp_path / 'src.py'
src.write_text('ok')
dst = tmp_path / 'dst.py'
os.symlink('/no/such/file', dst) # broken link

created = sac._sync_script_symlink(src, dst)

assert created is True
assert dst.is_symlink()
assert dst.resolve() == src.resolve()


class TestSyncScriptsToWorkspaces:
"""Integration tests for the full ``sync_scripts_to_workspaces`` flow."""

def test_creates_symlinks_in_workspace(self, project):
sac.sync_scripts_to_workspaces()

ws = project.home / '.openclaw' / 'workspace-aaa' / 'scripts'
assert ws.is_dir()

kb = ws / 'kanban_update.py'
assert kb.is_symlink(), 'expected symlink, got physical file'
assert kb.resolve() == (project.scripts / 'kanban_update.py').resolve()

rf = ws / 'refresh_live_data.py'
assert rf.is_symlink()

def test_skips_dunder_files(self, project):
sac.sync_scripts_to_workspaces()

ws = project.home / '.openclaw' / 'workspace-aaa' / 'scripts'
assert not (ws / '__init__.py').exists()

def test_legacy_workspace_main(self, project):
sac.sync_scripts_to_workspaces()

ws_main = project.home / '.openclaw' / 'workspace-main' / 'scripts'
assert ws_main.is_dir()

kb = ws_main / 'kanban_update.py'
assert kb.is_symlink()
assert kb.resolve() == (project.scripts / 'kanban_update.py').resolve()

def test_idempotent_rerun(self, project):
sac.sync_scripts_to_workspaces()
sac.sync_scripts_to_workspaces() # second run should be a no-op

ws = project.home / '.openclaw' / 'workspace-aaa' / 'scripts'
kb = ws / 'kanban_update.py'
assert kb.is_symlink()

def test_replaces_old_physical_copies(self, project):
"""Simulate pre-existing physical copies (old behaviour) and verify
they get replaced by symlinks on the next sync run."""
ws = project.home / '.openclaw' / 'workspace-aaa' / 'scripts'
ws.mkdir(parents=True, exist_ok=True)
old_copy = ws / 'kanban_update.py'
old_copy.write_text('# stale physical copy')

sac.sync_scripts_to_workspaces()

assert old_copy.is_symlink(), 'old physical copy should be replaced'
assert old_copy.resolve() == (project.scripts / 'kanban_update.py').resolve()

def test_file_resolves_to_project_root(self, project):
"""The whole point of #56: __file__ should resolve to project scripts/,
so Path(__file__).resolve().parent.parent == project root."""
sac.sync_scripts_to_workspaces()

ws = project.home / '.openclaw' / 'workspace-aaa' / 'scripts'
ws_script = ws / 'kanban_update.py'

# Simulate what kanban_update.py does at import time
resolved = ws_script.resolve()
computed_base = resolved.parent.parent
assert computed_base == project.root, (
f'Expected {project.root}, got {computed_base}; '
'symlink should resolve __file__ back to project root'
)
Loading