chore: remove temporary python mcp layer

This commit is contained in:
zhaoyilun
2026-07-05 14:56:41 +08:00
parent 573a09ade3
commit 87e0e47fbf
26 changed files with 122 additions and 6501 deletions

View File

@@ -1,49 +0,0 @@
# ISphereWinHelper First Phase Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [x]`) syntax for tracking.
**Goal:** Build the first C# read-only Windows helper for `version`, `self_check`, `scan_windows`, and `dump_uia` using the agreed Go/C# JSON boundary.
**Architecture:** The helper is a one-shot CLI process. Go will later send one JSON request on stdin and read one JSON response from stdout. This phase implements only read-only desktop/UIA probing and a PowerShell build script.
**Tech Stack:** C# .NET Framework, `csc.exe`, Windows UI Automation, Python `unittest` wrapper tests.
---
### Task 1: Contract and build test
**Files:**
- Create: `tests/test_win_helper.py`
- Create: `scripts/build-win-helper.ps1`
- Create: `native/ISphereWinHelper/Program.cs`
- Create: `native/ISphereWinHelper/HelperProtocol.cs`
- Create: `native/ISphereWinHelper/WindowScanner.cs`
- Create: `native/ISphereWinHelper/UiaDumper.cs`
- [x] Write a Python unittest that expects `scripts/build-win-helper.ps1` to build `runs/win-helper/ISphereWinHelper.exe`.
- [x] Run the test and verify it fails because the build script/helper does not exist yet.
- [x] Add the minimal C# helper and build script.
- [x] Run the build test and verify it passes.
### Task 2: JSON operation tests
**Files:**
- Modify: `tests/test_win_helper.py`
- Modify: `native/ISphereWinHelper/*.cs`
- [x] Add tests for `version`, `self_check`, `scan_windows`, invalid JSON, and unsupported op.
- [x] Run tests and verify they fail for missing behavior.
- [x] Implement JSON request/response handling and the four first-phase operations.
- [x] Run tests and verify they pass.
### Task 3: Verification
**Files:**
- No new production files.
- [x] Run `python -m unittest discover -s tests -p test_win_helper.py -v`.
- [x] Run `python -m compileall -q tests`.
- [x] Run `powershell -NoProfile -ExecutionPolicy Bypass -File scripts\build-win-helper.ps1`.
- [x] Run direct `version`, `scan_windows`, and dedicated WinForms `dump_uia` JSON requests against the helper.

File diff suppressed because it is too large Load Diff

View File

@@ -1,278 +0,0 @@
about_button_down
about_button_hover
about_button_normal
account_hasTask_logo
account_noTask_logo
account_plugin_logo
addfriendtip
addfriendtipsuccend
audiochat_back
audiochat_mic
audiochat_mic_disable
audiochat_mic_hover
audiochat_sound
audiochat_sound_disable
audiochat_sound_hover
audiochat_track
audiochat_track_disable
audiochat_trackbar
audiochat_trackbar_disable
away
blank
btn_group_click
btn_group_hover
btn_group_normal
chat_Discuss
chat_Group
chat_Man_Leave
chat_Man_OffLine
chat_Man_Online
chat_Woman_Leave
chat_Woman_OffLine
chat_Woman_Online
chatroom_member_admin
chatroom_member_owner
ClearHSNew
CopyHSNew
customeffigy_down_down
customeffigy_down_hover
customeffigy_down_normal
customeffigy_up_down
customeffigy_up_hover
customeffigy_up_normal
CutHSNew
file_transfer_forwarding_close
file_transfer_forwarding_success_not_all
file_transfer_receive
file_transfer_send
file_transfer_send_directory
finger_add
FontDialogHSNew
group_Big_Head
groupnotice_back
groupnotice_down
groupnotice_hover
groupnotice_normal
groupnotice_nothing
head_Group_Online_16
head_Info_Big_Man_Leave
head_Info_Big_Man_Offline
head_Info_Big_Woman_Leave
head_Info_Big_Woman_Offline
head_Info_device_leave_56
head_Info_device_offline_56
head_Info_device_online_56
head_Info_Man_Leave
head_Info_Man_Leave_16
head_Info_Man_Offline
head_Info_Man_OffLine_16
head_Info_Man_Offline_60
head_Info_Man_Online_16
head_Info_Man_Online_60
head_Info_Right_Man_Leave
head_Info_Right_Man_Leave_60
head_Info_Right_Man_Offline
head_Info_Right_Woman_Leave
head_Info_Right_Woman_Leave_60
head_Info_Right_Woman_Offline
head_Info_Woman_Leave
head_Info_Woman_Leave_16
head_Info_Woman_Offline
head_Info_Woman_OffLine_16
head_Info_Woman_Offline_60
head_Info_Woman_Online_16
head_Info_Woman_Online_60
head_Man
head_Man_Info
head_Man_Info_Big_Right
head_Man_Info_Right
head_Man_Leave
head_Man_OffLine
head_Woman
head_Woman_Info
head_Woman_Info_Big_Right
head_Woman_Info_Right
head_Woman_Leave
head_Woman_OffLine
headImage_leaveNew
information
InsertPictureHSNew
Menu_Main_About
MsgReadBurn_Recv_Img
MsgReadBurn_Recv_RTF
MsgReadBurn_Sender_Img
MsgReadBurn_Sender_RTF
multi_add_down
multi_add_hover
multi_add_normal
multi_exit_down
multi_exit_hover
multi_exit_normal
multiaudio_video_closed
multiav_sound
multiav_video
multiav_wait_for_join
multivideo_mic_closed
no_img
online
PasteHSNew
pic_favorNew
plugin_add_nor
plugin_add_pre
plugin_appnone
plugin_del_nor
plugin_del_pre
plugin_error
plugin_home
plugin_ios_last_normal
plugin_ios_next_normal
plugin_ios_table
plugin_limitmark
plugin_link_hover_notok
plugin_link_normal_notok
plugin_loading
plugin_order_checked
plugin_order_down_enable
plugin_order_down_normal
plugin_order_uncheck
plugin_order_up_enable
plugin_order_up_normal
plugin_pager_dot_down
plugin_pager_dot_hover
plugin_pager_dot_normal
plugin_pager_left_down
plugin_pager_left_hover
plugin_pager_left_normal
plugin_pager_right_down
plugin_pager_right_hover
plugin_pager_right_normal
plugin_tabclose_down
plugin_tabclose_hover
plugin_tabclose_normal
plugin_tabdrop_down
plugin_tabdrop_hover
plugin_tabdrop_normal
plugin_toolbar_head
readburn_notify
RecallHSNew
receipt_cancel
receipt_success
RoamImg_Back
RoamImg_loading
RoamImg_UnLoad
rosterRemarks_button_down
rosterRemarks_button_hover
rosterRemarks_button_normal
scroll_down_hover
scroll_down_normal
scroll_up_hover
scroll_up_normal
status_away
status_dropdownitem_selected
status_noaction
status_online
status_online_state_down
status_online_state_hover
sys_logo_Discuss_16
sys_logo_Group_16
sys_logo_leave16
sys_logo_normal16
sys_logo_offline16
sys_msg_validation_16
sys_mydevice_16
tool_Button_Down
tool_Button_Hover
tool_Chat_ExpressionNew
tool_File_Default
tree_close_folder
tree_Man_Leave
tree_Man_OffLine
tree_Man_Online
tree_man_online_ico
tree_open_folder
tree_Woman_Leave
tree_Woman_OffLine
tree_Woman_Online
tree_woman_online_ico
videochat_after_down
videochat_after_hover
videochat_after_normal
videochat_back
videochat_before_down
videochat_before_hover
videochat_before_normal
videochat_button_down
videochat_button_hover
videochat_button_normal
videochat_close_down
videochat_close_hover
videochat_close_normal
videochat_max_down
videochat_max_hover
videochat_max_normal
videochat_mic
videochat_mic_disable
videochat_min_down
videochat_min_hover
videochat_min_normal
videochat_sound
videochat_sound_disable
visual_phone_succeed
vncicon
weather_blizzard
weather_blizzard_small
weather_bwealowingsand
weather_bwealowingsand_small
weather_clear_daytime
weather_clear_daytime_small
weather_cloudy_daytime
weather_cloudy_daytime_small
weather_cloudy_night
weather_cloudy_night_small
weather_day
weather_delete
weather_delete_on
weather_drop_down
weather_flyash
weather_flyash_small
weather_fog
weather_fog_small
weather_greatsnow
weather_greatsnow_small
weather_hail
weather_hail_small
weather_halewater
weather_halewater_small
weather_hardrain
weather_hardrain_small
weather_haze
weather_haze_small
weather_icerain
weather_icerain_small
weather_light_rain
weather_light_rain_small
weather_light_snow
weather_light_snow_small
weather_moderaterain
weather_moderaterain_small
weather_panel
weather_sandstorm
weather_sandstorm_small
weather_select
weather_select_on
weather_shower
weather_shower_small
weather_sleet
weather_sleet_small
weather_snowcloud
weather_snowcloud_small
weather_thundershower
weather_thundershower_small
weatuer_optionboxhover
weatuer_snowshower
weatuer_snowshower_small
weatuer_town
workgroup_empty
workgroup_query
workgroup_share_file_clock
workgroup_share_file_loading

View File

