fix(channel): assign session ownership and restrict ownerless access

flocks/channel/inbound/dispatcher.py

@@ -847,19 +847,24 @@ async def _handle_session_command(
         scope_override: Optional[str],
     ) -> None:
         from flocks.session.session import Session
-        from flocks.channel.inbound.session_binding import _build_title
+        from flocks.channel.inbound.session_binding import (
+            _build_title,
+            resolve_channel_session_owner_kwargs,
+        )
 
         session = await Session.get_by_id(binding.session_id)
         if not session:
             await callbacks.deliver_text("当前会话不存在，请发送一条普通消息后重试。")
             return
 
+        owner_kwargs = await resolve_channel_session_owner_kwargs(session)
         new_session = await Session.create(
             project_id=session.project_id,
             directory=session.directory,
             title=_build_title(msg),
             agent=session.agent,
             **Session.inherited_model_kwargs(session),
+            **owner_kwargs,
         )
         new_binding = await self.binding_service.rebind(
             msg,

flocks/channel/inbound/session_binding.py

@@ -78,6 +78,46 @@ class SessionBinding:
 # corruption vector.
 _db_owner_pid: Optional[int] = None
 
+
+async def resolve_channel_session_owner_kwargs(source_session=None) -> dict[str, str]:
+    """Return ownership kwargs for a channel-created session.
+
+    Channel dispatch runs outside the HTTP auth middleware, so
+    ``Session.create`` cannot infer the owner from ``current_auth_user``.
+    When an existing channel session is being replaced, preserve its owner.
+    Otherwise, attach new channel sessions to the local admin if one exists.
+    Installs without local accounts remain ownerless for backward-compatible
+    no-login operation.
+    """
+    owner_user_id = getattr(source_session, "owner_user_id", None) if source_session else None
+    owner_username = getattr(source_session, "owner_username", None) if source_session else None
+    if owner_user_id or owner_username:
+        owner_kwargs: dict[str, str] = {}
+        if owner_user_id:
+            owner_kwargs["owner_user_id"] = str(owner_user_id)
+        if owner_username:
+            owner_kwargs["owner_username"] = str(owner_username)
+        return owner_kwargs
+
+    try:
+        from flocks.auth.service import AuthService
+
+        if not await AuthService.has_users():
+            return {}
+        users = await AuthService.list_users()
+    except Exception as exc:
+        log.warn("channel.owner.resolve_failed", {"error": str(exc)})
+        return {}
+
+    admin = next((user for user in users if getattr(user, "role", None) == "admin"), None)
+    if admin is None:
+        return {}
+    return {
+        "owner_user_id": str(admin.id),
+        "owner_username": str(admin.username),
+    }
+
+
 # Register channel_bindings DDL with Storage so the tables are created
 # during Storage.init() as well (idempotent CREATE IF NOT EXISTS).
 try:
@@ -488,11 +528,13 @@ async def _create_session(
         from flocks.session.session import Session
 
         title = _build_title(msg)
+        owner_kwargs = await resolve_channel_session_owner_kwargs()
         session = await Session.create(
             project_id="channel",
             directory=_resolve_session_directory(directory),
             title=title,
             agent=default_agent,
+            **owner_kwargs,
         )
         return session.id
 

flocks/server/routes/session.py

@@ -132,7 +132,7 @@ def _session_to_response(session: SessionModel) -> SessionResponse:
     current_user = get_current_auth_user()
     can_write = SessionPolicy.can_write(session, current_user)
     can_delete = SessionPolicy.can_delete(session, current_user)
-    is_shared = SessionPolicy.is_local_shared(session)
+    is_shared = SessionPolicy.is_shared(session)
 
     return SessionResponse(
         id=session.id,

flocks/session/policy.py

@@ -51,6 +51,19 @@ def is_local_shared(session: "SessionInfo") -> bool:
             return False
         return bool(metadata.get("shared_local"))
 
+    @staticmethod
+    def _has_no_owner(session: "SessionInfo") -> bool:
+        return not session.owner_user_id and not session.owner_username
+
+    @classmethod
+    def is_shared(cls, session: "SessionInfo") -> bool:
+        """Whether the session is explicitly shared to all local users.
+
+        Ownerless sessions are a legacy / unauthenticated compatibility state,
+        not a sharing state. The UI badge should only reflect explicit sharing.
+        """
+        return cls.is_local_shared(session)
+
     @staticmethod
     def _shared_read_user_ids(session: "SessionInfo") -> set[str]:
         metadata = getattr(session, "metadata", None)
@@ -77,30 +90,42 @@ def can_read(cls, session: "SessionInfo", user: Optional["AuthUser"] = None) ->
         Whether the session should be visible in listings / fetch.
 
         - No auth context (CLI/internal runtime): keep legacy permissive behaviour.
-        - Logged-in users: owner or local-shared readers.
+        - Logged-in users: owner, local-shared readers, shared readers, or admins
+          managing ownerless legacy/channel sessions.
         """
         resolved = cls._resolve_user(user)
         if resolved is None:
             return True
         if cls.is_owner(session, resolved):
             return True
+        if cls._has_no_owner(session) and cls.is_admin(resolved):
+            return True
         return cls.is_shared_read_only(session, resolved)
 
     @classmethod
     def can_write(cls, session: "SessionInfo", user: Optional["AuthUser"] = None) -> bool:
         """
         Session write permission.
 
-        Shared users are read-only. Only owner can write.
+        Owner can always write. Admins may repair/manage ownerless sessions
+        accumulated before local ownership was available.
         """
         resolved = cls._resolve_user(user)
         if resolved is None:
             return False
-        return cls.is_owner(session, resolved)
+        if cls.is_owner(session, resolved):
+            return True
+        if cls._has_no_owner(session) and cls.is_admin(resolved):
+            return True
+        return False
 
     @classmethod
     def can_delete(cls, session: "SessionInfo", user: Optional["AuthUser"]) -> bool:
         resolved = cls._resolve_user(user)
         if resolved is None:
             return False
-        return cls.is_owner(session, resolved)
+        if cls.is_owner(session, resolved):
+            return True
+        if cls._has_no_owner(session) and cls.is_admin(resolved):
+            return True
+        return False

tests/channel/test_unified_prompt_context.py

@@ -37,6 +37,7 @@
 from flocks.channel.inbound.session_binding import (
     SessionBindingService,
     _resolve_session_directory,
+    resolve_channel_session_owner_kwargs,
 )
 from flocks.config.config import ChannelConfig
 
@@ -228,6 +229,75 @@ async def _fake_create(**kwargs):
         assert captured["directory"] == "/instance/dir"
 
 
+class TestChannelSessionOwnerPropagation:
+    @pytest.mark.asyncio
+    async def test_create_session_assigns_local_admin_owner(self):
+        captured = {}
+
+        class _StubSession:
+            id = "ses_admin_owned"
+
+        async def _fake_create(**kwargs):
+            captured.update(kwargs)
+            return _StubSession()
+
+        admin = SimpleNamespace(id="usr_admin", username="admin", role="admin")
+
+        with patch("flocks.session.session.Session.create", new=_fake_create), \
+             patch("flocks.auth.service.AuthService.has_users", new=AsyncMock(return_value=True)), \
+             patch("flocks.auth.service.AuthService.list_users", new=AsyncMock(return_value=[admin])):
+            sid = await SessionBindingService._create_session(
+                _msg(),
+                default_agent="rex",
+                directory="/explicit/dir",
+            )
+
+        assert sid == "ses_admin_owned"
+        assert captured["owner_user_id"] == "usr_admin"
+        assert captured["owner_username"] == "admin"
+
+    @pytest.mark.asyncio
+    async def test_create_session_stays_ownerless_without_local_accounts(self):
+        captured = {}
+
+        class _StubSession:
+            id = "ses_ownerless"
+
+        async def _fake_create(**kwargs):
+            captured.update(kwargs)
+            return _StubSession()
+
+        with patch("flocks.session.session.Session.create", new=_fake_create), \
+             patch("flocks.auth.service.AuthService.has_users", new=AsyncMock(return_value=False)), \
+             patch("flocks.auth.service.AuthService.list_users", new=AsyncMock()) as list_users:
+            sid = await SessionBindingService._create_session(
+                _msg(),
+                default_agent="rex",
+                directory="/explicit/dir",
+            )
+
+        assert sid == "ses_ownerless"
+        assert "owner_user_id" not in captured
+        assert "owner_username" not in captured
+        list_users.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_owner_kwargs_preserve_existing_session_owner(self):
+        existing = SimpleNamespace(
+            owner_user_id="usr_existing",
+            owner_username="existing",
+        )
+
+        with patch("flocks.auth.service.AuthService.has_users", new=AsyncMock()) as has_users:
+            owner_kwargs = await resolve_channel_session_owner_kwargs(existing)
+
+        assert owner_kwargs == {
+            "owner_user_id": "usr_existing",
+            "owner_username": "existing",
+        }
+        has_users.assert_not_awaited()
+
+
 # ---------------------------------------------------------------------------
 # 2. default_agent unification
 # ---------------------------------------------------------------------------

tests/session/test_session_policy.py

@@ -75,3 +75,22 @@ def test_can_read_local_shared_visible_to_all_local_users():
     assert SessionPolicy.can_read(session, owner) is True
     assert SessionPolicy.can_read(session, admin) is True
     assert SessionPolicy.can_read(session, stranger) is True
+
+
+def test_ownerless_session_is_not_marked_shared():
+    session = _make_session(owner_user_id=None, owner_username=None)
+    assert SessionPolicy.is_shared(session) is False
+
+
+def test_ownerless_session_admin_can_manage_but_member_cannot():
+    admin = _make_user(user_id="usr_admin", username="admin", role="admin")
+    member = _make_user(user_id="usr_member", username="member", role="member")
+    session = _make_session(owner_user_id=None, owner_username=None)
+
+    assert SessionPolicy.can_read(session, admin) is True
+    assert SessionPolicy.can_write(session, admin) is True
+    assert SessionPolicy.can_delete(session, admin) is True
+
+    assert SessionPolicy.can_read(session, member) is False
+    assert SessionPolicy.can_write(session, member) is False
+    assert SessionPolicy.can_delete(session, member) is False

tests/user_workflow/.gitignore

@@ -0,0 +1,4 @@
+# 本地用户自定义 workflow 测试目录
+# 默认忽略目录内所有内容，仅保留当前 .gitignore 以便目录可被提交。
+*
+!.gitignore