@@ -1,20 +0,0 @@
[project]
name = "isphere-ai-bridge"
version = "0.1.0"
description = "Read-only iSphere MCP service for AI digital employees"
requires-python = ">=3.10"
dependencies = ["mcp==1.28.1"]
[project.scripts]
isphere-ai-bridge = "isphere_ai_bridge.cli:main"
isphere-ai-bridge-mcp = "isphere_ai_bridge.server:main"
[build-system]
requires = ["setuptools>=69"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
pythonpath = ["src"]

View File

@@ -1,124 +0,0 @@
param(
[string]$OutputDir = "runs/native-helper",
[string]$Configuration = "Release",
[ValidateSet("x86", "anycpu")]
[string]$Platform = "x86"
)
$ErrorActionPreference = "Stop"
$repo = Resolve-Path -LiteralPath (Join-Path $PSScriptRoot "..")
$source = Join-Path $repo "native/ISphereOfflineLabHelper/Program.cs"
$outDirPath = Join-Path $repo $OutputDir
$outExe = Join-Path $outDirPath "ISphereOfflineLabHelper.exe"
$stubSource = Join-Path $outDirPath "IMPP.Resource.stub.cs"
$stubResources = Join-Path $outDirPath "IMPP.Resource.Properties.Resources.resources"
$stubDll = Join-Path $outDirPath "IMPP.Resource.dll"
if (-not (Test-Path -LiteralPath $source)) {
throw "source not found: $source"
}
$candidates = @(
"$env:WINDIR\Microsoft.NET\Framework64\v4.0.30319\csc.exe",
"$env:WINDIR\Microsoft.NET\Framework\v4.0.30319\csc.exe"
)
$csc = $candidates | Where-Object { Test-Path -LiteralPath $_ } | Select-Object -First 1
if (-not $csc) {
throw ".NET Framework csc.exe v4.0.30319 not found"
}
New-Item -ItemType Directory -Force -Path $outDirPath | Out-Null
Add-Type -AssemblyName System.Drawing
$resourceNameFile = Join-Path $repo "native/ISphereOfflineLabHelper/resource-names.txt"
$resourceNames = New-Object System.Collections.Generic.SortedSet[string]([System.StringComparer]::Ordinal)
if (Test-Path -LiteralPath $resourceNameFile) {
Get-Content -LiteralPath $resourceNameFile | ForEach-Object {
$name = $_.Trim()
if ($name) {
[void]$resourceNames.Add($name)
}
}
}
foreach ($name in @(
"about_logo",
"ipload_2",
"login_backimg",
"sys_Logo_BackImg",
"sys_Logo_16",
"sys_logo_50",
"sys_logo_all",
"sys_logo_offline16",
"sys_small",
"toolstript_search_background",
"window_default_head"
)) {
[void]$resourceNames.Add($name)
}
$resourceWriter = New-Object System.Resources.ResourceWriter($stubResources)
try {
$bitmap = New-Object System.Drawing.Bitmap 1, 1
$bitmap.SetPixel(0, 0, [System.Drawing.Color]::Transparent)
foreach ($name in $resourceNames) {
$resourceWriter.AddResource($name, $bitmap)
}
}
finally {
$resourceWriter.Generate()
$resourceWriter.Close()
}
@"
namespace IMPP.Resource.Properties
{
internal static class Resources
{
}
}
"@ | Set-Content -LiteralPath $stubSource -Encoding UTF8
& $csc `
/nologo `
/target:library `
/platform:$Platform `
/optimize+ `
/out:$stubDll `
/reference:System.dll `
/reference:System.Drawing.dll `
/resource:$stubResources,IMPP.Resource.Properties.Resources.resources `
$stubSource
if ($LASTEXITCODE -ne 0) {
throw "IMPP.Resource stub compilation failed with exit code $LASTEXITCODE"
}
& $csc `
/nologo `
/target:exe `
/platform:$Platform `
/optimize+ `
/warn:0 `
/out:$outExe `
/reference:System.dll `
/reference:System.Core.dll `
/reference:System.Web.Extensions.dll `
/reference:System.Windows.Forms.dll `
/reference:System.Drawing.dll `
$source
if ($LASTEXITCODE -ne 0) {
throw "native helper compilation failed with exit code $LASTEXITCODE"
}
$result = [ordered]@{
ok = $true
configuration = $Configuration
platform = $Platform
csc = $csc
source = $source
output = $outExe
resource_stub = $stubDll
}
$result | ConvertTo-Json -Depth 4 -Compress

View File

@@ -0,0 +1,122 @@
param(
[string]$HelperExe = "runs/win-helper/ISphereWinHelper.exe",
[switch]$SkipBuild
)
$ErrorActionPreference = "Stop"
$repo = Resolve-Path -LiteralPath (Join-Path $PSScriptRoot "..")
$helperPath = Join-Path $repo $HelperExe
function Invoke-HelperJson([hashtable]$Request) {
$json = $Request | ConvertTo-Json -Depth 8 -Compress
$output = $json | & $helperPath --json
if ($LASTEXITCODE -ne 0) {
throw "helper exited with code $LASTEXITCODE"
}
try {
return $output | ConvertFrom-Json
}
catch {
throw "helper output was not JSON: $output"
}
}
if (-not $SkipBuild) {
& powershell -NoProfile -ExecutionPolicy Bypass -File (Join-Path $repo "scripts\build-win-helper.ps1") | Out-Host
if ($LASTEXITCODE -ne 0) {
throw "build-win-helper.ps1 failed with exit code $LASTEXITCODE"
}
}
if (-not (Test-Path -LiteralPath $helperPath)) {
throw "helper not found: $helperPath"
}
$version = Invoke-HelperJson @{
protocol = "isphere.helper.v1"
request_id = "verify-version"
op = "version"
timeout_ms = 5000
args = @{}
}
if (-not $version.ok -or $version.data.helper_name -ne "ISphereWinHelper") {
throw "version check failed: $($version | ConvertTo-Json -Depth 8 -Compress)"
}
$selfCheck = Invoke-HelperJson @{
protocol = "isphere.helper.v1"
request_id = "verify-self-check"
op = "self_check"
timeout_ms = 5000
args = @{}
}
if (-not $selfCheck.ok) {
throw "self_check failed: $($selfCheck | ConvertTo-Json -Depth 8 -Compress)"
}
$scan = Invoke-HelperJson @{
protocol = "isphere.helper.v1"
request_id = "verify-scan"
op = "scan_windows"
timeout_ms = 5000
args = @{ include_all_visible = $true }
}
if (-not $scan.ok -or $null -eq $scan.data.windows) {
throw "scan_windows failed: $($scan | ConvertTo-Json -Depth 8 -Compress)"
}
$dumpMissing = Invoke-HelperJson @{
protocol = "isphere.helper.v1"
request_id = "verify-dump-missing"
op = "dump_uia"
timeout_ms = 5000
args = @{ hwnd = "0x0"; max_depth = 1; include_text = $true; max_children = 20 }
}
if ($dumpMissing.ok -or $dumpMissing.error.code -notin @("WINDOW_NOT_FOUND", "UIA_DUMP_FAILED")) {
throw "dump_uia missing-window check failed: $($dumpMissing | ConvertTo-Json -Depth 8 -Compress)"
}
$title = "Codex WinHelper Verify " + [guid]::NewGuid().ToString("N").Substring(0, 8)
$formScript = "Add-Type -AssemblyName System.Windows.Forms; `$form = New-Object System.Windows.Forms.Form; `$form.Text = '$title'; `$form.Width = 420; `$form.Height = 180; `$button = New-Object System.Windows.Forms.Button; `$button.Text = 'ProbeButton'; `$button.Dock = 'Fill'; `$form.Controls.Add(`$button); [void]`$form.ShowDialog()"
$process = Start-Process powershell -ArgumentList @("-NoProfile", "-Command", $formScript) -PassThru
try {
$candidate = $null
for ($i = 0; $i -lt 20 -and -not $candidate; $i++) {
Start-Sleep -Milliseconds 500
$scanForForm = Invoke-HelperJson @{
protocol = "isphere.helper.v1"
request_id = "verify-form-scan"
op = "scan_windows"
timeout_ms = 5000
args = @{ include_all_visible = $true }
}
$candidate = $scanForForm.data.windows | Where-Object { $_.title -eq $title } | Select-Object -First 1
}
if (-not $candidate) {
throw "verification form window not found"
}
$dump = Invoke-HelperJson @{
protocol = "isphere.helper.v1"
request_id = "verify-form-dump"
op = "dump_uia"
timeout_ms = 5000
args = @{ hwnd = $candidate.hwnd; max_depth = 4; include_text = $true; max_children = 80 }
}
if (-not $dump.ok -or $dump.data.root.control_type -ne "Window") {
throw "dump_uia form check failed: $($dump | ConvertTo-Json -Depth 12 -Compress)"
}
}
finally {
if ($process -and -not $process.HasExited) {
Stop-Process -Id $process.Id -Force
}
}
[ordered]@{
ok = $true
helper = (Resolve-Path -LiteralPath $helperPath).Path
version = $version.data.helper_version
scan_window_count = $scan.data.windows.Count
uia_available = $selfCheck.data.uia_available
} | ConvertTo-Json -Depth 4 -Compress

View File

@@ -1,6 +0,0 @@
"""Read-only iSphere local message adapter POC."""
__all__ = ["__version__"]
__version__ = "0.1.0"

View File

@@ -1,226 +0,0 @@
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any
if __package__ in {None, ""}:
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from isphere_ai_bridge.draft import generate_drafts
from isphere_ai_bridge.draft_queue import DraftQueue
from isphere_ai_bridge.local_db import ISphereLocalDbReader
from isphere_ai_bridge.locator import discover_msglib_databases
from isphere_ai_bridge import mcp_api
from isphere_ai_bridge.mock_db import create_mock_msglib
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
prog="isphere-ai-bridge",
description="Developer CLI for the read-only iSphere MCP message service.",
)
subparsers = parser.add_subparsers(dest="command", required=True)
discover = subparsers.add_parser("discover", help="Find candidate MsgLib.db paths.")
discover.add_argument("--base", type=Path, default=None)
create_mock = subparsers.add_parser("create-mock-db", help="Create a mock MsgLib.db.")
create_mock.add_argument("--output", type=Path, required=True)
create_mock.add_argument("--force", action="store_true")
probe = subparsers.add_parser("probe", help="Probe a MsgLib.db in read-only mode.")
probe.add_argument("--db", type=Path, required=True)
read = subparsers.add_parser("read", help="Read normalized messages.")
read.add_argument("--db", type=Path, required=True)
read.add_argument("--limit", type=int, default=20)
read.add_argument("--jsonl", action="store_true")
recent = subparsers.add_parser("recent", help="Read recent conversations.")
recent.add_argument("--db", type=Path, required=True)
recent.add_argument("--limit", type=int, default=20)
draft = subparsers.add_parser("draft", help="Generate draft-only replies.")
draft.add_argument("--db", type=Path, required=True)
draft.add_argument("--limit", type=int, default=5)
queue_drafts = subparsers.add_parser(
"queue-drafts", help="Generate drafts and append them to review queue files."
)
queue_drafts.add_argument("--db", type=Path, required=True)
queue_drafts.add_argument("--queue", type=Path, required=True)
queue_drafts.add_argument("--audit", type=Path, required=True)
queue_drafts.add_argument("--limit", type=int, default=5)
lab_start = subparsers.add_parser(
"offline-lab-start", help="Create or reopen an offline simulated iSphere main-window session."
)
lab_start.add_argument("--workspace", default="runs/offline-lab")
lab_start.add_argument("--exe", default=None)
lab_start.add_argument("--reset", action="store_true")
lab_status = subparsers.add_parser("offline-lab-status", help="Show offline lab status.")
lab_status.add_argument("--workspace", default="runs/offline-lab")
lab_search = subparsers.add_parser("offline-lab-search", help="Search offline lab users.")
lab_search.add_argument("query", nargs="?", default="")
lab_search.add_argument("--workspace", default="runs/offline-lab")
lab_search.add_argument("--limit", type=int, default=20)
lab_open = subparsers.add_parser("offline-lab-open-chat", help="Open a simulated chat panel.")
lab_open.add_argument("user_jid")
lab_open.add_argument("--workspace", default="runs/offline-lab")
lab_send = subparsers.add_parser("offline-lab-send-message", help="Simulate sending a message.")
lab_send.add_argument("user_jid")
lab_send.add_argument("text")
lab_send.add_argument("--workspace", default="runs/offline-lab")
lab_recv = subparsers.add_parser("offline-lab-receive-message", help="Simulate receiving a message.")
lab_recv.add_argument("user_jid")
lab_recv.add_argument("text")
lab_recv.add_argument("--workspace", default="runs/offline-lab")
lab_send_file = subparsers.add_parser("offline-lab-send-file", help="Simulate sending a file.")
lab_send_file.add_argument("user_jid")
lab_send_file.add_argument("file_path")
lab_send_file.add_argument("--file-name", default=None)
lab_send_file.add_argument("--workspace", default="runs/offline-lab")
lab_recv_file = subparsers.add_parser("offline-lab-receive-file", help="Simulate receiving a file.")
lab_recv_file.add_argument("user_jid")
lab_recv_file.add_argument("file_name")
lab_recv_file.add_argument("--size", type=int, default=0)
lab_recv_file.add_argument("--workspace", default="runs/offline-lab")
lab_read = subparsers.add_parser("offline-lab-read", help="Read a simulated conversation.")
lab_read.add_argument("user_jid")
lab_read.add_argument("--limit", type=int, default=50)
lab_read.add_argument("--workspace", default="runs/offline-lab")
args = parser.parse_args(argv)
if args.command == "discover":
return _print_json([item.to_dict() for item in discover_msglib_databases(args.base)])
if args.command == "create-mock-db":
path = create_mock_msglib(args.output, force=args.force)
return _print_json({"created": str(path), "kind": "mock-msglib"})
if args.command == "probe":
return _print_json(ISphereLocalDbReader(args.db).probe())
if args.command == "read":
messages = [item.to_dict() for item in ISphereLocalDbReader(args.db).read_messages(args.limit)]
if args.jsonl:
for message in messages:
print(json.dumps(message, ensure_ascii=False))
return 0
return _print_json(messages)
if args.command == "recent":
return _print_json(ISphereLocalDbReader(args.db).read_recent_conversations(args.limit))
if args.command == "draft":
reader = ISphereLocalDbReader(args.db)
drafts = [item.to_dict() for item in generate_drafts(reader.read_messages(args.limit))]
return _print_json(drafts)
if args.command == "queue-drafts":
reader = ISphereLocalDbReader(args.db)
drafts = generate_drafts(reader.read_messages(args.limit))
entries = DraftQueue(args.queue, args.audit).append_drafts(drafts)
return _print_json(
{
"queued": len(entries),
"queue": str(args.queue),
"audit": str(args.audit),
"draft_ids": [entry["draft_id"] for entry in entries],
}
)
if args.command == "offline-lab-start":
return _print_json(
mcp_api.offline_lab_start(
exe_path=args.exe,
workspace=args.workspace,
reset=args.reset,
)
)
if args.command == "offline-lab-status":
return _print_json(mcp_api.offline_lab_status(workspace=args.workspace))
if args.command == "offline-lab-search":
return _print_json(
mcp_api.offline_lab_search_users(
query=args.query,
limit=args.limit,
workspace=args.workspace,
)
)
if args.command == "offline-lab-open-chat":
return _print_json(mcp_api.offline_lab_open_chat(args.user_jid, workspace=args.workspace))
if args.command == "offline-lab-send-message":
return _print_json(
mcp_api.offline_lab_send_message(
user_jid=args.user_jid,
text=args.text,
workspace=args.workspace,
)
)
if args.command == "offline-lab-receive-message":
return _print_json(
mcp_api.offline_lab_receive_message(
user_jid=args.user_jid,
text=args.text,
workspace=args.workspace,
)
)
if args.command == "offline-lab-send-file":
return _print_json(
mcp_api.offline_lab_send_file(
user_jid=args.user_jid,
file_path=args.file_path,
file_name=args.file_name,
workspace=args.workspace,
)
)
if args.command == "offline-lab-receive-file":
return _print_json(
mcp_api.offline_lab_receive_file(
user_jid=args.user_jid,
file_name=args.file_name,
size=args.size,
workspace=args.workspace,
)
)
if args.command == "offline-lab-read":
return _print_json(
mcp_api.offline_lab_read_conversation(
user_jid=args.user_jid,
limit=args.limit,
workspace=args.workspace,
)
)
parser.error(f"Unknown command: {args.command}")
return 2
def _print_json(value: Any) -> int:
print(json.dumps(value, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -1,24 +0,0 @@
from __future__ import annotations
from .models import Draft, NormalizedMessage
def generate_draft(message: NormalizedMessage) -> Draft:
subject = message.content_text.strip()
if len(subject) > 80:
subject = subject[:77] + "..."
draft_text = (
"Draft only. I have received this message and will prepare a response "
f"after checking the context: {subject}"
)
return Draft(
message_id=message.message_id,
conversation_id=message.conversation_id,
draft_text=draft_text,
source_message_ref=message.raw_ref,
)
def generate_drafts(messages: list[NormalizedMessage]) -> list[Draft]:
return [generate_draft(message) for message in messages]

View File

@@ -1,74 +0,0 @@
from __future__ import annotations
import hashlib
import json
from collections.abc import Callable, Iterable
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from .models import Draft
class DraftQueue:
def __init__(
self,
queue_path: Path,
audit_path: Path,
clock: Callable[[], str] | None = None,
):
self.queue_path = Path(queue_path)
self.audit_path = Path(audit_path)
self.clock = clock or _utc_now
def append_drafts(self, drafts: Iterable[Draft]) -> list[dict[str, Any]]:
created: list[dict[str, Any]] = []
for draft in drafts:
created_at = self.clock()
draft_id = _draft_id(draft, created_at)
record = {
"draft_id": draft_id,
"message_id": draft.message_id,
"conversation_id": draft.conversation_id,
"draft_text": draft.draft_text,
"source_message_ref": draft.source_message_ref,
"status": "pending_review",
"created_at": created_at,
}
audit_record = {
"event_type": "draft_created",
"draft_id": draft_id,
"message_id": draft.message_id,
"conversation_id": draft.conversation_id,
"source_message_ref": draft.source_message_ref,
"status": "pending_review",
"created_at": created_at,
"content_sha256": _sha256(draft.draft_text),
}
_append_jsonl(self.queue_path, record)
_append_jsonl(self.audit_path, audit_record)
created.append(record)
return created
def _append_jsonl(path: Path, record: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8", newline="\n") as handle:
handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True))
handle.write("\n")
def _draft_id(draft: Draft, created_at: str) -> str:
value = "|".join(
[draft.message_id, draft.conversation_id, draft.source_message_ref, created_at]
)
return "draft-" + _sha256(value)[:16]
def _sha256(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def _utc_now() -> str:
return datetime.now(timezone.utc).isoformat()

View File

@@ -1,249 +0,0 @@
from __future__ import annotations
import os
import sqlite3
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import quote
from .models import NormalizedMessage
class LocalDatabaseError(RuntimeError):
pass
class ReadOnlyConnection(sqlite3.Connection):
def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> bool:
result = super().__exit__(exc_type, exc_value, traceback)
self.close()
return result
def read_only_uri(path: Path) -> str:
resolved = Path(path).resolve()
normalized = str(resolved).replace(os.sep, "/")
return f"file:{quote(normalized, safe=':/')}?mode=ro"
def open_readonly_connection(path: Path) -> sqlite3.Connection:
db_path = Path(path)
if not db_path.exists():
raise LocalDatabaseError(f"MsgLib.db not found: {db_path}")
try:
con = sqlite3.connect(
read_only_uri(db_path),
uri=True,
timeout=1,
cached_statements=0,
factory=ReadOnlyConnection,
)
con.row_factory = sqlite3.Row
cursor = con.execute("PRAGMA query_only = ON")
cursor.close()
return con
except sqlite3.DatabaseError as exc:
raise LocalDatabaseError(
"Could not open MsgLib.db with the standard sqlite3 read-only driver. "
"A real iSphere database may require the packaged System.Data.SQLite password path."
) from exc
class ISphereLocalDbReader:
def __init__(self, db_path: Path):
self.db_path = Path(db_path)
def probe(self) -> dict[str, Any]:
with open_readonly_connection(self.db_path) as con:
tables = [
row["name"]
for row in _query_rows(
con,
"select name from sqlite_master where type='table' order by name",
)
]
return {
"path": str(self.db_path),
"can_open_readonly": True,
"tables": tables,
"message_tables": [
table
for table in tables
if table
in {
"tblPersonMsg",
"tblMsgGroupPersonMsg",
"tblRecent",
"TD_SystemMessageRecord",
"tblPluginMsg",
}
],
}
def read_messages(self, limit: int = 50) -> list[NormalizedMessage]:
if limit <= 0:
return []
with open_readonly_connection(self.db_path) as con:
tables = set(self._table_names(con))
messages: list[NormalizedMessage] = []
if "tblPersonMsg" in tables:
messages.extend(self._read_direct_messages(con, limit))
if "tblMsgGroupPersonMsg" in tables:
messages.extend(self._read_group_messages(con, limit))
if "TD_SystemMessageRecord" in tables:
messages.extend(self._read_system_messages(con, limit))
return sorted(messages, key=lambda item: item.created_at, reverse=True)[:limit]
def read_recent_conversations(self, limit: int = 50) -> list[dict[str, Any]]:
if limit <= 0:
return []
with open_readonly_connection(self.db_path) as con:
if "tblRecent" not in set(self._table_names(con)):
return []
rows = _query_rows(
con,
"""
select Id, PersonId, PersonName, ActionTime, FType
from tblRecent
order by ActionTime desc
limit ?
""",
(limit,),
)
return [dict(row) for row in rows]
def _read_direct_messages(
self, con: sqlite3.Connection, limit: int
) -> Iterable[NormalizedMessage]:
rows = _query_rows(
con,
"""
select rowid as _rowid, *
from tblPersonMsg
order by ActionTime desc, rowid desc
limit ?
""",
(limit,),
)
for row in rows:
yield NormalizedMessage(
message_id=_value(row, "Id"),
conversation_id=_first(row, "ObjPersonId", "SndPersonId", "RecvPersonId"),
conversation_name=_first(row, "ObjPerson", "SndPerson", "RecvPerson"),
conversation_type="direct",
sender_id=_value(row, "SndPersonId"),
sender_name=_value(row, "SndPerson"),
content_type=_content_type(_value(row, "MsgType")),
content_text=_first(row, "MsgContent", "MsgText", "MessageBody"),
created_at=_value(row, "ActionTime"),
received_at=_value(row, "ActionTime"),
raw_ref=f"{self.db_path}#tblPersonMsg:{_value(row, 'Id')}",
raw_type=_value(row, "MsgType"),
)
def _read_group_messages(
self, con: sqlite3.Connection, limit: int
) -> Iterable[NormalizedMessage]:
rows = _query_rows(
con,
"""
select rowid as _rowid, *
from tblMsgGroupPersonMsg
order by MsgTime desc, rowid desc
limit ?
""",
(limit,),
)
for row in rows:
yield NormalizedMessage(
message_id=_value(row, "Id"),
conversation_id=_value(row, "MsgGroupId"),
conversation_name=_value(row, "MsgGroupName"),
conversation_type="group",
sender_id=_value(row, "SndPersonId"),
sender_name=_value(row, "SndPerson"),
content_type=_content_type(_value(row, "MsgType")),
content_text=_first(row, "MsgContent", "MsgText", "MessageBody"),
created_at=_value(row, "MsgTime"),
received_at=_value(row, "MsgTime"),
raw_ref=f"{self.db_path}#tblMsgGroupPersonMsg:{_value(row, 'Id')}",
raw_type=_value(row, "MsgType"),
)
def _read_system_messages(
self, con: sqlite3.Connection, limit: int
) -> Iterable[NormalizedMessage]:
rows = _query_rows(
con,
"""
select rowid as _rowid, *
from TD_SystemMessageRecord
order by MsgTime desc, rowid desc
limit ?
""",
(limit,),
)
for row in rows:
yield NormalizedMessage(
message_id=_value(row, "MsgGuid"),
conversation_id="system",
conversation_name="System",
conversation_type="system",
sender_id=_value(row, "SendPersonJid"),
sender_name=_value(row, "SendPersonName"),
content_type="notify",
content_text=_value(row, "MsgText"),
created_at=_value(row, "MsgTime"),
received_at=_value(row, "MsgTime"),
raw_ref=f"{self.db_path}#TD_SystemMessageRecord:{_value(row, 'MsgGuid')}",
raw_type=_value(row, "MsgType"),
)
@staticmethod
def _table_names(con: sqlite3.Connection) -> list[str]:
rows = _query_rows(
con, "select name from sqlite_master where type='table' order by name"
)
return [row["name"] for row in rows]
def _query_rows(
con: sqlite3.Connection, sql: str, params: tuple[Any, ...] = ()
) -> list[sqlite3.Row]:
cursor = con.execute(sql, params)
try:
return list(cursor.fetchall())
finally:
cursor.close()
def _value(row: sqlite3.Row, key: str) -> str:
if key not in row.keys():
return ""
value = row[key]
return "" if value is None else str(value)
def _first(row: sqlite3.Row, *keys: str) -> str:
for key in keys:
value = _value(row, key).strip()
if value:
return value
return ""
def _content_type(raw_type: str) -> str:
normalized = raw_type.upper()
if "IMG" in normalized or "IMAGE" in normalized:
return "image"
if "FILE" in normalized or "FOLDER" in normalized:
return "file"
if "SYSTEM" in normalized or "NOTICE" in normalized or "AUTH" in normalized:
return "notify"
if "SUPER" in normalized or "VOICE" in normalized:
return "rich"
return "text"
IphereLocalDbReader = ISphereLocalDbReader

View File

@@ -1,57 +0,0 @@
from __future__ import annotations
import os
from pathlib import Path
from .models import CandidateDatabase
def default_isphere_root() -> Path:
user_profile = os.environ.get("USERPROFILE")
if user_profile:
return Path(user_profile) / "Documents" / "isphere"
return Path.home() / "Documents" / "isphere"
def discover_msglib_databases(base: Path | None = None) -> list[CandidateDatabase]:
root = Path(base) if base else default_isphere_root()
if not root.exists():
return []
candidates: list[CandidateDatabase] = []
account_dirs = sorted(root.glob("*/*/Account/*"))
for account_dir in account_dirs:
if not account_dir.is_dir():
continue
db_path = account_dir / "MsgLib.db"
parts = account_dir.relative_to(root).parts
connection_id = parts[0] if len(parts) > 0 else None
organization_id = parts[1] if len(parts) > 1 else None
account_id = parts[3] if len(parts) > 3 else account_dir.name
candidates.append(
CandidateDatabase(
path=db_path,
exists=db_path.exists(),
account_id=account_id,
organization_id=organization_id,
connection_id=connection_id,
reason="Account directory contains expected MsgLib.db path",
)
)
existing = {candidate.path.resolve() for candidate in candidates if candidate.exists}
for db_path in sorted(root.rglob("MsgLib.db")):
resolved = db_path.resolve()
if resolved in existing:
continue
candidates.append(
CandidateDatabase(
path=db_path,
exists=True,
reason="Loose MsgLib.db match under iSphere documents root",
)
)
return candidates

View File

@@ -1,502 +0,0 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from .draft import generate_drafts
from .draft_queue import DraftQueue
from .local_db import ISphereLocalDbReader
from .locator import discover_msglib_databases
from .mock_db import create_mock_msglib
from . import offline_lab
from . import native_helper
from . import real_lab
DEFAULT_DRAFT_QUEUE = Path("runs/mcp/drafts.jsonl")
DEFAULT_AUDIT_LOG = Path("runs/mcp/audit.jsonl")
MAX_LIMIT = 100
def discover_message_sources(base: str | None = None) -> dict[str, Any]:
candidates = discover_msglib_databases(Path(base) if base else None)
return {
"sources": [candidate.to_dict() for candidate in candidates],
"count": len(candidates),
"mode": "read-only-discovery",
}
def create_mock_source(output: str = "runs/mcp/MsgLib.db", force: bool = False) -> dict[str, Any]:
path = create_mock_msglib(Path(output), force=force)
return {"created": str(path), "kind": "mock-msglib"}
def read_messages(db_path: str, limit: int = 20) -> dict[str, Any]:
bounded_limit = _bounded_limit(limit)
reader = ISphereLocalDbReader(Path(db_path))
messages = [message.to_dict() for message in reader.read_messages(bounded_limit)]
return {
"source": str(db_path),
"limit": bounded_limit,
"count": len(messages),
"messages": messages,
}
def read_recent_conversations(db_path: str, limit: int = 20) -> dict[str, Any]:
bounded_limit = _bounded_limit(limit)
reader = ISphereLocalDbReader(Path(db_path))
conversations = reader.read_recent_conversations(bounded_limit)
return {
"source": str(db_path),
"limit": bounded_limit,
"count": len(conversations),
"conversations": conversations,
}
def generate_draft_preview(db_path: str, limit: int = 5) -> dict[str, Any]:
bounded_limit = _bounded_limit(limit)
reader = ISphereLocalDbReader(Path(db_path))
drafts = [draft.to_dict() for draft in generate_drafts(reader.read_messages(bounded_limit))]
return {
"source": str(db_path),
"limit": bounded_limit,
"count": len(drafts),
"drafts": drafts,
"status": "draft-only-not-queued",
}
def queue_drafts_for_review(
db_path: str,
queue_path: str = str(DEFAULT_DRAFT_QUEUE),
audit_path: str = str(DEFAULT_AUDIT_LOG),
limit: int = 5,
) -> dict[str, Any]:
bounded_limit = _bounded_limit(limit)
reader = ISphereLocalDbReader(Path(db_path))
drafts = generate_drafts(reader.read_messages(bounded_limit))
entries = DraftQueue(Path(queue_path), Path(audit_path)).append_drafts(drafts)
return {
"source": str(db_path),
"queued": len(entries),
"queue_path": queue_path,
"audit_path": audit_path,
"draft_ids": [entry["draft_id"] for entry in entries],
"status": "pending_review",
}
def read_audit_events(audit_path: str = str(DEFAULT_AUDIT_LOG), limit: int = 20) -> dict[str, Any]:
bounded_limit = _bounded_limit(limit)
path = Path(audit_path)
if not path.exists():
return {"audit_path": audit_path, "count": 0, "events": []}
lines = path.read_text(encoding="utf-8").splitlines()
events = [json.loads(line) for line in lines[-bounded_limit:] if line.strip()]
return {"audit_path": audit_path, "count": len(events), "events": events}
def offline_lab_start(
exe_path: str | None = None,
workspace: str = str(offline_lab.DEFAULT_WORKSPACE),
reset: bool = False,
) -> dict[str, Any]:
return offline_lab.start(exe_path=exe_path, workspace=workspace, reset=reset)
def offline_lab_status(workspace: str = str(offline_lab.DEFAULT_WORKSPACE)) -> dict[str, Any]:
return offline_lab.status(workspace=workspace)
def offline_lab_search_users(
query: str = "",
limit: int = 20,
workspace: str = str(offline_lab.DEFAULT_WORKSPACE),
) -> dict[str, Any]:
return offline_lab.search_users(query=query, limit=limit, workspace=workspace)
def offline_lab_open_chat(
user_jid: str,
workspace: str = str(offline_lab.DEFAULT_WORKSPACE),
) -> dict[str, Any]:
return offline_lab.open_chat(user_jid=user_jid, workspace=workspace)
def offline_lab_send_message(
user_jid: str,
text: str,
workspace: str = str(offline_lab.DEFAULT_WORKSPACE),
) -> dict[str, Any]:
return offline_lab.send_message(user_jid=user_jid, text=text, workspace=workspace)
def offline_lab_receive_message(
user_jid: str,
text: str,
workspace: str = str(offline_lab.DEFAULT_WORKSPACE),
) -> dict[str, Any]:
return offline_lab.receive_message(user_jid=user_jid, text=text, workspace=workspace)
def offline_lab_send_file(
user_jid: str,
file_path: str,
file_name: str | None = None,
workspace: str = str(offline_lab.DEFAULT_WORKSPACE),
) -> dict[str, Any]:
return offline_lab.send_file(
user_jid=user_jid,
file_path=file_path,
file_name=file_name,
workspace=workspace,
)
def offline_lab_receive_file(
user_jid: str,
file_name: str,
size: int = 0,
workspace: str = str(offline_lab.DEFAULT_WORKSPACE),
) -> dict[str, Any]:
return offline_lab.receive_file(
user_jid=user_jid,
file_name=file_name,
size=size,
workspace=workspace,
)
def offline_lab_read_conversation(
user_jid: str,
limit: int = 50,
workspace: str = str(offline_lab.DEFAULT_WORKSPACE),
) -> dict[str, Any]:
return offline_lab.read_conversation(user_jid=user_jid, limit=limit, workspace=workspace)
def native_lab_helper_paths() -> dict[str, Any]:
return native_helper.helper_paths()
def native_lab_build_helper(output_dir: str = "runs/native-helper") -> dict[str, Any]:
return native_helper.build_native_helper(output_dir=output_dir)
def native_lab_inspect(
client_dir: str | None = None,
helper_exe: str | None = None,
) -> dict[str, Any]:
return native_helper.native_lab_inspect(client_dir=client_dir, helper_exe=helper_exe)
def native_lab_probe_main_window(
client_dir: str | None = None,
helper_exe: str | None = None,
construct: bool = False,
probe_chat: bool = False,
probe_actions: bool = False,
probe_files: bool = False,
probe_send_text: bool = False,
probe_receive_text: bool = False,
probe_send_file: bool = False,
probe_receive_file: bool = False,
user_jid: str = "mock-bob@offline.lab",
text: str = "native helper direct chat probe",
file_path: str | None = None,
file_name: str | None = None,
file_size: int = 0,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_helper.native_lab_probe_main_window(
client_dir=client_dir,
helper_exe=helper_exe,
construct=construct,
probe_chat=probe_chat,
probe_actions=probe_actions,
probe_files=probe_files,
probe_send_text=probe_send_text,
probe_receive_text=probe_receive_text,
probe_send_file=probe_send_file,
probe_receive_file=probe_receive_file,
user_jid=user_jid,
text=text,
file_path=file_path,
file_name=file_name,
file_size=file_size,
timeout_ms=timeout_ms,
)
def native_lab_direct_open_main_window(
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_helper.native_lab_direct_open_main_window(
client_dir=client_dir,
helper_exe=helper_exe,
timeout_ms=timeout_ms,
)
def native_lab_direct_send_message(
user_jid: str,
text: str,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_helper.native_lab_direct_send_message(
user_jid=user_jid,
text=text,
client_dir=client_dir,
helper_exe=helper_exe,
timeout_ms=timeout_ms,
)
def native_lab_direct_receive_message(
user_jid: str,
text: str,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_helper.native_lab_direct_receive_message(
user_jid=user_jid,
text=text,
client_dir=client_dir,
helper_exe=helper_exe,
timeout_ms=timeout_ms,
)
def native_lab_direct_send_file(
user_jid: str,
file_path: str | None = None,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_helper.native_lab_direct_send_file(
user_jid=user_jid,
file_path=file_path,
client_dir=client_dir,
helper_exe=helper_exe,
timeout_ms=timeout_ms,
)
def native_lab_direct_receive_file(
user_jid: str,
file_name: str,
size: int = 0,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_helper.native_lab_direct_receive_file(
user_jid=user_jid,
file_name=file_name,
size=size,
client_dir=client_dir,
helper_exe=helper_exe,
timeout_ms=timeout_ms,
)
def native_lab_start(
client_dir: str | None = None,
workspace: str = str(native_helper.DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
reset: bool = False,
) -> dict[str, Any]:
return native_helper.native_lab_start(
client_dir=client_dir,
workspace=workspace,
helper_exe=helper_exe,
reset=reset,
)
def native_lab_status(
workspace: str = str(native_helper.DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return native_helper.native_lab_status(workspace=workspace, helper_exe=helper_exe)
def native_lab_search_users(
query: str = "",
limit: int = 20,
workspace: str = str(native_helper.DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return native_helper.native_lab_search_users(
query=query,
limit=limit,
workspace=workspace,
helper_exe=helper_exe,
)
def native_lab_open_chat(
user_jid: str,
workspace: str = str(native_helper.DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return native_helper.native_lab_open_chat(
user_jid=user_jid,
workspace=workspace,
helper_exe=helper_exe,
)
def native_lab_send_message(
user_jid: str,
text: str,
workspace: str = str(native_helper.DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return native_helper.native_lab_send_message(
user_jid=user_jid,
text=text,
workspace=workspace,
helper_exe=helper_exe,
)
def native_lab_receive_message(
user_jid: str,
text: str,
workspace: str = str(native_helper.DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return native_helper.native_lab_receive_message(
user_jid=user_jid,
text=text,
workspace=workspace,
helper_exe=helper_exe,
)
def native_lab_send_file(
user_jid: str,
file_path: str,
file_name: str | None = None,
workspace: str = str(native_helper.DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return native_helper.native_lab_send_file(
user_jid=user_jid,
file_path=file_path,
file_name=file_name,
workspace=workspace,
helper_exe=helper_exe,
)
def native_lab_receive_file(
user_jid: str,
file_name: str,
size: int = 0,
workspace: str = str(native_helper.DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return native_helper.native_lab_receive_file(
user_jid=user_jid,
file_name=file_name,
size=size,
workspace=workspace,
helper_exe=helper_exe,
)
def native_lab_read_conversation(
user_jid: str,
limit: int = 50,
workspace: str = str(native_helper.DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return native_helper.native_lab_read_conversation(
user_jid=user_jid,
limit=limit,
workspace=workspace,
helper_exe=helper_exe,
)
def real_lab_plan_tree() -> dict[str, Any]:
return real_lab.plan_tree()
def real_lab_required_files() -> dict[str, Any]:
return real_lab.required_files_checklist()
def real_lab_action_contract() -> dict[str, Any]:
return real_lab.action_contract()
def real_lab_status(workspace: str = str(real_lab.DEFAULT_WORKSPACE)) -> dict[str, Any]:
return real_lab.status(workspace=workspace)
def real_lab_scan_windows(
workspace: str = str(real_lab.DEFAULT_WORKSPACE),
save_snapshot: bool = False,
) -> dict[str, Any]:
return real_lab.scan_windows(workspace=workspace, save_snapshot=save_snapshot)
def real_lab_win_helper_paths() -> dict[str, Any]:
return real_lab.win_helper_paths()
def real_lab_win_helper_build(output_dir: str = "runs/win-helper") -> dict[str, Any]:
return real_lab.build_win_helper(output_dir=output_dir)
def real_lab_win_helper_version(helper_exe: str | None = None) -> dict[str, Any]:
return real_lab.win_helper_version(helper_exe=helper_exe)
def real_lab_win_helper_self_check(helper_exe: str | None = None) -> dict[str, Any]:
return real_lab.win_helper_self_check(helper_exe=helper_exe)
def real_lab_win_helper_scan_windows(
include_all_visible: bool = False,
helper_exe: str | None = None,
) -> dict[str, Any]:
return real_lab.win_helper_scan_windows(
include_all_visible=include_all_visible,
helper_exe=helper_exe,
)
def real_lab_win_helper_dump_uia(
hwnd: str,
max_depth: int = 8,
include_text: bool = True,
max_children: int = 200,
helper_exe: str | None = None,
) -> dict[str, Any]:
return real_lab.win_helper_dump_uia(
hwnd=hwnd,
max_depth=max_depth,
include_text=include_text,
max_children=max_children,
helper_exe=helper_exe,
)
def _bounded_limit(limit: int) -> int:
if limit <= 0:
return 0
return min(limit, MAX_LIMIT)

View File

@@ -1,218 +0,0 @@
from __future__ import annotations
import sqlite3
from pathlib import Path
def create_mock_msglib(path: Path, force: bool = False) -> Path:
db_path = Path(path)
if db_path.exists() and not force:
raise FileExistsError(f"Refusing to overwrite existing file: {db_path}")
db_path.parent.mkdir(parents=True, exist_ok=True)
if db_path.exists():
db_path.unlink()
con = sqlite3.connect(db_path)
try:
con.executescript(
"""
create table tblPersonMsg (
Id varchar(100) primary key not null,
SndPerson varchar(200) not null,
SndPersonId varchar(100) not null,
MsgType varchar(100) not null,
MsgText text null,
MsgFont varchar(200) null,
FStatus varchar(100) not null,
ActionTime datetime(8),
RecvPerson varchar(200) null,
RecvPersonId varchar(100) null,
ObjPerson varchar(200) null,
ObjPersonId varchar(100) null,
SndPersonSex int null,
RecvPersonSex int null,
MsgReadState boolean null,
MsgRecaptionState boolean null,
IsReceipt boolean null,
MsgReceiptState boolean null,
MessageBody text null,
InterOut varchar(100) null,
MsgContent text null,
MsgVersion text null,
MsgReadStatus text null
);
create table tblMsgGroupPersonMsg (
Id varchar(100) primary key not null,
SndPerson varchar(200) null,
SndPersonId varchar(100) null,
MsgType varchar(100) null,
MsgText text null,
MsgFont varchar(200) null,
MsgTime datetime(8) null,
MsgGroupId varchar(100) null,
MsgGroupName varchar(200) null,
SessionTitle varchar(500) null,
SessionPersonId text null,
SessionPersonName text null,
GroupType varchar(100) null,
SndPersonSex bigint null,
MsgReadState bit null,
MsgRecaptionState bit null,
IsReceipt boolean null,
MsgReceiptState boolean null,
MessageBody text null,
InterOut varchar(100) null,
MsgContent text null,
MsgVersion text null,
MsgReadStatus text null,
MemInfoVersion varchar(20) null,
ReadMembers text default '',
UnreadTotal int null
);
create table tblRecent (
Id varchar(100) primary key not null,
PersonId varchar(100) null,
PersonName varchar(200) null,
ActionTime datetime(8) null,
FType varchar(100) null
);
create table TD_SystemMessageRecord (
MsgGuid varchar(100) primary key not null,
MsgType varchar(100) null,
MsgTitle varchar(200) null,
MsgText text null,
SendPersonJid varchar(100) null,
SendPersonName varchar(200) null,
MsgTime datetime(8) null,
MsgReadState boolean null
);
"""
)
con.executemany(
"""
insert into tblPersonMsg
(Id, SndPerson, SndPersonId, MsgType, MsgText, FStatus, ActionTime,
RecvPerson, RecvPersonId, ObjPerson, ObjPersonId, MsgReadState,
MsgRecaptionState, IsReceipt, MsgReceiptState, MessageBody,
MsgContent, MsgVersion, MsgReadStatus)
values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[
(
"p2p-001",
"Alice",
"alice@example",
"PERSON",
"Can you review the latest customer note?",
"online",
"2026-07-05 09:10:00",
"Digital Employee",
"bot@example",
"Alice",
"alice@example",
0,
0,
0,
0,
"Can you review the latest customer note?",
"Can you review the latest customer note?",
"1001",
"",
),
(
"p2p-002",
"Digital Employee",
"bot@example",
"PERSON",
"I will prepare a draft.",
"online",
"2026-07-05 09:11:00",
"Alice",
"alice@example",
"Alice",
"alice@example",
1,
0,
0,
0,
"I will prepare a draft.",
"I will prepare a draft.",
"1002",
"offline",
),
],
)
con.execute(
"""
insert into tblMsgGroupPersonMsg
(Id, SndPerson, SndPersonId, MsgType, MsgText, MsgTime, MsgGroupId,
MsgGroupName, SessionTitle, SessionPersonId, SessionPersonName,
GroupType, MsgReadState, MsgRecaptionState, IsReceipt, MessageBody,
MsgContent, MsgVersion, MsgReadStatus, UnreadTotal)
values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"grp-001",
"Bob",
"bob@example",
"PERSON",
"Please summarize today's project blockers.",
"2026-07-05 09:12:30",
"project-room@example",
"Project Room",
"Project Room",
"alice@example,bob@example",
"Alice,Bob",
"WORK",
0,
0,
0,
"Please summarize today's project blockers.",
"Please summarize today's project blockers.",
"2001",
"",
2,
),
)
con.execute(
"""
insert into TD_SystemMessageRecord
(MsgGuid, MsgType, MsgTitle, MsgText, SendPersonJid, SendPersonName,
MsgTime, MsgReadState)
values (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"sys-001",
"group_notice",
"Notice",
"A group notice was updated.",
"system@example",
"System",
"2026-07-05 09:13:00",
0,
),
)
con.executemany(
"""
insert into tblRecent (Id, PersonId, PersonName, ActionTime, FType)
values (?, ?, ?, ?, ?)
""",
[
("p2p-002", "alice@example", "Alice", "2026-07-05 09:11:00", "PERSON"),
(
"grp-001",
"project-room@example",
"Project Room",
"2026-07-05 09:12:30",
"WORK",
),
],
)
con.commit()
finally:
con.close()
return db_path

View File

@@ -1,58 +0,0 @@
from __future__ import annotations
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Literal
ConversationType = Literal["direct", "group", "system"]
ContentType = Literal["text", "image", "file", "rich", "notify"]
@dataclass(frozen=True)
class CandidateDatabase:
path: Path
exists: bool
account_id: str | None = None
organization_id: str | None = None
connection_id: str | None = None
reason: str = ""
def to_dict(self) -> dict[str, Any]:
data = asdict(self)
data["path"] = str(self.path)
return data
@dataclass(frozen=True)
class NormalizedMessage:
message_id: str
conversation_id: str
conversation_name: str
conversation_type: ConversationType
sender_id: str
sender_name: str
content_type: ContentType
content_text: str
attachments: list[dict[str, Any]] = field(default_factory=list)
created_at: str = ""
received_at: str = ""
source: str = "local-db"
raw_ref: str = ""
raw_type: str = ""
def to_dict(self) -> dict[str, Any]:
return asdict(self)
@dataclass(frozen=True)
class Draft:
message_id: str
conversation_id: str
draft_text: str
source_message_ref: str
status: str = "draft-only"
def to_dict(self) -> dict[str, Any]:
return asdict(self)

View File

@@ -1,450 +0,0 @@
from __future__ import annotations
import json
import subprocess
from pathlib import Path
from typing import Any
DEFAULT_CLIENT_DIR = Path("evidence/fresh-zero-20260705/keyfiles-flat")
DEFAULT_HELPER_EXE = Path("runs/native-helper/ISphereOfflineLabHelper.exe")
DEFAULT_NATIVE_WORKSPACE = Path("runs/native-offline-lab")
def build_native_helper(output_dir: str = "runs/native-helper") -> dict[str, Any]:
script = _repo_root() / "scripts" / "build-native-helper.ps1"
if not script.exists():
return {
"ok": False,
"error": "build_script_missing",
"script": str(script),
}
completed = subprocess.run(
[
"powershell",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-File",
str(script),
"-OutputDir",
output_dir,
],
cwd=_repo_root(),
text=True,
encoding="utf-8",
errors="replace",
capture_output=True,
)
return _json_process_result(completed)
def native_lab_inspect(
client_dir: str | None = None,
helper_exe: str | None = None,
) -> dict[str, Any]:
return _run_helper(
["inspect", "--client-dir", str(_resolve_client_dir(client_dir))],
helper_exe=helper_exe,
)
def native_lab_probe_main_window(
client_dir: str | None = None,
helper_exe: str | None = None,
construct: bool = False,
probe_chat: bool = False,
probe_actions: bool = False,
probe_files: bool = False,
probe_send_text: bool = False,
probe_receive_text: bool = False,
probe_send_file: bool = False,
probe_receive_file: bool = False,
user_jid: str = "mock-bob@offline.lab",
text: str = "native helper direct chat probe",
file_path: str | None = None,
file_name: str | None = None,
file_size: int = 0,
timeout_ms: int = 15000,
) -> dict[str, Any]:
args = [
"probe-main-window",
"--client-dir",
str(_resolve_client_dir(client_dir)),
"--timeout-ms",
str(timeout_ms),
]
if construct:
args.append("--construct")
if probe_chat:
args.append("--probe-chat")
if probe_actions:
args.append("--probe-actions")
if probe_files:
args.append("--probe-files")
if probe_send_text:
args.append("--probe-send-text")
if probe_receive_text:
args.append("--probe-receive-text")
if probe_send_file:
args.append("--probe-send-file")
if probe_receive_file:
args.append("--probe-receive-file")
if user_jid:
args.extend(["--user-jid", user_jid])
if text:
args.extend(["--text", text])
if file_path:
args.extend(["--file-path", file_path])
if file_name:
args.extend(["--file-name", file_name])
if file_size:
args.extend(["--file-size", str(file_size)])
return _run_helper(args, helper_exe=helper_exe)
def native_lab_direct_open_main_window(
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_lab_probe_main_window(
client_dir=client_dir,
helper_exe=helper_exe,
construct=True,
timeout_ms=timeout_ms,
)
def native_lab_direct_send_message(
user_jid: str,
text: str,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_lab_probe_main_window(
client_dir=client_dir,
helper_exe=helper_exe,
probe_send_text=True,
user_jid=user_jid,
text=text,
timeout_ms=timeout_ms,
)
def native_lab_direct_receive_message(
user_jid: str,
text: str,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_lab_probe_main_window(
client_dir=client_dir,
helper_exe=helper_exe,
probe_receive_text=True,
user_jid=user_jid,
text=text,
timeout_ms=timeout_ms,
)
def native_lab_direct_send_file(
user_jid: str,
file_path: str | None = None,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_lab_probe_main_window(
client_dir=client_dir,
helper_exe=helper_exe,
probe_send_file=True,
user_jid=user_jid,
file_path=file_path,
timeout_ms=timeout_ms,
)
def native_lab_direct_receive_file(
user_jid: str,
file_name: str,
size: int = 0,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict[str, Any]:
return native_lab_probe_main_window(
client_dir=client_dir,
helper_exe=helper_exe,
probe_receive_file=True,
user_jid=user_jid,
file_name=file_name,
file_size=size,
timeout_ms=timeout_ms,
)
def native_lab_start(
client_dir: str | None = None,
workspace: str = str(DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
reset: bool = False,
) -> dict[str, Any]:
args = [
"start",
"--client-dir",
str(_resolve_client_dir(client_dir)),
"--workspace",
workspace,
]
if reset:
args.append("--reset")
return _run_helper(args, helper_exe=helper_exe)
def native_lab_status(
workspace: str = str(DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return _run_helper(["status", "--workspace", workspace], helper_exe=helper_exe)
def native_lab_search_users(
query: str = "",
limit: int = 20,
workspace: str = str(DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return _run_helper(
[
"search-users",
"--workspace",
workspace,
"--query",
query,
"--limit",
str(limit),
],
helper_exe=helper_exe,
)
def native_lab_open_chat(
user_jid: str,
workspace: str = str(DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return _run_helper(
["open-chat", "--workspace", workspace, "--user-jid", user_jid],
helper_exe=helper_exe,
)
def native_lab_send_message(
user_jid: str,
text: str,
workspace: str = str(DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return _run_helper(
[
"send-message",
"--workspace",
workspace,
"--user-jid",
user_jid,
"--text",
text,
],
helper_exe=helper_exe,
)
def native_lab_receive_message(
user_jid: str,
text: str,
workspace: str = str(DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return _run_helper(
[
"receive-message",
"--workspace",
workspace,
"--user-jid",
user_jid,
"--text",
text,
],
helper_exe=helper_exe,
)
def native_lab_send_file(
user_jid: str,
file_path: str,
file_name: str | None = None,
workspace: str = str(DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
args = [
"send-file",
"--workspace",
workspace,
"--user-jid",
user_jid,
"--file-path",
file_path,
]
if file_name:
args.extend(["--file-name", file_name])
return _run_helper(args, helper_exe=helper_exe)
def native_lab_receive_file(
user_jid: str,
file_name: str,
size: int = 0,
workspace: str = str(DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return _run_helper(
[
"receive-file",
"--workspace",
workspace,
"--user-jid",
user_jid,
"--file-name",
file_name,
"--size",
str(size),
],
helper_exe=helper_exe,
)
def native_lab_read_conversation(
user_jid: str,
limit: int = 50,
workspace: str = str(DEFAULT_NATIVE_WORKSPACE),
helper_exe: str | None = None,
) -> dict[str, Any]:
return _run_helper(
[
"read-conversation",
"--workspace",
workspace,
"--user-jid",
user_jid,
"--limit",
str(limit),
],
helper_exe=helper_exe,
)
def helper_paths() -> dict[str, Any]:
root = _repo_root()
helper = root / DEFAULT_HELPER_EXE
client = root / DEFAULT_CLIENT_DIR
build_script = root / "scripts" / "build-native-helper.ps1"
return {
"repo_root": str(root),
"helper_exe": str(helper),
"helper_exists": helper.exists(),
"default_client_dir": str(client),
"default_client_dir_exists": client.exists(),
"build_script": str(build_script),
"build_script_exists": build_script.exists(),
"build_command": (
"powershell -NoProfile -ExecutionPolicy Bypass "
"-File scripts\\build-native-helper.ps1"
),
}
def _run_helper(args: list[str], helper_exe: str | None = None) -> dict[str, Any]:
helper = _resolve_helper_exe(helper_exe)
if not helper.exists():
info = helper_paths()
return {
"ok": False,
"error": "native_helper_not_built",
"helper_exe": str(helper),
"build_command": info["build_command"],
"next_step": "run native_lab_build_helper or execute the build command",
}
completed = subprocess.run(
[str(helper), *args],
cwd=_repo_root(),
text=True,
encoding="utf-8",
errors="replace",
capture_output=True,
)
return _json_process_result(completed)
def _json_process_result(completed: subprocess.CompletedProcess[str]) -> dict[str, Any]:
parsed = _parse_last_json_line(completed.stdout)
if parsed is None:
return {
"ok": False,
"returncode": completed.returncode,
"stdout": completed.stdout,
"stderr": completed.stderr,
"error": "no_json_output",
}
if isinstance(parsed, dict):
parsed.setdefault("ok", completed.returncode == 0)
parsed["returncode"] = completed.returncode
if completed.stderr.strip():
parsed["stderr"] = completed.stderr
return parsed
return {
"ok": completed.returncode == 0,
"returncode": completed.returncode,
"value": parsed,
"stderr": completed.stderr,
}
def _parse_last_json_line(stdout: str) -> Any | None:
stripped_stdout = stdout.strip()
if stripped_stdout:
try:
return json.loads(stripped_stdout)
except json.JSONDecodeError:
pass
for line in reversed(stdout.splitlines()):
stripped = line.strip()
if not stripped:
continue
try:
return json.loads(stripped)
except json.JSONDecodeError:
continue
return None
def _resolve_client_dir(client_dir: str | None) -> Path:
path = Path(client_dir) if client_dir else DEFAULT_CLIENT_DIR
if not path.is_absolute():
path = _repo_root() / path
return path
def _resolve_helper_exe(helper_exe: str | None) -> Path:
path = Path(helper_exe) if helper_exe else DEFAULT_HELPER_EXE
if not path.is_absolute():
path = _repo_root() / path
return path
def _repo_root() -> Path:
return Path(__file__).resolve().parents[2]

View File

@@ -1,505 +0,0 @@
from __future__ import annotations
from copy import deepcopy
from datetime import datetime
from pathlib import Path
from typing import Any
from uuid import uuid4
DEFAULT_WORKSPACE = Path("runs/offline-lab")
MAX_LIMIT = 100
def start(
exe_path: str | None = None,
workspace: str | Path = DEFAULT_WORKSPACE,
reset: bool = False,
) -> dict[str, Any]:
"""Create or reopen an offline-lab session.
This is the Python-side state machine for the MCP bridge. It intentionally
simulates the iSphere UI surfaces instead of touching the original client.
The native launcher/patcher can later replace this backend without changing
the MCP tool contract.
"""
path = _state_path(workspace)
if path.exists() and not reset:
state = _load_state(path)
state["main_window"]["opened"] = True
state["updated_at"] = _now()
_append_audit(state, "offline_lab_reopened", {"workspace": str(Path(workspace))})
_save_state(path, state)
return _session_response(state, path, reused=True)
state = _initial_state(exe_path)
_append_audit(
state,
"offline_lab_started",
{
"exe_path": state["exe_path"],
"exe_exists": state["exe_exists"],
"workspace": str(Path(workspace)),
},
)
_save_state(path, state)
return _session_response(state, path, reused=False)
def status(workspace: str | Path = DEFAULT_WORKSPACE) -> dict[str, Any]:
path = _state_path(workspace)
if not path.exists():
return {
"started": False,
"workspace": str(Path(workspace)),
"state_path": str(path),
"next_step": "call offline_lab_start",
}
state = _load_state(path)
return {
"started": True,
"workspace": str(Path(workspace)),
"state_path": str(path),
"session_id": state["session_id"],
"mode": state["mode"],
"main_window": state["main_window"],
"active_conversation_id": state.get("active_conversation_id"),
"user_count": len(state["users"]),
"conversation_count": len(state["conversations"]),
"message_count": sum(len(conv["messages"]) for conv in state["conversations"].values()),
"native_bridge": state["native_bridge"],
"updated_at": state["updated_at"],
}
def search_users(
query: str = "",
limit: int = 20,
workspace: str | Path = DEFAULT_WORKSPACE,
) -> dict[str, Any]:
path, state = _require_state(workspace)
query_norm = query.casefold().strip()
users = []
for user in state["users"]:
searchable = " ".join(
str(user.get(key, ""))
for key in ("user_id", "jid", "name", "domain", "company_id", "department")
).casefold()
if not query_norm or query_norm in searchable:
users.append(deepcopy(user))
bounded_limit = _bounded_limit(limit)
result = users[:bounded_limit]
_append_audit(state, "offline_lab_search_users", {"query": query, "result_count": len(result)})
_save_state(path, state)
return {
"mode": state["mode"],
"query": query,
"limit": bounded_limit,
"count": len(result),
"users": result,
}
def open_chat(user_jid: str, workspace: str | Path = DEFAULT_WORKSPACE) -> dict[str, Any]:
path, state = _require_state(workspace)
user = _require_user(state, user_jid)
conversation = _ensure_conversation(state, user)
state["active_conversation_id"] = conversation["conversation_id"]
state["main_window"]["active_panel"] = "p2p-chat"
state["updated_at"] = _now()
_append_audit(
state,
"offline_lab_open_chat",
{"conversation_id": conversation["conversation_id"], "user_jid": user_jid},
)
_save_state(path, state)
return {
"opened": True,
"mode": state["mode"],
"main_window": deepcopy(state["main_window"]),
"conversation": _conversation_view(conversation, limit=0),
}
def send_message(
user_jid: str,
text: str,
workspace: str | Path = DEFAULT_WORKSPACE,
) -> dict[str, Any]:
path, state = _require_state(workspace)
user = _require_user(state, user_jid)
conversation = _ensure_conversation(state, user)
message = _message(
conversation=conversation,
direction="outgoing",
sender_id="offline-operator",
sender_name="离线操作员",
content_type="text",
content_text=text,
attachments=[],
)
conversation["messages"].append(message)
state["active_conversation_id"] = conversation["conversation_id"]
state["updated_at"] = _now()
_append_audit(
state,
"offline_lab_send_message_simulated",
{
"conversation_id": conversation["conversation_id"],
"message_id": message["message_id"],
"content_type": "text",
},
)
_save_state(path, state)
return {
"sent": True,
"simulated": True,
"mode": state["mode"],
"transport": "offline-lab-memory",
"message": deepcopy(message),
}
def receive_message(
user_jid: str,
text: str,
workspace: str | Path = DEFAULT_WORKSPACE,
) -> dict[str, Any]:
path, state = _require_state(workspace)
user = _require_user(state, user_jid)
conversation = _ensure_conversation(state, user)
message = _message(
conversation=conversation,
direction="incoming",
sender_id=user["user_id"],
sender_name=user["name"],
content_type="text",
content_text=text,
attachments=[],
)
conversation["messages"].append(message)
state["active_conversation_id"] = conversation["conversation_id"]
state["updated_at"] = _now()
_append_audit(
state,
"offline_lab_receive_message_simulated",
{
"conversation_id": conversation["conversation_id"],
"message_id": message["message_id"],
"content_type": "text",
},
)
_save_state(path, state)
return {
"received": True,
"simulated": True,
"mode": state["mode"],
"transport": "offline-lab-memory",
"message": deepcopy(message),
}
def send_file(
user_jid: str,
file_path: str,
file_name: str | None = None,
workspace: str | Path = DEFAULT_WORKSPACE,
) -> dict[str, Any]:
path, state = _require_state(workspace)
user = _require_user(state, user_jid)
conversation = _ensure_conversation(state, user)
source = Path(file_path)
exists = source.exists()
attachment = {
"name": file_name or source.name,
"path": str(source),
"exists": exists,
"size": source.stat().st_size if exists else 0,
"direction": "outgoing",
"transfer_state": "simulated-sent",
}
message = _message(
conversation=conversation,
direction="outgoing",
sender_id="offline-operator",
sender_name="离线操作员",
content_type="file",
content_text=attachment["name"],
attachments=[attachment],
)
conversation["messages"].append(message)
state["active_conversation_id"] = conversation["conversation_id"]
state["updated_at"] = _now()
_append_audit(
state,
"offline_lab_send_file_simulated",
{
"conversation_id": conversation["conversation_id"],
"message_id": message["message_id"],
"file_name": attachment["name"],
"exists": exists,
},
)
_save_state(path, state)
return {
"sent": True,
"simulated": True,
"mode": state["mode"],
"transport": "offline-lab-memory",
"file_event": deepcopy(message),
}
def receive_file(
user_jid: str,
file_name: str,
size: int = 0,
workspace: str | Path = DEFAULT_WORKSPACE,
) -> dict[str, Any]:
path, state = _require_state(workspace)
user = _require_user(state, user_jid)
conversation = _ensure_conversation(state, user)
attachment = {
"name": file_name,
"path": "",
"exists": False,
"size": max(0, int(size)),
"direction": "incoming",
"transfer_state": "simulated-received",
}
message = _message(
conversation=conversation,
direction="incoming",
sender_id=user["user_id"],
sender_name=user["name"],
content_type="file",
content_text=file_name,
attachments=[attachment],
)
conversation["messages"].append(message)
state["active_conversation_id"] = conversation["conversation_id"]
state["updated_at"] = _now()
_append_audit(
state,
"offline_lab_receive_file_simulated",
{
"conversation_id": conversation["conversation_id"],
"message_id": message["message_id"],
"file_name": file_name,
"size": attachment["size"],
},
)
_save_state(path, state)
return {
"received": True,
"simulated": True,
"mode": state["mode"],
"transport": "offline-lab-memory",
"file_event": deepcopy(message),
}
def read_conversation(
user_jid: str,
limit: int = 50,
workspace: str | Path = DEFAULT_WORKSPACE,
) -> dict[str, Any]:
_, state = _require_state(workspace)
user = _require_user(state, user_jid)
conversation = _ensure_conversation(state, user)
bounded_limit = _bounded_limit(limit)
return {
"mode": state["mode"],
"limit": bounded_limit,
"conversation": _conversation_view(conversation, bounded_limit),
}
def _initial_state(exe_path: str | None) -> dict[str, Any]:
exe_exists = Path(exe_path).exists() if exe_path else False
users = [
_user("U1001", "mock-alice@offline.lab", "张小真", "研发部"),
_user("U1002", "mock-bob@offline.lab", "李文件", "运营部"),
_user("U1003", "mock-carol@offline.lab", "王搜索", "客服部"),
_user("U1004", "mock-dave@offline.lab", "陈接收", "测试部"),
]
conversations = {}
for user in users:
conversations[user["jid"]] = {
"conversation_id": user["jid"],
"conversation_name": user["name"],
"conversation_type": "direct",
"user_jid": user["jid"],
"messages": [],
}
now = _now()
return {
"session_id": f"offline-lab-{uuid4().hex[:12]}",
"mode": "offline-lab-simulated",
"created_at": now,
"updated_at": now,
"exe_path": str(exe_path) if exe_path else None,
"exe_exists": exe_exists,
"main_window": {
"opened": True,
"caption": "iSphere Offline Lab Main",
"source": "reverse-map: frmLogin.LogonSuccessUI -> frmMain(frmLogin, notifyIcon)",
"active_panel": "contacts",
},
"native_bridge": {
"status": "not-attached",
"backend": "python-state-machine",
"next_step": "replace this backend with native helper/patch that instantiates frmMain with mock IMPPManager state",
},
"active_conversation_id": None,
"users": users,
"conversations": conversations,
"audit": [],
}
def _user(user_id: str, jid: str, name: str, department: str) -> dict[str, Any]:
return {
"user_id": user_id,
"jid": jid,
"name": name,
"domain": "offline.lab",
"company_id": "offline-company",
"department": department,
"enable": True,
"chat_able": True,
"source": "mock",
}
def _message(
conversation: dict[str, Any],
direction: str,
sender_id: str,
sender_name: str,
content_type: str,
content_text: str,
attachments: list[dict[str, Any]],
) -> dict[str, Any]:
now = _now()
message_id = f"{direction}-{uuid4().hex[:12]}"
return {
"message_id": message_id,
"conversation_id": conversation["conversation_id"],
"conversation_name": conversation["conversation_name"],
"conversation_type": conversation["conversation_type"],
"sender_id": sender_id,
"sender_name": sender_name,
"direction": direction,
"content_type": content_type,
"content_text": content_text,
"attachments": attachments,
"created_at": now,
"received_at": now,
"source": "offline-lab-mock",
"raw_ref": f"offline-lab:{conversation['conversation_id']}:{message_id}",
"ui_hook": _ui_hook(content_type, direction),
}
def _ui_hook(content_type: str, direction: str) -> str:
if content_type == "file" and direction == "incoming":
return "frmP2PChat.ProccessRecvMessage -> UC_FileReceive"
if content_type == "file":
return "frmP2PChat.SendTcpFiles/SendOfflineFiles -> UC_FileSend"
if direction == "incoming":
return "frmP2PChat.ProccessRecvMessage"
return "frmP2PChat.SaveAndShowMessage"
def _conversation_view(conversation: dict[str, Any], limit: int) -> dict[str, Any]:
view = deepcopy(conversation)
if limit <= 0:
view["messages"] = []
else:
view["messages"] = view["messages"][-limit:]
view["message_count"] = len(conversation["messages"])
return view
def _session_response(state: dict[str, Any], path: Path, reused: bool) -> dict[str, Any]:
return {
"started": True,
"reused": reused,
"session_id": state["session_id"],
"mode": state["mode"],
"workspace": str(path.parent),
"state_path": str(path),
"main_window": deepcopy(state["main_window"]),
"user_count": len(state["users"]),
"native_bridge": deepcopy(state["native_bridge"]),
}
def _ensure_conversation(state: dict[str, Any], user: dict[str, Any]) -> dict[str, Any]:
if user["jid"] not in state["conversations"]:
state["conversations"][user["jid"]] = {
"conversation_id": user["jid"],
"conversation_name": user["name"],
"conversation_type": "direct",
"user_jid": user["jid"],
"messages": [],
}
return state["conversations"][user["jid"]]
def _require_user(state: dict[str, Any], user_jid: str) -> dict[str, Any]:
for user in state["users"]:
if user["jid"] == user_jid:
return user
raise ValueError(f"unknown offline lab user_jid: {user_jid}")
def _require_state(workspace: str | Path) -> tuple[Path, dict[str, Any]]:
path = _state_path(workspace)
if not path.exists():
raise FileNotFoundError(f"offline lab state not found: {path}; call offline_lab_start first")
return path, _load_state(path)
def _append_audit(state: dict[str, Any], event_type: str, data: dict[str, Any]) -> None:
state["audit"].append(
{
"event_id": f"audit-{uuid4().hex[:12]}",
"event_type": event_type,
"created_at": _now(),
"data": data,
}
)
def _state_path(workspace: str | Path) -> Path:
workspace_path = Path(workspace)
return workspace_path / "session.json"
def _load_state(path: Path) -> dict[str, Any]:
import json
return json.loads(path.read_text(encoding="utf-8"))
def _save_state(path: Path, state: dict[str, Any]) -> None:
import json
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
def _bounded_limit(limit: int) -> int:
if limit <= 0:
return 0
return min(int(limit), MAX_LIMIT)
def _now() -> str:
return datetime.now().astimezone().isoformat(timespec="seconds")

View File

@@ -1,576 +0,0 @@
from __future__ import annotations
import json
import subprocess
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
DEFAULT_WORKSPACE = Path("runs/real-loggedin-lab")
DEFAULT_WIN_HELPER_EXE = Path("runs/win-helper/ISphereWinHelper.exe")
WIN_HELPER_PROTOCOL = "isphere.helper.v1"
PROCESS_NAME_HINTS = (
"IMPlatformClient",
"IMPP.ISphere",
"iSphere",
"importal",
)
@dataclass(frozen=True)
class ProcessWindow:
process_name: str
process_id: int
main_window_title: str
main_window_handle: int
def to_dict(self) -> dict[str, Any]:
return {
"process_name": self.process_name,
"process_id": self.process_id,
"main_window_title": self.main_window_title,
"main_window_handle": self.main_window_handle,
}
def plan_tree() -> dict[str, Any]:
return {
"mode": "real-loggedin-planning",
"objective": (
"用户手动登录 iSphere 后MCP 通过受控动作执行联系人搜索、打开会话、"
"写草稿、发送消息、发送文件、检查新消息和接收文件。"
),
"phases": [
{
"id": "A",
"name": "pre_real_data_work",
"status": "ready-now",
"tasks": [
"planning_tree",
"required_file_checklist",
"action_contract",
"safe_process_window_discovery",
"mcp_registration",
"tests_and_docs",
],
},
{
"id": "B",
"name": "real_file_intake",
"status": "blocked-waiting-real-data",
"tasks": [
"inventory_full_install_dir",
"inventory_user_data_dir",
"compare_against_keyfiles_flat",
"locate_real_msglib_and_logs",
"record_hashes_and_versions",
],
},
{
"id": "C",
"name": "logged_in_window_discovery",
"status": "blocked-waiting-logged-in-client",
"tasks": [
"find_process_candidates",
"capture_top_level_windows",
"capture_uia_tree",
"identify_search_chat_file_controls",
"save_selector_profile",
],
},
{
"id": "D",
"name": "read_and_search_operations",
"status": "planned",
"tasks": [
"real_lab_status",
"real_lab_search_contacts",
"real_lab_open_conversation",
"real_lab_read_current_conversation",
"real_lab_watch_latest_messages",
],
},
{
"id": "E",
"name": "draft_first_write_operations",
"status": "planned-approval-required",
"tasks": [
"real_lab_write_draft",
"verify_active_conversation",
"queue_approval_record",
"real_lab_send_after_approval",
],
},
{
"id": "F",
"name": "file_operations",
"status": "planned-approval-required",
"tasks": [
"real_lab_attach_file_draft",
"verify_file_and_conversation",
"real_lab_send_file_after_approval",
"detect_incoming_file_cards",
"real_lab_receive_file_after_approval",
],
},
{
"id": "G",
"name": "native_bridge_upgrade",
"status": "future",
"tasks": [
"long_lived_helper_process",
"ipc_command_channel",
"bind_to_logged_in_state_or_plugin_surface",
"replace_selectors_with_direct_calls_where_stable",
],
},
],
"acceptance_gates": [
"用户手动登录MCP 不自动输入账号密码",
"发送消息和上传/接收文件必须带 approval_id",
"每次动作写入审计事件",
"选择器必须来自真实已登录窗口快照",
"先测试联系人和会话定位,再开放发送动作",
],
}
def required_files_checklist() -> dict[str, Any]:
return {
"mode": "real-loggedin-file-intake",
"recommended_timing": "copy-later",
"target_roots": {
"install_dir": "evidence/real-install/",
"user_data": "evidence/real-userdata/",
"window_snapshots": "runs/real-loggedin-lab/window-snapshots/",
},
"sections": [
{
"id": "install_dir",
"title": "完整安装目录",
"priority": "required",
"examples": [
"IMPlatformClient.exe",
"IMPP.ISphere.exe",
"*.dll",
"*.config",
"*.xml",
"plugins/",
"Plugin/",
"Resources/",
"Skin/",
"Cef/",
"Log/",
],
"reason": "补齐真实运行所需插件、资源、皮肤、CEF 和配置文件。",
},
{
"id": "user_data",
"title": "当前 Windows 用户数据目录",
"priority": "required",
"examples": [
"%APPDATA% 中 iSphere/IMPP/IMPlatform/importal 相关目录",
"%LOCALAPPDATA% 中 iSphere/IMPP/IMPlatform/importal 相关目录",
"*.config",
"*.xml",
"*.ini",
"*.json",
"*.db",
"*.sqlite",
"*.log",
],
"reason": "定位真实缓存、消息库、日志、文件接收路径和用户级配置。",
},
{
"id": "logged_in_window",
"title": "用户手动登录后的窗口状态",
"priority": "required-for-rpa",
"examples": [
"保持主窗口在线",
"打开一次联系人搜索",
"打开一个测试联系人聊天窗口",
"不要让 MCP 自动输入登录凭据",
],
"reason": "采集真实窗口标题、句柄和控件树,生成 selector profile。",
},
{
"id": "redaction",
"title": "敏感信息处理",
"priority": "recommended",
"examples": [
"不复制密码",
"不复制私钥",
"不主动导出大批量生产聊天正文",
"日志如含 token 先保留原目录,后续用只读脚本做结构化脱敏",
],
"reason": "减少无关敏感数据进入证据目录。",
},
],
}
def action_contract() -> dict[str, Any]:
return {
"mode": "real-loggedin-action-contract",
"hard_limits": [
"no-login-automation",
"no-arbitrary-mouse-keyboard-control",
"no-token-extraction",
"no-send-or-file-action-without-approval-id",
],
"audit_fields": [
"timestamp",
"tool",
"target_window",
"target_contact",
"conversation_hint",
"content_summary",
"file_path",
"approval_id",
"result",
"evidence_ref",
],
"actions": [
{
"name": "real_lab_status",
"phase": "discovery",
"requires_approval": False,
"writes": "none",
"ready_before_real_data": True,
},
{
"name": "real_lab_scan_windows",
"phase": "discovery",
"requires_approval": False,
"writes": "optional snapshot under runs/real-loggedin-lab",
"ready_before_real_data": True,
},
{
"name": "real_lab_search_contacts",
"phase": "read-action",
"requires_approval": False,
"writes": "audit-only",
"ready_before_real_data": False,
},
{
"name": "real_lab_open_conversation",
"phase": "read-action",
"requires_approval": False,
"writes": "audit-only",
"ready_before_real_data": False,
},
{
"name": "real_lab_write_draft",
"phase": "draft",
"requires_approval": False,
"writes": "client draft box and audit",
"ready_before_real_data": False,
},
{
"name": "real_lab_send_after_approval",
"phase": "send",
"requires_approval": True,
"writes": "real client send action and audit",
"ready_before_real_data": False,
},
{
"name": "real_lab_send_file_after_approval",
"phase": "file-send",
"requires_approval": True,
"writes": "real client upload/send action and audit",
"ready_before_real_data": False,
},
{
"name": "real_lab_receive_file_after_approval",
"phase": "file-receive",
"requires_approval": True,
"writes": "real client download/receive action and audit",
"ready_before_real_data": False,
},
],
}
def status(workspace: str = str(DEFAULT_WORKSPACE)) -> dict[str, Any]:
windows = _discover_process_windows()
ready = len(windows) > 0
return {
"mode": "real-loggedin-status",
"ready": ready,
"workspace": str(Path(workspace)),
"process_name_hints": list(PROCESS_NAME_HINTS),
"process_candidates": [window.to_dict() for window in windows],
"next_step": (
"run real_lab_scan_windows, then capture UI selector profile"
if ready
else "open iSphere manually, log in, then rerun real_lab_status"
),
}
def scan_windows(workspace: str = str(DEFAULT_WORKSPACE), save_snapshot: bool = False) -> dict[str, Any]:
windows = [window.to_dict() for window in _discover_process_windows()]
result: dict[str, Any] = {
"mode": "real-loggedin-window-scan",
"workspace": str(Path(workspace)),
"windows": windows,
"count": len(windows),
"snapshot_saved": None,
"next_step": (
"copy real install/user data and run UI Automation tree capture"
if windows
else "open and log in to iSphere on this machine, then rerun scan"
),
}
if save_snapshot:
snapshot_path = _write_snapshot(Path(workspace), result)
result["snapshot_saved"] = str(snapshot_path)
return result
def win_helper_paths() -> dict[str, Any]:
root = _repo_root()
helper = root / DEFAULT_WIN_HELPER_EXE
build_script = root / "scripts" / "build-win-helper.ps1"
return {
"repo_root": str(root),
"helper_exe": str(helper),
"helper_exists": helper.exists(),
"build_script": str(build_script),
"build_script_exists": build_script.exists(),
"build_command": (
"powershell -NoProfile -ExecutionPolicy Bypass "
"-File scripts\\build-win-helper.ps1"
),
}
def build_win_helper(output_dir: str = "runs/win-helper") -> dict[str, Any]:
script = _repo_root() / "scripts" / "build-win-helper.ps1"
if not script.exists():
return {
"ok": False,
"error": "win_helper_build_script_missing",
"script": str(script),
}
completed = subprocess.run(
[
"powershell",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-File",
str(script),
"-OutputDir",
output_dir,
],
cwd=_repo_root(),
text=True,
encoding="utf-8",
errors="replace",
capture_output=True,
timeout=60,
)
parsed = _parse_json(completed.stdout)
if isinstance(parsed, dict):
parsed.setdefault("ok", completed.returncode == 0)
parsed["returncode"] = completed.returncode
if completed.stderr.strip():
parsed["stderr"] = completed.stderr
return parsed
return {
"ok": False,
"returncode": completed.returncode,
"stdout": completed.stdout,
"stderr": completed.stderr,
"error": "no_json_output",
}
def win_helper_version(helper_exe: str | None = None) -> dict[str, Any]:
return _run_win_helper("version", {}, helper_exe=helper_exe)
def win_helper_self_check(helper_exe: str | None = None) -> dict[str, Any]:
return _run_win_helper("self_check", {}, helper_exe=helper_exe)
def win_helper_scan_windows(
include_all_visible: bool = False,
helper_exe: str | None = None,
) -> dict[str, Any]:
return _run_win_helper(
"scan_windows",
{"include_all_visible": include_all_visible},
helper_exe=helper_exe,
)
def win_helper_dump_uia(
hwnd: str,
max_depth: int = 8,
include_text: bool = True,
max_children: int = 200,
helper_exe: str | None = None,
) -> dict[str, Any]:
return _run_win_helper(
"dump_uia",
{
"hwnd": hwnd,
"max_depth": max_depth,
"include_text": include_text,
"max_children": max_children,
},
helper_exe=helper_exe,
)
def _run_win_helper(
op: str,
args: dict[str, Any],
helper_exe: str | None = None,
timeout_ms: int = 5000,
) -> dict[str, Any]:
helper = _resolve_win_helper_exe(helper_exe)
if not helper.exists():
paths = win_helper_paths()
return {
"ok": False,
"error": "win_helper_not_built",
"helper_exe": str(helper),
"build_command": paths["build_command"],
"next_step": "run real_lab_win_helper_build or execute the build command",
}
request = {
"protocol": WIN_HELPER_PROTOCOL,
"request_id": f"python-{op}",
"op": op,
"timeout_ms": timeout_ms,
"args": args,
}
try:
completed = subprocess.run(
[str(helper), "--json"],
cwd=_repo_root(),
input=json.dumps(request, ensure_ascii=False),
text=True,
encoding="utf-8",
errors="replace",
capture_output=True,
timeout=max(1, timeout_ms / 1000 + 5),
)
except (OSError, subprocess.SubprocessError) as exc:
return {
"ok": False,
"error": "win_helper_failed",
"message": str(exc),
"helper_exe": str(helper),
}
parsed = _parse_json(completed.stdout)
if isinstance(parsed, dict):
parsed["returncode"] = completed.returncode
if completed.stderr.strip():
parsed["stderr"] = completed.stderr
return parsed
return {
"ok": False,
"returncode": completed.returncode,
"stdout": completed.stdout,
"stderr": completed.stderr,
"error": "no_json_output",
}
def _discover_process_windows() -> list[ProcessWindow]:
script = """
Get-Process |
Where-Object {
$_.ProcessName -match 'IMPlatform|IMPP|iSphere|importal' -or
$_.MainWindowTitle -match 'iSphere|IMPlatform|协同|门户|消息'
} |
Select-Object ProcessName,Id,MainWindowTitle,MainWindowHandle |
ConvertTo-Json -Depth 3 -Compress
"""
try:
completed = subprocess.run(
["powershell", "-NoProfile", "-Command", script],
text=True,
encoding="utf-8",
errors="replace",
capture_output=True,
timeout=10,
)
except (OSError, subprocess.SubprocessError):
return []
if completed.returncode != 0 or not completed.stdout.strip():
return []
try:
parsed = json.loads(completed.stdout)
except json.JSONDecodeError:
return []
items = parsed if isinstance(parsed, list) else [parsed]
windows: list[ProcessWindow] = []
for item in items:
if not isinstance(item, dict):
continue
try:
process_name = str(item.get("ProcessName") or "")
process_id = int(item.get("Id") or 0)
main_window_title = str(item.get("MainWindowTitle") or "")
main_window_handle = int(item.get("MainWindowHandle") or 0)
except (TypeError, ValueError):
continue
if not process_name or process_id <= 0:
continue
windows.append(
ProcessWindow(
process_name=process_name,
process_id=process_id,
main_window_title=main_window_title,
main_window_handle=main_window_handle,
)
)
return windows
def _write_snapshot(workspace: Path, payload: dict[str, Any]) -> Path:
snapshot_dir = workspace / "window-snapshots"
snapshot_dir.mkdir(parents=True, exist_ok=True)
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
snapshot_path = snapshot_dir / f"top-level-windows-{stamp}.json"
snapshot_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
return snapshot_path
def _parse_json(stdout: str) -> Any | None:
stripped = stdout.strip()
if stripped:
try:
return json.loads(stripped)
except json.JSONDecodeError:
pass
for line in reversed(stdout.splitlines()):
candidate = line.strip()
if not candidate:
continue
try:
return json.loads(candidate)
except json.JSONDecodeError:
continue
return None
def _resolve_win_helper_exe(helper_exe: str | None) -> Path:
path = Path(helper_exe) if helper_exe else DEFAULT_WIN_HELPER_EXE
if not path.is_absolute():
path = _repo_root() / path
return path
def _repo_root() -> Path:
return Path(__file__).resolve().parents[2]

View File

@@ -1,528 +0,0 @@
from __future__ import annotations
from mcp.server.fastmcp import FastMCP
from . import mcp_api
mcp = FastMCP(
"isphere-ai-bridge",
instructions=(
"MCP service for authorized iSphere message handoff. "
"Production-facing tools are read-only message access and draft-only output. "
"offline_lab_* tools run a local simulated helper state machine for reverse-engineering "
"and UI-automation experiments; they do not log in, contact XMPP, send real messages, "
"mark read, extract tokens, or mutate the original iSphere client. "
"native_lab_* tools call the compiled .NET Framework helper, which can inspect a copied "
"IMPlatformClient.exe, probe no-login frmMain construction, and run the same offline "
"command protocol against helper state. "
"real_lab_* tools are for a future user-logged-in client path: current tools only expose "
"planning, required-file intake, action contracts, and read-only process/window discovery."
),
)
@mcp.tool()
def discover_message_sources(base: str | None = None) -> dict:
"""Find candidate local iSphere MsgLib.db paths without opening or modifying them."""
return mcp_api.discover_message_sources(base)
@mcp.tool()
def create_mock_source(output: str = "runs/mcp/MsgLib.db", force: bool = False) -> dict:
"""Create a mock MsgLib.db for MCP testing without using real client data."""
return mcp_api.create_mock_source(output, force)
@mcp.tool()
def read_messages(db_path: str, limit: int = 20) -> dict:
"""Read normalized messages from a MsgLib.db using a read-only connection."""
return mcp_api.read_messages(db_path, limit)
@mcp.tool()
def read_recent_conversations(db_path: str, limit: int = 20) -> dict:
"""Read recent conversations from a MsgLib.db using a read-only connection."""
return mcp_api.read_recent_conversations(db_path, limit)
@mcp.tool()
def generate_draft_preview(db_path: str, limit: int = 5) -> dict:
"""Generate draft-only replies in memory; does not write to iSphere or send messages."""
return mcp_api.generate_draft_preview(db_path, limit)
@mcp.tool()
def queue_drafts_for_review(
db_path: str,
queue_path: str = "runs/mcp/drafts.jsonl",
audit_path: str = "runs/mcp/audit.jsonl",
limit: int = 5,
) -> dict:
"""Append draft-only replies to a review queue and redacted audit log."""
return mcp_api.queue_drafts_for_review(db_path, queue_path, audit_path, limit)
@mcp.tool()
def read_audit_events(audit_path: str = "runs/mcp/audit.jsonl", limit: int = 20) -> dict:
"""Read redacted MCP audit events; audit entries do not contain draft text."""
return mcp_api.read_audit_events(audit_path, limit)
@mcp.tool()
def real_lab_plan_tree() -> dict:
"""Return the implementation planning tree for the user-logged-in real iSphere MCP layer."""
return mcp_api.real_lab_plan_tree()
@mcp.tool()
def real_lab_required_files() -> dict:
"""Return the exact real install/user-data/window evidence checklist needed later."""
return mcp_api.real_lab_required_files()
@mcp.tool()
def real_lab_action_contract() -> dict:
"""Return the controlled action contract and approval gates for real logged-in operation."""
return mcp_api.real_lab_action_contract()
@mcp.tool()
def real_lab_status(workspace: str = "runs/real-loggedin-lab") -> dict:
"""Safely check whether a likely logged-in iSphere/IMPlatform window is currently visible."""
return mcp_api.real_lab_status(workspace=workspace)
@mcp.tool()
def real_lab_scan_windows(
workspace: str = "runs/real-loggedin-lab",
save_snapshot: bool = False,
) -> dict:
"""Read top-level candidate process/window metadata without clicking or sending input."""
return mcp_api.real_lab_scan_windows(workspace=workspace, save_snapshot=save_snapshot)
@mcp.tool()
def real_lab_win_helper_version(helper_exe: str | None = None) -> dict:
"""Call the C# WinHelper version op through stdin/stdout JSON."""
return mcp_api.real_lab_win_helper_version(helper_exe=helper_exe)
@mcp.tool()
def real_lab_win_helper_self_check(helper_exe: str | None = None) -> dict:
"""Call the C# WinHelper self_check op without clicking, typing, or modifying iSphere."""
return mcp_api.real_lab_win_helper_self_check(helper_exe=helper_exe)
@mcp.tool()
def real_lab_win_helper_scan_windows(
include_all_visible: bool = False,
helper_exe: str | None = None,
) -> dict:
"""Call the C# WinHelper scan_windows op for read-only top-level window discovery."""
return mcp_api.real_lab_win_helper_scan_windows(
include_all_visible=include_all_visible,
helper_exe=helper_exe,
)
@mcp.tool()
def real_lab_win_helper_dump_uia(
hwnd: str,
max_depth: int = 8,
include_text: bool = True,
max_children: int = 200,
helper_exe: str | None = None,
) -> dict:
"""Call the C# WinHelper dump_uia op for read-only UI Automation tree capture."""
return mcp_api.real_lab_win_helper_dump_uia(
hwnd=hwnd,
max_depth=max_depth,
include_text=include_text,
max_children=max_children,
helper_exe=helper_exe,
)
@mcp.tool()
def offline_lab_start(
exe_path: str | None = None,
workspace: str = "runs/offline-lab",
reset: bool = False,
) -> dict:
"""Open or create a simulated offline iSphere main-window session."""
return mcp_api.offline_lab_start(exe_path=exe_path, workspace=workspace, reset=reset)
@mcp.tool()
def offline_lab_status(workspace: str = "runs/offline-lab") -> dict:
"""Return the current offline-lab main-window/session status."""
return mcp_api.offline_lab_status(workspace)
@mcp.tool()
def offline_lab_search_users(
query: str = "",
limit: int = 20,
workspace: str = "runs/offline-lab",
) -> dict:
"""Search simulated iSphere users without requiring login."""
return mcp_api.offline_lab_search_users(query=query, limit=limit, workspace=workspace)
@mcp.tool()
def offline_lab_open_chat(user_jid: str, workspace: str = "runs/offline-lab") -> dict:
"""Open a simulated direct chat panel for a user."""
return mcp_api.offline_lab_open_chat(user_jid=user_jid, workspace=workspace)
@mcp.tool()
def offline_lab_send_message(
user_jid: str,
text: str,
workspace: str = "runs/offline-lab",
) -> dict:
"""Simulate sending a text message into the offline chat UI state."""
return mcp_api.offline_lab_send_message(user_jid=user_jid, text=text, workspace=workspace)
@mcp.tool()
def offline_lab_receive_message(
user_jid: str,
text: str,
workspace: str = "runs/offline-lab",
) -> dict:
"""Simulate receiving a text message into the offline chat UI state."""
return mcp_api.offline_lab_receive_message(user_jid=user_jid, text=text, workspace=workspace)
@mcp.tool()
def offline_lab_send_file(
user_jid: str,
file_path: str,
file_name: str | None = None,
workspace: str = "runs/offline-lab",
) -> dict:
"""Simulate sending a file attachment through the offline chat UI state."""
return mcp_api.offline_lab_send_file(
user_jid=user_jid,
file_path=file_path,
file_name=file_name,
workspace=workspace,
)
@mcp.tool()
def offline_lab_receive_file(
user_jid: str,
file_name: str,
size: int = 0,
workspace: str = "runs/offline-lab",
) -> dict:
"""Simulate receiving a file attachment through the offline chat UI state."""
return mcp_api.offline_lab_receive_file(
user_jid=user_jid,
file_name=file_name,
size=size,
workspace=workspace,
)
@mcp.tool()
def offline_lab_read_conversation(
user_jid: str,
limit: int = 50,
workspace: str = "runs/offline-lab",
) -> dict:
"""Read messages and file events from a simulated offline conversation."""
return mcp_api.offline_lab_read_conversation(user_jid=user_jid, limit=limit, workspace=workspace)
@mcp.tool()
def native_lab_helper_paths() -> dict:
"""Return native helper paths, build command, and default iSphere copy location."""
return mcp_api.native_lab_helper_paths()
@mcp.tool()
def native_lab_build_helper(output_dir: str = "runs/native-helper") -> dict:
"""Compile the .NET Framework x86 helper used to inspect/control an iSphere copy."""
return mcp_api.native_lab_build_helper(output_dir=output_dir)
@mcp.tool()
def native_lab_inspect(client_dir: str | None = None, helper_exe: str | None = None) -> dict:
"""Reflect over copied IMPlatformClient.exe and verify reversed offline entry points."""
return mcp_api.native_lab_inspect(client_dir=client_dir, helper_exe=helper_exe)
@mcp.tool()
def native_lab_probe_main_window(
client_dir: str | None = None,
helper_exe: str | None = None,
construct: bool = False,
probe_chat: bool = False,
probe_actions: bool = False,
probe_files: bool = False,
probe_send_text: bool = False,
probe_receive_text: bool = False,
probe_send_file: bool = False,
probe_receive_file: bool = False,
user_jid: str = "mock-bob@offline.lab",
text: str = "native helper direct chat probe",
file_path: str | None = None,
file_name: str | None = None,
file_size: int = 0,
timeout_ms: int = 15000,
) -> dict:
"""Bootstrap copied iSphere state and optionally construct frmMain/frmP2PChat plus text/file hooks without logging in."""
return mcp_api.native_lab_probe_main_window(
client_dir=client_dir,
helper_exe=helper_exe,
construct=construct,
probe_chat=probe_chat,
probe_actions=probe_actions,
probe_files=probe_files,
probe_send_text=probe_send_text,
probe_receive_text=probe_receive_text,
probe_send_file=probe_send_file,
probe_receive_file=probe_receive_file,
user_jid=user_jid,
text=text,
file_path=file_path,
file_name=file_name,
file_size=file_size,
timeout_ms=timeout_ms,
)
@mcp.tool()
def native_lab_direct_open_main_window(
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict:
"""Construct the copied iSphere frmMain without login in a short-lived native helper run."""
return mcp_api.native_lab_direct_open_main_window(
client_dir=client_dir,
helper_exe=helper_exe,
timeout_ms=timeout_ms,
)
@mcp.tool()
def native_lab_direct_send_message(
user_jid: str,
text: str,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict:
"""Invoke the copied-client frmP2PChat.SaveAndShowMessage hook without login."""
return mcp_api.native_lab_direct_send_message(
user_jid=user_jid,
text=text,
client_dir=client_dir,
helper_exe=helper_exe,
timeout_ms=timeout_ms,
)
@mcp.tool()
def native_lab_direct_receive_message(
user_jid: str,
text: str,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict:
"""Invoke the copied-client frmP2PChat.ProccessRecvMessage hook without login."""
return mcp_api.native_lab_direct_receive_message(
user_jid=user_jid,
text=text,
client_dir=client_dir,
helper_exe=helper_exe,
timeout_ms=timeout_ms,
)
@mcp.tool()
def native_lab_direct_send_file(
user_jid: str,
file_path: str | None = None,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict:
"""Construct the copied-client UC_FileSend control without login."""
return mcp_api.native_lab_direct_send_file(
user_jid=user_jid,
file_path=file_path,
client_dir=client_dir,
helper_exe=helper_exe,
timeout_ms=timeout_ms,
)
@mcp.tool()
def native_lab_direct_receive_file(
user_jid: str,
file_name: str,
size: int = 0,
client_dir: str | None = None,
helper_exe: str | None = None,
timeout_ms: int = 15000,
) -> dict:
"""Construct the copied-client UC_FileReceive control without login."""
return mcp_api.native_lab_direct_receive_file(
user_jid=user_jid,
file_name=file_name,
size=size,
client_dir=client_dir,
helper_exe=helper_exe,
timeout_ms=timeout_ms,
)
@mcp.tool()
def native_lab_start(
client_dir: str | None = None,
workspace: str = "runs/native-offline-lab",
helper_exe: str | None = None,
reset: bool = False,
) -> dict:
"""Start the native-helper offline lab session after inspecting an iSphere copy."""
return mcp_api.native_lab_start(
client_dir=client_dir,
workspace=workspace,
helper_exe=helper_exe,
reset=reset,
)
@mcp.tool()
def native_lab_status(workspace: str = "runs/native-offline-lab", helper_exe: str | None = None) -> dict:
"""Return native-helper offline session status."""
return mcp_api.native_lab_status(workspace=workspace, helper_exe=helper_exe)
@mcp.tool()
def native_lab_search_users(
query: str = "",
limit: int = 20,
workspace: str = "runs/native-offline-lab",
helper_exe: str | None = None,
) -> dict:
"""Search mock users through the native-helper command protocol."""
return mcp_api.native_lab_search_users(
query=query,
limit=limit,
workspace=workspace,
helper_exe=helper_exe,
)
@mcp.tool()
def native_lab_open_chat(
user_jid: str,
workspace: str = "runs/native-offline-lab",
helper_exe: str | None = None,
) -> dict:
"""Open a simulated direct chat through the native-helper command protocol."""
return mcp_api.native_lab_open_chat(
user_jid=user_jid,
workspace=workspace,
helper_exe=helper_exe,
)
@mcp.tool()
def native_lab_send_message(
user_jid: str,
text: str,
workspace: str = "runs/native-offline-lab",
helper_exe: str | None = None,
) -> dict:
"""Simulate outgoing text through the native-helper command protocol."""
return mcp_api.native_lab_send_message(
user_jid=user_jid,
text=text,
workspace=workspace,
helper_exe=helper_exe,
)
@mcp.tool()
def native_lab_receive_message(
user_jid: str,
text: str,
workspace: str = "runs/native-offline-lab",
helper_exe: str | None = None,
) -> dict:
"""Simulate incoming text through the native-helper command protocol."""
return mcp_api.native_lab_receive_message(
user_jid=user_jid,
text=text,
workspace=workspace,
helper_exe=helper_exe,
)
@mcp.tool()
def native_lab_send_file(
user_jid: str,
file_path: str,
file_name: str | None = None,
workspace: str = "runs/native-offline-lab",
helper_exe: str | None = None,
) -> dict:
"""Simulate outgoing file through the native-helper command protocol."""
return mcp_api.native_lab_send_file(
user_jid=user_jid,
file_path=file_path,
file_name=file_name,
workspace=workspace,
helper_exe=helper_exe,
)
@mcp.tool()
def native_lab_receive_file(
user_jid: str,
file_name: str,
size: int = 0,
workspace: str = "runs/native-offline-lab",
helper_exe: str | None = None,
) -> dict:
"""Simulate incoming file through the native-helper command protocol."""
return mcp_api.native_lab_receive_file(
user_jid=user_jid,
file_name=file_name,
size=size,
workspace=workspace,
helper_exe=helper_exe,
)
@mcp.tool()
def native_lab_read_conversation(
user_jid: str,
limit: int = 50,
workspace: str = "runs/native-offline-lab",
helper_exe: str | None = None,
) -> dict:
"""Read a simulated conversation through the native-helper command protocol."""
return mcp_api.native_lab_read_conversation(
user_jid=user_jid,
limit=limit,
workspace=workspace,
helper_exe=helper_exe,
)
def main() -> None:
mcp.run("stdio")
if __name__ == "__main__":
main()

View File

@@ -1,90 +0,0 @@
from __future__ import annotations
import json
import tempfile
import unittest
from contextlib import redirect_stdout
from io import StringIO
from pathlib import Path
from isphere_ai_bridge.cli import main
from isphere_ai_bridge.draft_queue import DraftQueue
from isphere_ai_bridge.models import Draft
from isphere_ai_bridge.mock_db import create_mock_msglib
class DraftQueueTests(unittest.TestCase):
def test_append_drafts_writes_pending_queue_and_redacted_audit_event(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
queue_path = Path(tmp) / "drafts.jsonl"
audit_path = Path(tmp) / "audit.jsonl"
queue = DraftQueue(
queue_path=queue_path,
audit_path=audit_path,
clock=lambda: "2026-07-05T01:40:00+08:00",
)
draft = Draft(
message_id="msg-001",
conversation_id="alice@example",
draft_text="Draft only. Please review before sending.",
source_message_ref="MsgLib.db#tblPersonMsg:msg-001",
)
entries = queue.append_drafts([draft])
self.assertEqual(1, len(entries))
self.assertEqual("pending_review", entries[0]["status"])
self.assertEqual("msg-001", entries[0]["message_id"])
self.assertEqual("Draft only. Please review before sending.", entries[0]["draft_text"])
queue_lines = queue_path.read_text(encoding="utf-8").splitlines()
self.assertEqual(1, len(queue_lines))
queue_record = json.loads(queue_lines[0])
self.assertEqual(entries[0]["draft_id"], queue_record["draft_id"])
self.assertEqual("pending_review", queue_record["status"])
audit_lines = audit_path.read_text(encoding="utf-8").splitlines()
self.assertEqual(1, len(audit_lines))
audit_record = json.loads(audit_lines[0])
self.assertEqual("draft_created", audit_record["event_type"])
self.assertEqual(entries[0]["draft_id"], audit_record["draft_id"])
self.assertIn("content_sha256", audit_record)
self.assertNotIn("draft_text", audit_record)
def test_queue_drafts_cli_appends_drafts_and_audit_events(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
db_path = create_mock_msglib(root / "MsgLib.db")
queue_path = root / "drafts.jsonl"
audit_path = root / "audit.jsonl"
stdout = StringIO()
with redirect_stdout(stdout):
exit_code = main(
[
"queue-drafts",
"--db",
str(db_path),
"--queue",
str(queue_path),
"--audit",
str(audit_path),
"--limit",
"2",
]
)
self.assertEqual(0, exit_code)
output = json.loads(stdout.getvalue())
self.assertEqual(2, output["queued"])
self.assertEqual(2, len(queue_path.read_text(encoding="utf-8").splitlines()))
audit_records = [
json.loads(line)
for line in audit_path.read_text(encoding="utf-8").splitlines()
]
self.assertEqual(2, len(audit_records))
self.assertNotIn("draft_text", audit_records[0])
if __name__ == "__main__":
unittest.main()

View File

@@ -1,73 +0,0 @@
from __future__ import annotations
import sqlite3
import tempfile
import unittest
from pathlib import Path
from isphere_ai_bridge.draft import generate_drafts
from isphere_ai_bridge.local_db import ISphereLocalDbReader, open_readonly_connection
from isphere_ai_bridge.locator import discover_msglib_databases
from isphere_ai_bridge.mock_db import create_mock_msglib
class LocalDbPocTests(unittest.TestCase):
def test_mock_database_reads_normalized_messages(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
db_path = create_mock_msglib(Path(tmp) / "MsgLib.db")
reader = ISphereLocalDbReader(db_path)
probe = reader.probe()
self.assertTrue(probe["can_open_readonly"])
self.assertIn("tblPersonMsg", probe["message_tables"])
self.assertIn("tblMsgGroupPersonMsg", probe["message_tables"])
messages = reader.read_messages(limit=10)
self.assertEqual(4, len(messages))
self.assertEqual("sys-001", messages[0].message_id)
self.assertEqual("system", messages[0].conversation_type)
self.assertEqual("grp-001", messages[1].message_id)
self.assertEqual("group", messages[1].conversation_type)
self.assertEqual("p2p-002", messages[2].message_id)
self.assertEqual("direct", messages[2].conversation_type)
self.assertIn("#tblPersonMsg:p2p-002", messages[2].raw_ref)
def test_connection_is_read_only(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
db_path = create_mock_msglib(Path(tmp) / "MsgLib.db")
with open_readonly_connection(db_path) as con:
with self.assertRaises(sqlite3.OperationalError):
con.execute("create table should_not_write (id int)")
def test_discovers_expected_account_msglib_path(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
base = Path(tmp) / "isphere"
db_path = (
base
/ "connection.10083"
/ "27000000"
/ "Account"
/ "EMP001"
/ "MsgLib.db"
)
create_mock_msglib(db_path)
candidates = discover_msglib_databases(base)
self.assertEqual(1, len(candidates))
self.assertTrue(candidates[0].exists)
self.assertEqual("EMP001", candidates[0].account_id)
self.assertEqual(db_path, candidates[0].path)
def test_generates_draft_only_outputs(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
db_path = create_mock_msglib(Path(tmp) / "MsgLib.db")
messages = ISphereLocalDbReader(db_path).read_messages(limit=2)
drafts = generate_drafts(messages)
self.assertEqual(2, len(drafts))
self.assertEqual("draft-only", drafts[0].status)
self.assertIn("Draft only", drafts[0].draft_text)
if __name__ == "__main__":
unittest.main()

View File

@@ -1,51 +0,0 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from isphere_ai_bridge import mcp_api
class MCPApiTests(unittest.TestCase):
def test_mcp_api_reads_mock_messages_and_queues_drafts(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
db_path = root / "MsgLib.db"
queue_path = root / "drafts.jsonl"
audit_path = root / "audit.jsonl"
created = mcp_api.create_mock_source(str(db_path))
self.assertEqual(str(db_path), created["created"])
messages = mcp_api.read_messages(str(db_path), limit=2)
self.assertEqual(2, messages["count"])
self.assertEqual("local-db", messages["messages"][0]["source"])
queued = mcp_api.queue_drafts_for_review(
str(db_path),
queue_path=str(queue_path),
audit_path=str(audit_path),
limit=2,
)
self.assertEqual(2, queued["queued"])
self.assertEqual("pending_review", queued["status"])
audit = mcp_api.read_audit_events(str(audit_path), limit=5)
self.assertEqual(2, audit["count"])
self.assertIn("content_sha256", audit["events"][0])
self.assertNotIn("draft_text", audit["events"][0])
def test_limit_is_bounded_for_mcp_tools(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
db_path = Path(tmp) / "MsgLib.db"
mcp_api.create_mock_source(str(db_path))
messages = mcp_api.read_messages(str(db_path), limit=1000)
self.assertEqual(mcp_api.MAX_LIMIT, messages["limit"])
if __name__ == "__main__":
unittest.main()

View File

@@ -1,159 +0,0 @@
from __future__ import annotations
import os
import tempfile
import unittest
from pathlib import Path
from isphere_ai_bridge import native_helper
def _has_framework_csc() -> bool:
windir = Path(os.environ.get("WINDIR", r"C:\Windows"))
return (
windir / "Microsoft.NET" / "Framework64" / "v4.0.30319" / "csc.exe"
).exists() or (
windir / "Microsoft.NET" / "Framework" / "v4.0.30319" / "csc.exe"
).exists()
class NativeHelperTests(unittest.TestCase):
def test_helper_paths_are_actionable(self) -> None:
paths = native_helper.helper_paths()
self.assertIn("build_command", paths)
self.assertTrue(paths["build_script_exists"])
self.assertTrue(paths["default_client_dir_exists"])
@unittest.skipUnless(_has_framework_csc(), ".NET Framework csc.exe is required")
def test_native_helper_build_inspect_and_offline_protocol(self) -> None:
build = native_helper.build_native_helper()
self.assertTrue(build["ok"], build)
self.assertEqual(0, build["returncode"])
self.assertTrue(Path(build["output"]).exists())
self.assertTrue(Path(build["resource_stub"]).exists())
self.assertEqual("x86", build["platform"])
inspect = native_helper.native_lab_inspect()
self.assertTrue(inspect["ok"], inspect)
self.assertTrue(inspect["required_ready"])
type_names = {item["name"] for item in inspect["types"]}
self.assertIn("IMPP.Client.frmMain", type_names)
self.assertIn("IMPP.Client.Business.ChatManager.SingleChat.frmP2PChat", type_names)
bootstrap = native_helper.native_lab_probe_main_window()
self.assertTrue(bootstrap["ok"], bootstrap)
bootstrap_steps = {item["name"]: item["ok"] for item in bootstrap["steps"]}
self.assertTrue(bootstrap_steps["imppmanager_update_user"])
self.assertFalse(bootstrap["main_window_constructed"])
constructed = native_helper.native_lab_probe_main_window(
construct=True,
probe_chat=True,
probe_actions=True,
probe_files=True,
user_jid="mock-bob@offline.lab",
text="native helper direct test",
file_name="incoming-probe-native.txt",
file_size=321,
timeout_ms=15000,
)
self.assertTrue(constructed["ok"], constructed)
self.assertTrue(constructed["main_window_constructed"])
self.assertTrue(constructed["chat_model_constructed"])
self.assertTrue(constructed["chat_window_constructed"])
self.assertTrue(constructed["send_text_invoked"])
self.assertTrue(constructed["receive_text_invoked"])
self.assertTrue(constructed["file_send_control_constructed"])
self.assertTrue(constructed["file_receive_control_constructed"])
construct_steps = {item["name"]: item["ok"] for item in constructed["steps"]}
self.assertTrue(construct_steps["patch_problematic_ui_methods"])
self.assertTrue(construct_steps["instantiate_frmMain"])
self.assertTrue(construct_steps["instantiate_frmP2PChat"])
self.assertTrue(construct_steps["invoke_frmP2PChat_SaveAndShowMessage"])
self.assertTrue(construct_steps["invoke_frmP2PChat_ProccessRecvMessage"])
self.assertTrue(construct_steps["instantiate_UC_FileSend"])
self.assertTrue(construct_steps["instantiate_UC_FileReceive"])
direct_open = native_helper.native_lab_direct_open_main_window(timeout_ms=15000)
self.assertTrue(direct_open["ok"], direct_open)
self.assertTrue(direct_open["main_window_constructed"])
direct_send = native_helper.native_lab_direct_send_message(
"mock-bob@offline.lab",
"direct wrapper send",
timeout_ms=15000,
)
self.assertTrue(direct_send["ok"], direct_send)
self.assertTrue(direct_send["send_text_invoked"])
self.assertFalse(direct_send["receive_text_invoked"])
direct_receive = native_helper.native_lab_direct_receive_message(
"mock-bob@offline.lab",
"direct wrapper receive",
timeout_ms=15000,
)
self.assertTrue(direct_receive["ok"], direct_receive)
self.assertFalse(direct_receive["send_text_invoked"])
self.assertTrue(direct_receive["receive_text_invoked"])
direct_send_file = native_helper.native_lab_direct_send_file(
"mock-bob@offline.lab",
timeout_ms=15000,
)
self.assertTrue(direct_send_file["ok"], direct_send_file)
self.assertTrue(direct_send_file["file_send_control_constructed"])
self.assertFalse(direct_send_file["file_receive_control_constructed"])
direct_receive_file = native_helper.native_lab_direct_receive_file(
"mock-bob@offline.lab",
"direct-wrapper-incoming.txt",
size=222,
timeout_ms=15000,
)
self.assertTrue(direct_receive_file["ok"], direct_receive_file)
self.assertFalse(direct_receive_file["file_send_control_constructed"])
self.assertTrue(direct_receive_file["file_receive_control_constructed"])
with tempfile.TemporaryDirectory() as tmp:
workspace = str(Path(tmp) / "native-lab")
start = native_helper.native_lab_start(workspace=workspace, reset=True)
self.assertTrue(start["started"])
self.assertEqual("offline-lab-native-helper", start["mode"])
self.assertEqual("inspected", start["native_bridge"]["status"])
users = native_helper.native_lab_search_users("文件", workspace=workspace)
self.assertEqual(1, users["count"])
user_jid = users["users"][0]["jid"]
opened = native_helper.native_lab_open_chat(user_jid, workspace=workspace)
self.assertTrue(opened["opened"])
self.assertEqual("p2p-chat", opened["main_window"]["active_panel"])
sent = native_helper.native_lab_send_message(
user_jid,
"native helper test send",
workspace=workspace,
)
self.assertTrue(sent["sent"])
self.assertEqual("frmP2PChat.SaveAndShowMessage", sent["message"]["ui_hook"])
received_file = native_helper.native_lab_receive_file(
user_jid,
"incoming-native.txt",
size=789,
workspace=workspace,
)
self.assertTrue(received_file["received"])
self.assertEqual(
"frmP2PChat.ProccessRecvMessage -> UC_FileReceive",
received_file["file_event"]["ui_hook"],
)
conversation = native_helper.native_lab_read_conversation(user_jid, workspace=workspace)
self.assertEqual(2, conversation["conversation"]["message_count"])
if __name__ == "__main__":
unittest.main()

View File

@@ -1,91 +0,0 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from isphere_ai_bridge import mcp_api
class OfflineLabTests(unittest.TestCase):
def test_offline_lab_simulates_main_window_search_chat_and_files(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
workspace = str(Path(tmp) / "offline-lab")
start = mcp_api.offline_lab_start(workspace=workspace, reset=True)
self.assertTrue(start["started"])
self.assertEqual("offline-lab-simulated", start["mode"])
self.assertTrue(start["main_window"]["opened"])
users = mcp_api.offline_lab_search_users("文件", workspace=workspace)
self.assertEqual(1, users["count"])
user_jid = users["users"][0]["jid"]
opened = mcp_api.offline_lab_open_chat(user_jid, workspace=workspace)
self.assertTrue(opened["opened"])
self.assertEqual("p2p-chat", opened["main_window"]["active_panel"])
sent = mcp_api.offline_lab_send_message(
user_jid,
"这是一条离线模拟发送消息",
workspace=workspace,
)
self.assertTrue(sent["sent"])
self.assertEqual("outgoing", sent["message"]["direction"])
self.assertEqual("frmP2PChat.SaveAndShowMessage", sent["message"]["ui_hook"])
received = mcp_api.offline_lab_receive_message(
user_jid,
"这是一条离线模拟接收消息",
workspace=workspace,
)
self.assertTrue(received["received"])
self.assertEqual("incoming", received["message"]["direction"])
self.assertEqual("frmP2PChat.ProccessRecvMessage", received["message"]["ui_hook"])
outbound_file = Path(tmp) / "demo.txt"
outbound_file.write_text("file payload", encoding="utf-8")
send_file = mcp_api.offline_lab_send_file(
user_jid,
str(outbound_file),
workspace=workspace,
)
self.assertTrue(send_file["sent"])
self.assertEqual("file", send_file["file_event"]["content_type"])
self.assertEqual(True, send_file["file_event"]["attachments"][0]["exists"])
self.assertEqual(
"frmP2PChat.SendTcpFiles/SendOfflineFiles -> UC_FileSend",
send_file["file_event"]["ui_hook"],
)
receive_file = mcp_api.offline_lab_receive_file(
user_jid,
"incoming-report.pdf",
size=4096,
workspace=workspace,
)
self.assertTrue(receive_file["received"])
self.assertEqual(4096, receive_file["file_event"]["attachments"][0]["size"])
self.assertEqual(
"frmP2PChat.ProccessRecvMessage -> UC_FileReceive",
receive_file["file_event"]["ui_hook"],
)
conversation = mcp_api.offline_lab_read_conversation(user_jid, workspace=workspace)
self.assertEqual(4, conversation["conversation"]["message_count"])
self.assertEqual(4, len(conversation["conversation"]["messages"]))
status = mcp_api.offline_lab_status(workspace=workspace)
self.assertTrue(status["started"])
self.assertEqual(4, status["message_count"])
def test_offline_lab_status_before_start_is_actionable(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
status = mcp_api.offline_lab_status(workspace=str(Path(tmp) / "missing"))
self.assertFalse(status["started"])
self.assertEqual("call offline_lab_start", status["next_step"])
if __name__ == "__main__":
unittest.main()

View File

@@ -1,92 +0,0 @@
from __future__ import annotations
import asyncio
import unittest
from isphere_ai_bridge import real_lab
class RealLoggedInLabTests(unittest.TestCase):
def test_plan_tree_reports_ready_and_blocked_phases(self) -> None:
tree = real_lab.plan_tree()
self.assertEqual("real-loggedin-planning", tree["mode"])
self.assertEqual("A", tree["phases"][0]["id"])
self.assertEqual("ready-now", tree["phases"][0]["status"])
self.assertTrue(
any(phase["status"] == "blocked-waiting-real-data" for phase in tree["phases"])
)
def test_required_files_checklist_has_expected_sections(self) -> None:
checklist = real_lab.required_files_checklist()
section_ids = {section["id"] for section in checklist["sections"]}
self.assertTrue({"install_dir", "user_data", "logged_in_window"}.issubset(section_ids))
self.assertEqual("copy-later", checklist["recommended_timing"])
def test_action_contract_requires_approval_for_send_and_file_upload(self) -> None:
contract = real_lab.action_contract()
actions = {action["name"]: action for action in contract["actions"]}
self.assertFalse(actions["real_lab_search_contacts"]["requires_approval"])
self.assertTrue(actions["real_lab_send_after_approval"]["requires_approval"])
self.assertTrue(actions["real_lab_send_file_after_approval"]["requires_approval"])
self.assertEqual("no-login-automation", contract["hard_limits"][0])
def test_status_and_scan_windows_are_structured_without_real_client(self) -> None:
status = real_lab.status()
scan = real_lab.scan_windows()
self.assertIn("ready", status)
self.assertIn("process_candidates", status)
self.assertIn("next_step", status)
self.assertIn("windows", scan)
self.assertIn("next_step", scan)
def test_win_helper_version_and_self_check_are_structured(self) -> None:
built = real_lab.build_win_helper()
self.assertTrue(built["ok"], built)
version = real_lab.win_helper_version()
self.assertTrue(version["ok"], version)
self.assertEqual("version", version["op"])
self.assertEqual("ISphereWinHelper", version["data"]["helper_name"])
self_check = real_lab.win_helper_self_check()
self.assertTrue(self_check["ok"], self_check)
self.assertEqual("self_check", self_check["op"])
self.assertIn("uia_available", self_check["data"])
def test_win_helper_scan_and_missing_uia_dump_are_structured(self) -> None:
built = real_lab.build_win_helper()
self.assertTrue(built["ok"], built)
scan = real_lab.win_helper_scan_windows(include_all_visible=True)
self.assertTrue(scan["ok"], scan)
self.assertEqual("scan_windows", scan["op"])
self.assertIsInstance(scan["data"]["windows"], list)
dump = real_lab.win_helper_dump_uia("0x0")
self.assertFalse(dump["ok"], dump)
self.assertIn(dump["error"]["code"], {"WINDOW_NOT_FOUND", "UIA_DUMP_FAILED"})
def test_real_lab_tools_are_registered(self) -> None:
from isphere_ai_bridge.server import mcp
names = [tool.name for tool in asyncio.run(mcp.list_tools())]
required = {
"real_lab_plan_tree",
"real_lab_required_files",
"real_lab_action_contract",
"real_lab_status",
"real_lab_scan_windows",
"real_lab_win_helper_version",
"real_lab_win_helper_self_check",
"real_lab_win_helper_scan_windows",
"real_lab_win_helper_dump_uia",
}
self.assertTrue(required.issubset(set(names)))
if __name__ == "__main__":
unittest.main()

View File

@@ -1,130 +0,0 @@
import json
import subprocess
import unittest
from pathlib import Path
REPO = Path(__file__).resolve().parents[1]
BUILD_SCRIPT = REPO / "scripts" / "build-win-helper.ps1"
HELPER_EXE = REPO / "runs" / "win-helper" / "ISphereWinHelper.exe"
def build_helper():
result = subprocess.run(
[
"powershell",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-File",
str(BUILD_SCRIPT),
],
cwd=REPO,
text=True,
encoding="utf-8",
errors="replace",
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=60,
)
if result.returncode != 0:
raise AssertionError(result.stdout + result.stderr)
def invoke_helper(payload):
if isinstance(payload, str):
stdin = payload
else:
stdin = json.dumps(payload, ensure_ascii=False)
result = subprocess.run(
[str(HELPER_EXE), "--json"],
cwd=REPO,
input=stdin,
text=True,
encoding="utf-8",
errors="replace",
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=30,
)
try:
decoded = json.loads(result.stdout)
except json.JSONDecodeError as exc:
raise AssertionError(f"stdout was not JSON: {result.stdout!r}; stderr={result.stderr!r}") from exc
return result, decoded
class WinHelperBuildTests(unittest.TestCase):
def test_build_script_creates_helper_exe(self):
if HELPER_EXE.exists():
HELPER_EXE.unlink()
build_helper()
self.assertTrue(HELPER_EXE.exists(), f"helper exe missing: {HELPER_EXE}")
class WinHelperProtocolTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
build_helper()
def request(self, op, args=None):
result, decoded = invoke_helper(
{
"protocol": "isphere.helper.v1",
"request_id": f"test-{op}",
"op": op,
"timeout_ms": 5000,
"args": args or {},
}
)
self.assertEqual(result.returncode, 0, result.stderr)
return decoded
def test_version_returns_contract_metadata(self):
decoded = self.request("version")
self.assertTrue(decoded["ok"])
self.assertEqual(decoded["protocol"], "isphere.helper.v1")
self.assertEqual(decoded["request_id"], "test-version")
self.assertEqual(decoded["op"], "version")
self.assertEqual(decoded["data"]["helper_name"], "ISphereWinHelper")
self.assertEqual(decoded["data"]["protocol"], "isphere.helper.v1")
def test_self_check_reports_desktop_and_uia_status(self):
decoded = self.request("self_check")
self.assertTrue(decoded["ok"])
self.assertIn("desktop_access", decoded["data"])
self.assertIn("uia_available", decoded["data"])
def test_scan_windows_returns_json_window_list(self):
decoded = self.request("scan_windows", {"include_all_visible": True})
self.assertTrue(decoded["ok"])
self.assertIsInstance(decoded["data"]["windows"], list)
def test_unsupported_op_returns_structured_error(self):
decoded = self.request("does_not_exist")
self.assertFalse(decoded["ok"])
self.assertEqual(decoded["error"]["code"], "UNSUPPORTED_OP")
def test_invalid_json_returns_structured_error(self):
result, decoded = invoke_helper("{not json")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertFalse(decoded["ok"])
self.assertEqual(decoded["protocol"], "isphere.helper.v1")
self.assertEqual(decoded["error"]["code"], "INVALID_REQUEST")
def test_dump_uia_with_missing_window_returns_structured_error(self):
decoded = self.request("dump_uia", {"hwnd": "0x0"})
self.assertFalse(decoded["ok"])
self.assertIn(decoded["error"]["code"], {"WINDOW_NOT_FOUND", "UIA_DUMP_FAILED"})
if __name__ == "__main__":
unittest.main()