diff --git a/CHANGELOG.md b/CHANGELOG.md index 46d4886..0cb4536 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,10 @@ This project follows a practical release-log style: user-visible capabilities, s - Renamed Janus to unify agent identity under Stram: Stram is now both the runtime and the agent, not two separately named things. The `stram/janus` package moved to `stram/agent`, `Janus*` classes and identifiers became `Agent*` (e.g. `JanusStore` -> `AgentStore`, `JanusEventRouter` -> `AgentEventRouter`), and REST API fields/routes renamed accordingly (e.g. `janus_memory`/`janus_state` -> `agent_memory`/`agent_state`, `/janus/*` routes -> `/agent/*`). Desktop app nav items and UI labels previously named "Janus" now read "Agent". Any integration relying on the old `janus_*` field names, module paths, or `/janus/*` endpoints must update to the `agent_*` equivalents. +### Fixed + +- Existing local agent state now survives the rename. On first start after upgrading, an existing `janus.sqlite3` database (with its WAL sidecars) is carried over to `agent.sqlite3` and the `janus_activations` table is renamed to `agent_activations`, so task contexts, episodes, memory candidates, and activations recorded before the rename are not orphaned behind an empty new database. + ## 1.0.0 - First production release ### Highlights diff --git a/apps/macos/Sources/AppViewModel.swift b/apps/macos/Sources/AppViewModel.swift index 15edb9a..2a953ee 100644 --- a/apps/macos/Sources/AppViewModel.swift +++ b/apps/macos/Sources/AppViewModel.swift @@ -72,6 +72,9 @@ final class AppViewModel: ObservableObject { private var api: AgentAPIClient private let voiceWakeService = VoiceWakeService() private var channelListenerTask: Task? + private var notifiedApprovalTokens: Set = [] + private var decidingTokens: Set = [] + private var approvalWatchTask: Task? private var channelListenerTickRunning = false private var lastHandledVoiceTranscript = "" private var isCapturingVoiceTask = false @@ -219,6 +222,114 @@ final class AppViewModel: ObservableObject { } } + /// Nothing else polls approvals: `refreshApprovals()` only runs on bootstrap, window + /// activation, or a manual refresh, so without this watch a new permission request + /// can sit unseen and the notification would have nothing to fire on. + func startApprovalWatch() { + guard approvalWatchTask == nil else { return } + ApprovalNotifier.shared.configure( + approve: { [weak self] token in + Task { await self?.approve(token: token) } + }, + reject: { [weak self] token in + Task { await self?.reject(token: token) } + }, + authorizationDenied: { [weak self] in + self?.notice = "Notifications are off, so Stram cannot alert you when a run needs permission. Turn them on in System Settings > Notifications, or watch the Permissions page instead." + } + ) + ApprovalNotifier.shared.requestAuthorization() + + approvalWatchTask = Task { [weak self] in + while !Task.isCancelled { + guard let self else { return } + await self.pollApprovalsForNotification() + // ponytail: fast only while something can produce or clear an approval, + // otherwise back off. A flat 2s poll would hammer the API and the battery + // all day. Stopping outright is wrong — an autonomous tick can park on an + // approval with no UI event to restart the watch. + let interval = self.approvalPollInterval + try? await Task.sleep(for: interval) + } + } + } + + func stopApprovalWatch() { + approvalWatchTask?.cancel() + approvalWatchTask = nil + } + + private var approvalPollInterval: Duration { + (isSending || !approvals.isEmpty) ? .seconds(2) : .seconds(20) + } + + private func pollApprovalsForNotification() async { + let pending: [ApprovalItem] + do { + // Deliberately not `refreshApprovals()`: that swallows errors into an empty + // list, and treating a failed poll as "nothing pending" withdraws live banners + // while the run is still parked (daemon restart, wake from sleep), then + // re-alerts on recovery. + pending = try await api.approvals() + } catch { + return + } + approvals = pending + if selectedApproval == nil { + selectedApproval = pending.first + } + let pendingTokens = Set(pending.map(\.approvalToken)) + // Anything decided elsewhere, or whose run was cancelled, is no longer pending: + // withdraw its banner and forget it so a later token reusing the set is unaffected. + for token in notifiedApprovalTokens.subtracting(pendingTokens) { + ApprovalNotifier.shared.withdraw(token: token) + } + notifiedApprovalTokens.formIntersection(pendingTokens) + for approval in pending where !notifiedApprovalTokens.contains(approval.approvalToken) + && !decidingTokens.contains(approval.approvalToken) { + // Only record it as notified if the notification actually went out: before the + // system grant lands `notify` is a no-op, and a token marked then would never + // be alerted again while it stays pending. + if ApprovalNotifier.shared.notify(approval) { + notifiedApprovalTokens.insert(approval.approvalToken) + } + } + } + + func approve(token: String, label: String? = nil) async { + await decideApproval(token: token, label: label, approve: true) + } + + func reject(token: String, label: String? = nil) async { + await decideApproval(token: token, label: label, approve: false) + } + + private func decideApproval(token: String, label: String?, approve: Bool) async { + let verb = approve ? "Approved" : "Rejected" + let name = label ?? String(token.prefix(8)) + // A poll tick that overlaps this decision sees a list fetched before the server + // committed it; without this the token looks pending-and-unnotified and gets a + // fresh banner for a request the user just decided. + decidingTokens.insert(token) + defer { decidingTokens.remove(token) } + do { + if approve { + _ = try await api.approve(token, note: "Approved from Stram Mac.") + } else { + _ = try await api.reject(token, note: "Rejected from Stram Mac.") + } + notice = "\(verb) \(name)." + } catch { + // Already decided on the Permissions page, or the run was cancelled. Say so + // once and forget the token — retrying would just fail again. + notice = "Could not \(approve ? "approve" : "reject") \(name): \(error.localizedDescription)" + } + ApprovalNotifier.shared.withdraw(token: token) + notifiedApprovalTokens.remove(token) + await refreshRuns() + await refreshApprovals() + } + func refreshChannels() async { do { channels = try await api.channels() @@ -638,10 +749,17 @@ final class AppViewModel: ObservableObject { return copy } + /// Stops every child process this app owns. Called on quit as well as manual stop: + /// an orphaned daemon keeps its session alive, and with it the grants that let + /// read-only capture tools run without prompting. + func stopChildProcesses() { + nativeCollectorProcess.stop() + agentProcess.stop() + } + func toggleAgentProcess() async { if agentProcess.isRunning { - nativeCollectorProcess.stop() - agentProcess.stop() + stopChildProcesses() status = .offline return } @@ -668,26 +786,12 @@ final class AppViewModel: ObservableObject { func approveSelected() async { guard let selectedApproval else { return } - do { - _ = try await api.approve(selectedApproval.approvalToken, note: "Approved from Stram Mac.") - notice = "Approved \(selectedApproval.toolName)." - await refreshRuns() - await refreshApprovals() - } catch { - notice = error.localizedDescription - } + await approve(token: selectedApproval.approvalToken, label: selectedApproval.toolName) } func rejectSelected() async { guard let selectedApproval else { return } - do { - _ = try await api.reject(selectedApproval.approvalToken, note: "Rejected from Stram Mac.") - notice = "Rejected \(selectedApproval.toolName)." - await refreshRuns() - await refreshApprovals() - } catch { - notice = error.localizedDescription - } + await reject(token: selectedApproval.approvalToken, label: selectedApproval.toolName) } func cancelSelectedRun() async { diff --git a/apps/macos/Sources/ApprovalNotifier.swift b/apps/macos/Sources/ApprovalNotifier.swift new file mode 100644 index 0000000..79b3353 --- /dev/null +++ b/apps/macos/Sources/ApprovalNotifier.swift @@ -0,0 +1,183 @@ +import Foundation +import UserNotifications + +// File-private so the `nonisolated` delegate callbacks can read them without +// hopping to the main actor. +private let approvalCategoryIdentifier = "APPROVAL" +private let approveActionIdentifier = "APPROVAL_APPROVE" +private let rejectActionIdentifier = "APPROVAL_REJECT" +private let approvalTokenKey = "approval_token" +/// Persisted so a user who denied notifications is told once, not every launch. +private let deniedNoticeShownKey = "StramMac.notificationsDeniedNoticeShown" + +/// Raises a native notification with Approve / Reject buttons when a run parks on a +/// permission request, so the user does not have to notice a spinner in chat and then +/// navigate to the Permissions page to unblock the run. +@MainActor +final class ApprovalNotifier: NSObject, UNUserNotificationCenterDelegate { + static let shared = ApprovalNotifier() + + private var approveHandler: ((String) -> Void)? + private var rejectHandler: ((String) -> Void)? + private var authorizationDeniedHandler: (() -> Void)? + private var authorized = false + /// A response that launched the app arrives before the view model can wire up the + /// handlers, so hold it here and run it as soon as `configure` supplies them. + private var pendingResponse: (token: String, approve: Bool)? + + func configure( + approve: @escaping (String) -> Void, + reject: @escaping (String) -> Void, + authorizationDenied: @escaping () -> Void + ) { + approveHandler = approve + rejectHandler = reject + authorizationDeniedHandler = authorizationDenied + if let pendingResponse { + self.pendingResponse = nil + (pendingResponse.approve ? approve : reject)(pendingResponse.token) + } + } + + /// `UNUserNotificationCenter.current()` traps when the process has no bundle + /// identifier, which is the case under `swift run`. Stay inert there instead of + /// crashing the dev flow; the built bundle has an identifier. + private var isAvailable: Bool { + Bundle.main.bundleIdentifier != nil + } + + /// Must run before app launch finishes: macOS delivers a response that launched the + /// process immediately, and drops it if the delegate is still nil. Registering the + /// category here too keeps the Approve / Reject buttons on notifications that were + /// delivered by a previous launch. + func registerDelegate() { + guard isAvailable else { return } + let center = UNUserNotificationCenter.current() + center.delegate = self + + let approve = UNNotificationAction( + identifier: approveActionIdentifier, + title: "Approve", + options: [.authenticationRequired] + ) + let reject = UNNotificationAction( + identifier: rejectActionIdentifier, + title: "Reject", + options: [.destructive] + ) + center.setNotificationCategories([ + UNNotificationCategory( + identifier: approvalCategoryIdentifier, + actions: [approve, reject], + intentIdentifiers: [], + options: [] + ) + ]) + } + + func requestAuthorization() { + guard isAvailable else { return } + UNUserNotificationCenter.current().requestAuthorization(options: [.alert, .sound]) { @Sendable granted, _ in + Task { @MainActor in + ApprovalNotifier.shared.authorized = granted + let defaults = UserDefaults.standard + if granted { + // Told again if permission is revoked later, but not on every launch. + defaults.removeObject(forKey: deniedNoticeShownKey) + } else if !defaults.bool(forKey: deniedNoticeShownKey) { + defaults.set(true, forKey: deniedNoticeShownKey) + // Read from main-actor state rather than captured: the handler is not Sendable. + ApprovalNotifier.shared.authorizationDeniedHandler?() + } + } + } + } + + /// Returns whether authorization and the bundle-id check passed, i.e. the request was + /// handed to the system. Delivery itself is asynchronous and not reported here. + func notify(_ approval: ApprovalItem) -> Bool { + guard isAvailable, authorized else { return false } + + let content = UNMutableNotificationContent() + content.title = "Stram needs permission" + // displayRisk already reads as "High attention" / "Medium attention". + content.subtitle = "\(approval.displayToolName) - \(approval.displayRisk)" + // The banner's Approve button decides immediately, so it must show what is being + // approved: `reason` is a per-risk-class constant and says nothing about the action. + content.body = Self.body(for: approval) + content.categoryIdentifier = approvalCategoryIdentifier + content.userInfo = [approvalTokenKey: approval.approvalToken] + content.sound = .default + + let request = UNNotificationRequest( + identifier: approval.approvalToken, + content: content, + trigger: nil + ) + UNUserNotificationCenter.current().add(request, withCompletionHandler: nil) + return true + } + + /// Request text plus a truncated one-line rendering of the arguments. A notification + /// body is short, so cap it rather than let the system silently clip mid-argument. + private static func body(for approval: ApprovalItem) -> String { + let head = approval.request.isEmpty ? approval.reason : approval.request + guard let input = approval.toolInput, input != .object([:]) else { return head } + var details = input.description + .split(whereSeparator: \.isNewline) + .joined(separator: ", ") + if details.count > 200 { + details = details.prefix(200) + "…" + } + return details.isEmpty ? head : "\(head)\n\(details)" + } + + func withdraw(token: String) { + guard isAvailable else { return } + let center = UNUserNotificationCenter.current() + center.removeDeliveredNotifications(withIdentifiers: [token]) + center.removePendingNotificationRequests(withIdentifiers: [token]) + } + + private func handle(token: String, approve: Bool) { + guard let handler = approve ? approveHandler : rejectHandler else { + // This response launched the app: the delegate is registered at launch but the + // view model configures the handlers a moment later. Queue instead of dropping. + pendingResponse = (token, approve) + return + } + handler(token) + } + + nonisolated func userNotificationCenter( + _ center: UNUserNotificationCenter, + didReceive response: UNNotificationResponse, + withCompletionHandler completionHandler: @escaping () -> Void + ) { + let token = response.notification.request.content.userInfo[approvalTokenKey] as? String + let actionIdentifier = response.actionIdentifier + if let token { + Task { @MainActor in + switch actionIdentifier { + case approveActionIdentifier: + ApprovalNotifier.shared.handle(token: token, approve: true) + case rejectActionIdentifier: + ApprovalNotifier.shared.handle(token: token, approve: false) + default: + break + } + } + } + // Called synchronously: the escaping handler is not Sendable, so it cannot be + // carried into the main-actor hop above. + completionHandler() + } + + nonisolated func userNotificationCenter( + _ center: UNUserNotificationCenter, + willPresent notification: UNNotification, + withCompletionHandler completionHandler: @escaping (UNNotificationPresentationOptions) -> Void + ) { + completionHandler([.banner, .sound]) + } +} diff --git a/apps/macos/Sources/LocalAgentProcess.swift b/apps/macos/Sources/LocalAgentProcess.swift index bc64f41..81b26ce 100644 --- a/apps/macos/Sources/LocalAgentProcess.swift +++ b/apps/macos/Sources/LocalAgentProcess.swift @@ -93,9 +93,24 @@ final class LocalAgentProcess: ObservableObject { appendLog("Started local Stram daemon on port \(settings.port).") } + /// Stops the daemon and waits for it to actually exit. + /// + /// The daemon owns the session grants for privacy-sensitive tools and clears them on + /// SIGTERM, so it must be gone — not merely signalled — before we return. Callers + /// include app termination, where nothing runs after us. func stop() { guard let process else { return } process.terminate() + // ponytail: 3s poll instead of waitUntilExit() so a daemon that ignores SIGTERM + // cannot hang quit; SIGKILL is the backstop. + let deadline = Date.now.addingTimeInterval(3) + while process.isRunning, Date.now < deadline { + usleep(50_000) + } + if process.isRunning { + kill(process.processIdentifier, SIGKILL) + process.waitUntilExit() + } self.process = nil isRunning = false appendLog("Stop requested.") diff --git a/apps/macos/Sources/StramMacApp.swift b/apps/macos/Sources/StramMacApp.swift index e1a35fc..5135fbd 100644 --- a/apps/macos/Sources/StramMacApp.swift +++ b/apps/macos/Sources/StramMacApp.swift @@ -4,6 +4,7 @@ import SwiftUI @main struct StramMacApp: App { @Environment(\.openWindow) private var openWindow + @NSApplicationDelegateAdaptor(StramAppDelegate.self) private var appDelegate @StateObject private var model = AppViewModel() init() { @@ -24,6 +25,7 @@ struct StramMacApp: App { } .task { await model.bootstrap() + model.startApprovalWatch() } } .windowStyle(.hiddenTitleBar) @@ -49,6 +51,7 @@ struct StramMacApp: App { } private func configureStatusBarActions() { + appDelegate.model = model let controller = StramStatusBarController.shared controller.openAction = { openMainWindow() @@ -89,6 +92,28 @@ struct StramMacApp: App { } } +/// Quitting the app must stop the daemon it spawned. Without this, `NSApp.terminate` +/// leaves `python -m stram serve` listening, its pid alive, and its session grants +/// honoured — so autonomous ticks and collectors can still capture the screen with no +/// prompt while Stram looks quit. +@MainActor +final class StramAppDelegate: NSObject, NSApplicationDelegate { + weak var model: AppViewModel? + + /// The notification delegate must exist before launch finishes. A user clicking Approve + /// on a banner left in Notification Center launches the app and macOS delivers that + /// response straight away — with a nil delegate it is silently lost, so registering it + /// from `bootstrap()` (a full refresh plus a daemon start) is seconds too late. + func applicationDidFinishLaunching(_ notification: Notification) { + ApprovalNotifier.shared.registerDelegate() + } + + func applicationWillTerminate(_ notification: Notification) { + model?.stopApprovalWatch() + model?.stopChildProcesses() + } +} + @MainActor private final class StramStatusBarController: NSObject { static let shared = StramStatusBarController() diff --git a/apps/windows/Stram.App/MainWindow.xaml.cs b/apps/windows/Stram.App/MainWindow.xaml.cs index 74a5920..76a927d 100644 --- a/apps/windows/Stram.App/MainWindow.xaml.cs +++ b/apps/windows/Stram.App/MainWindow.xaml.cs @@ -59,6 +59,7 @@ public MainWindow() SetTitleBar(AppTitleBar); RootGrid.Loaded += RootGrid_Loaded; + Closed += MainWindow_Closed; ChatLog.ItemsSource = _chat; ChatConversationList.ItemsSource = _chatConversations; ProcessLog.ItemsSource = _processLines; @@ -768,6 +769,14 @@ private async void StartAgentButton_Click(object sender, RoutedEventArgs e) } } + // Closing the window must stop the daemon it spawned. An orphaned agent process keeps + // its session alive, and with it the grants that let read-only capture tools run + // without prompting while Stram looks closed. + private void MainWindow_Closed(object sender, WindowEventArgs args) + { + _agentProcess.Stop(); + } + private void StopAgentButton_Click(object sender, RoutedEventArgs e) { _agentProcess.Stop(); diff --git a/docs/superpowers/plans/2026-08-06-connector-aware-approvals.md b/docs/superpowers/plans/2026-08-06-connector-aware-approvals.md new file mode 100644 index 0000000..65a268f --- /dev/null +++ b/docs/superpowers/plans/2026-08-06-connector-aware-approvals.md @@ -0,0 +1,1446 @@ +# Connector-Aware Approvals + Native Approval Notifications 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 (`- [ ]`) syntax for tracking. + +**Goal:** Stop asking for approval on read-only actions whose connector is already authorized, grant privacy-gated reads once per runtime session, and surface remaining approvals as native OS notifications with Approve/Reject buttons. + +**Architecture:** `Tool` gains `read_only` and `provider_id` metadata. `PolicyEngine` takes `config` by constructor and consults connector state (cached per instance) plus a session-scoped grant table before falling through to today's unchanged rules. Four thin GitHub read tools give the new fast path something to act on, replacing the shell as the only route to GitHub. Both desktop apps poll the existing `/approvals` endpoint and raise a native notification per unseen token. + +**Tech Stack:** Python 3.12 (stdlib only — `sqlite3`, `uuid`, `unittest`), Swift 6 / SwiftUI / UserNotifications (macOS 14+), C# / WinUI 3 / WindowsAppSDK 1.6 `AppNotificationManager`. + +## Global Constraints + +- Python: **stdlib only**. No new dependencies. Existing code is `from __future__ import annotations` throughout — match it. +- **Do not change any existing tool's `risk_level` or `requires_approval` value.** Adding `read_only=True` is permitted; changing risk is not. This is what keeps `tests/test_planning.py` (~200 embedded `requires_approval` values) green. +- `RiskLevel.BLOCKED` must be evaluated before any new fast path. No combination of flags may run a blocked tool. +- All connector lookups **fail closed**: any exception means "not connected", which means the approval prompt appears. +- `Tool` is `@dataclass(slots=True)` with positional fields. New fields go **after** `capability_group` so existing positional construction (e.g. `DummyTool("dummy", "test", RiskLevel.LOW)`) keeps working. +- `PolicyEngine.evaluate(tool, approved)` signature must **not** change. Config arrives via the constructor. +- macOS min version 14.0, bundle id `ai.stram.mac` (`script/build_and_run.sh:6-7`). +- Windows app is unpackaged (`WindowsPackageType=None`). Do not introduce MSIX. +- Run Python tests with `python -m pytest`. Run a single test with `python -m pytest tests/test_x.py::Class::test_name -v`. + +--- + +## File Structure + +**Create:** +- `stram/safety/grants.py` — session identity + `tool_grants` table. One responsibility: "has this privacy read been granted this session?" +- `tests/test_tool_grants.py` — grant store unit tests. + +**Modify:** +- `stram/tools/base.py:12-20` — add `read_only`, `provider_id` to `Tool`. +- `stram/tools/__init__.py:44-53` — propagate new fields through `_ToolAlias`. +- `stram/safety/policy.py` — constructor config, connector cache, two new rules. +- `stram/orchestrator.py:52`, `stram/runtime.py:31`, `stram/tools/workflow/implementation.py:655`, `stram/safety/permissions.py:26` — pass `config` to `PolicyEngine`. +- `stram/runtime.py` — record a grant after approving a privacy read. +- `stram/api.py:1612` — write the session id on server start. +- `stram/tools/github/implementation.py` — four read-only API tools. +- `stram/connectors/providers/manifests.py:185` — point `tool_hints` at them. +- `tests/test_policy.py`, `tests/test_tools.py`, `tests/test_approval_queue.py` — new coverage. +- `apps/macos/Sources/ApprovalNotifier.swift` (create), `AppViewModel.swift` — poll + notify. +- `apps/windows/Stram.App/Services/ApprovalNotifier.cs` (create), `MainWindow.xaml.cs` — poll + notify. + +--- + +## Task 1: Tool metadata and alias propagation + +**Files:** +- Modify: `stram/tools/base.py:12-20` +- Modify: `stram/tools/__init__.py:44-53` +- Test: `tests/test_tools.py` + +**Interfaces:** +- Consumes: nothing. +- Produces: `Tool.read_only: bool` (default `False`), `Tool.provider_id: str | None` (default `None`). Every later task reads these two attribute names exactly. + +- [ ] **Step 1: Write the failing test** + +Append to `tests/test_tools.py`: + +```python +def test_tool_defaults_are_not_read_only(self) -> None: + from stram.tools.base import Tool + from stram.schemas import RiskLevel + + class Probe(Tool): + def execute(self, tool_input, config): + raise NotImplementedError + + tool = Probe("probe", "test", RiskLevel.LOW) + self.assertFalse(tool.read_only) + self.assertIsNone(tool.provider_id) + +def test_alias_preserves_read_only_metadata(self) -> None: + from stram.tools import _ToolAlias + from stram.tools.base import Tool + from stram.schemas import RiskLevel + + class Probe(Tool): + def execute(self, tool_input, config): + raise NotImplementedError + + target = Probe("probe", "test", RiskLevel.LOW, read_only=True, provider_id="github") + alias = _ToolAlias("probe_alias", target) + self.assertTrue(alias.read_only) + self.assertEqual(alias.provider_id, "github") +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m pytest tests/test_tools.py -k "read_only" -v` +Expected: FAIL — `TypeError: __init__() got an unexpected keyword argument 'read_only'` + +- [ ] **Step 3: Add the fields** + +In `stram/tools/base.py`, the `Tool` dataclass becomes: + +```python +@dataclass(slots=True) +class Tool(ABC): + name: str + description: str + risk_level: RiskLevel + requires_approval: bool = False + input_schema: dict[str, Any] = field(default_factory=lambda: {"type": "object", "properties": {}}) + capability_group: str = "core" + read_only: bool = False + provider_id: str | None = None +``` + +- [ ] **Step 4: Propagate through the alias** + +In `stram/tools/__init__.py`, `_ToolAlias.__init__` gains two lines in its `super().__init__(...)` call: + +```python +class _ToolAlias(Tool): + def __init__(self, alias: str, target: Tool) -> None: + super().__init__( + name=alias, + description=f"Alias for {target.name}: {target.description}", + risk_level=target.risk_level, + requires_approval=target.requires_approval, + input_schema=target.input_schema, + capability_group=target.capability_group, + read_only=target.read_only, + provider_id=target.provider_id, + ) + self._target = target +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `python -m pytest tests/test_tools.py -v` +Expected: PASS, including the full existing file (no regressions — the new fields are defaulted). + +- [ ] **Step 6: Commit** + +```bash +git add stram/tools/base.py stram/tools/__init__.py tests/test_tools.py +git commit -m "feat: add read_only and provider_id metadata to Tool" +``` + +--- + +## Task 2: Session-scoped grant store + +**Files:** +- Create: `stram/safety/grants.py` +- Create: `tests/test_tool_grants.py` + +**Interfaces:** +- Consumes: `AgentConfig.approvals_db_path` (`stram/config.py:83`), `AgentConfig.data_dir`. +- Produces: + - `current_session_id(config: AgentConfig) -> str` — reads `data_dir/session_id`; returns `""` if absent. + - `start_session(config: AgentConfig) -> str` — writes a fresh uuid4, purges stale rows, returns the new id. + - `ToolGrantStore(db_path: Path)` with `.record(tool_name: str, session_id: str) -> None`, `.has(tool_name: str, session_id: str) -> bool`, `.purge_other_sessions(session_id: str) -> None`. + +- [ ] **Step 1: Write the failing test** + +Create `tests/test_tool_grants.py`: + +```python +import tempfile +import unittest +from pathlib import Path + +from stram.config import AgentConfig +from stram.safety.grants import ToolGrantStore, current_session_id, start_session + + +class ToolGrantTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.config = AgentConfig(workspace=Path(self._tmp.name), data_dir=Path("artifacts")).normalized() + + def tearDown(self) -> None: + self._tmp.cleanup() + + def test_no_session_file_means_no_session(self) -> None: + self.assertEqual(current_session_id(self.config), "") + + def test_start_session_writes_readable_id(self) -> None: + session = start_session(self.config) + self.assertTrue(session) + self.assertEqual(current_session_id(self.config), session) + + def test_grant_is_visible_within_session_only(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_clipboard_read", "session-a") + self.assertTrue(store.has("os_clipboard_read", "session-a")) + self.assertFalse(store.has("os_clipboard_read", "session-b")) + self.assertFalse(store.has("screenshot_capture", "session-a")) + + def test_empty_session_never_matches(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_clipboard_read", "") + self.assertFalse(store.has("os_clipboard_read", "")) + + def test_restart_purges_previous_session_grants(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_clipboard_read", "session-a") + store.purge_other_sessions("session-b") + self.assertFalse(store.has("os_clipboard_read", "session-a")) + + +if __name__ == "__main__": + unittest.main() +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `python -m pytest tests/test_tool_grants.py -v` +Expected: FAIL — `ModuleNotFoundError: No module named 'stram.safety.grants'` + +- [ ] **Step 3: Write the implementation** + +Create `stram/safety/grants.py`: + +```python +from __future__ import annotations + +import sqlite3 +from contextlib import closing +from datetime import datetime, timezone +from pathlib import Path +from uuid import uuid4 + +from stram.config import AgentConfig + +SESSION_FILE_NAME = "session_id" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _session_path(config: AgentConfig) -> Path: + return config.data_dir / SESSION_FILE_NAME + + +def current_session_id(config: AgentConfig) -> str: + """Return the running runtime's session id, or "" when no server wrote one.""" + path = _session_path(config) + try: + return path.read_text(encoding="utf-8").strip() + except OSError: + return "" + + +def start_session(config: AgentConfig) -> str: + """Mint a fresh session id and drop every grant from previous sessions.""" + session_id = str(uuid4()) + path = _session_path(config) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(session_id, encoding="utf-8") + ToolGrantStore(config.approvals_db_path).purge_other_sessions(session_id) + return session_id + + +class ToolGrantStore: + """Session-scoped 'ask once' grants for read-only tools with no connector.""" + + def __init__(self, db_path: Path) -> None: + self.db_path = db_path + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self._init_db() + + def _connect(self) -> sqlite3.Connection: + return sqlite3.connect(self.db_path) + + def _init_db(self) -> None: + with closing(self._connect()) as connection: + connection.execute( + """ + CREATE TABLE IF NOT EXISTS tool_grants ( + tool_name TEXT NOT NULL, + session_id TEXT NOT NULL, + granted_at TEXT NOT NULL, + PRIMARY KEY (tool_name, session_id) + ) + """ + ) + connection.commit() + + def record(self, tool_name: str, session_id: str) -> None: + if not tool_name or not session_id: + return + with closing(self._connect()) as connection: + connection.execute( + "INSERT OR REPLACE INTO tool_grants (tool_name, session_id, granted_at) VALUES (?, ?, ?)", + (tool_name, session_id, _now()), + ) + connection.commit() + + def has(self, tool_name: str, session_id: str) -> bool: + if not tool_name or not session_id: + return False + with closing(self._connect()) as connection: + row = connection.execute( + "SELECT 1 FROM tool_grants WHERE tool_name = ? AND session_id = ?", + (tool_name, session_id), + ).fetchone() + return row is not None + + def purge_other_sessions(self, session_id: str) -> None: + with closing(self._connect()) as connection: + connection.execute("DELETE FROM tool_grants WHERE session_id != ?", (session_id,)) + connection.commit() +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `python -m pytest tests/test_tool_grants.py -v` +Expected: PASS, 5 tests. + +- [ ] **Step 5: Commit** + +```bash +git add stram/safety/grants.py tests/test_tool_grants.py +git commit -m "feat: add session-scoped tool grant store" +``` + +--- + +## Task 3: Connector-aware PolicyEngine + +**Files:** +- Modify: `stram/safety/policy.py` +- Test: `tests/test_policy.py` + +**Interfaces:** +- Consumes: `Tool.read_only`, `Tool.provider_id` (Task 1); `ToolGrantStore`, `current_session_id` (Task 2). +- Produces: `PolicyEngine(config: AgentConfig | None = None, *, connected_lookup: Callable[[str], bool] | None = None)`. `evaluate(tool, approved=False) -> PolicyDecision` — signature unchanged. + +**`connected_lookup` is an injectable seam.** Production passes nothing and gets the real `ConnectorRuntime` path. Tests pass a fake so they never touch private attributes. Do not have tests reach into `_connected_cache`. + +**Why the connected-state cache exists:** `permissions_snapshot` (`stram/safety/permissions.py:31-32`) calls `evaluate` twice for every tool in `default_tools()`. Building a `ConnectorRuntime` inside `evaluate` would open the connector SQLite database hundreds of times per snapshot. Cache per `PolicyEngine` instance; since an engine is built per run, decisions also stay self-consistent within a run. + +- [ ] **Step 1: Write the failing tests** + +Replace the body of `tests/test_policy.py` with: + +```python +import tempfile +import unittest +from pathlib import Path + +from stram.config import AgentConfig +from stram.safety.grants import ToolGrantStore, start_session +from stram.safety.policy import PolicyEngine +from stram.schemas import RiskLevel +from stram.tools.base import Tool + + +class DummyTool(Tool): + def execute(self, tool_input, config): + raise NotImplementedError + + +class PolicyTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.config = AgentConfig(workspace=Path(self._tmp.name), data_dir=Path("artifacts")).normalized() + + def tearDown(self) -> None: + self._tmp.cleanup() + + def test_low_risk_tool_is_allowed(self) -> None: + tool = DummyTool("dummy", "test", RiskLevel.LOW) + decision = PolicyEngine().evaluate(tool) + self.assertTrue(decision.allowed) + self.assertFalse(decision.requires_approval) + + def test_high_risk_tool_requires_approval(self) -> None: + tool = DummyTool("dummy", "test", RiskLevel.HIGH) + decision = PolicyEngine().evaluate(tool) + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_blocked_tool_is_never_allowed(self) -> None: + tool = DummyTool("dummy", "test", RiskLevel.BLOCKED) + decision = PolicyEngine().evaluate(tool, approved=True) + self.assertFalse(decision.allowed) + self.assertFalse(decision.requires_approval) + + def test_blocked_read_only_tool_is_still_blocked(self) -> None: + tool = DummyTool("dummy", "test", RiskLevel.BLOCKED, read_only=True, provider_id="github") + engine = PolicyEngine(self.config, connected_lookup=lambda provider_id: True) + decision = engine.evaluate(tool, approved=True) + self.assertFalse(decision.allowed) + + def test_read_only_tool_on_connected_provider_skips_approval(self) -> None: + tool = DummyTool("gh_read", "test", RiskLevel.HIGH, read_only=True, provider_id="github") + engine = PolicyEngine(self.config, connected_lookup=lambda provider_id: True) + decision = engine.evaluate(tool) + self.assertTrue(decision.allowed) + self.assertFalse(decision.requires_approval) + + def test_read_only_tool_on_disconnected_provider_requires_approval(self) -> None: + tool = DummyTool("gh_read", "test", RiskLevel.HIGH, read_only=True, provider_id="github") + engine = PolicyEngine(self.config, connected_lookup=lambda provider_id: False) + decision = engine.evaluate(tool) + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_connector_lookup_failure_fails_closed(self) -> None: + def explode(provider_id: str) -> bool: + raise RuntimeError("connector store unavailable") + + tool = DummyTool("gh_read", "test", RiskLevel.HIGH, read_only=True, provider_id="github") + engine = PolicyEngine(self.config, connected_lookup=explode) + decision = engine.evaluate(tool) + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_unknown_provider_fails_closed_against_real_lookup(self) -> None: + tool = DummyTool("gh_read", "test", RiskLevel.HIGH, read_only=True, provider_id="nope_not_a_provider") + decision = PolicyEngine(self.config).evaluate(tool) + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_privacy_read_requires_approval_then_honours_grant(self) -> None: + tool = DummyTool("os_clipboard_read", "test", RiskLevel.HIGH, requires_approval=True, read_only=True) + session = start_session(self.config) + + first = PolicyEngine(self.config).evaluate(tool) + self.assertFalse(first.allowed) + self.assertTrue(first.requires_approval) + + ToolGrantStore(self.config.approvals_db_path).record("os_clipboard_read", session) + + second = PolicyEngine(self.config).evaluate(tool) + self.assertTrue(second.allowed) + self.assertFalse(second.requires_approval) + + def test_grant_from_another_session_is_ignored(self) -> None: + tool = DummyTool("os_clipboard_read", "test", RiskLevel.HIGH, requires_approval=True, read_only=True) + start_session(self.config) + ToolGrantStore(self.config.approvals_db_path).record("os_clipboard_read", "stale-session") + decision = PolicyEngine(self.config).evaluate(tool) + self.assertFalse(decision.allowed) + + def test_read_only_without_config_requires_approval(self) -> None: + tool = DummyTool("os_clipboard_read", "test", RiskLevel.HIGH, read_only=True) + decision = PolicyEngine().evaluate(tool) + self.assertFalse(decision.allowed) + + +if __name__ == "__main__": + unittest.main() +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/test_policy.py -v` +Expected: FAIL — `TypeError: PolicyEngine() takes no arguments` on the new tests. + +- [ ] **Step 3: Write the implementation** + +Replace `stram/safety/policy.py` with: + +```python +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass + +from stram.config import AgentConfig +from stram.safety.grants import ToolGrantStore, current_session_id +from stram.schemas import RiskLevel +from stram.tools.base import Tool + + +@dataclass(frozen=True, slots=True) +class PolicyDecision: + allowed: bool + requires_approval: bool + reason: str + + +class PolicyEngine: + """Central action gate. All tool calls pass through here before execution. + + Assumes the config given here is the same one Executor.execute is called + with; at every construction site today it is. + """ + + def __init__( + self, + config: AgentConfig | None = None, + *, + connected_lookup: Callable[[str], bool] | None = None, + ) -> None: + self.config = config + self._connected_lookup = connected_lookup + self._connected_cache: dict[str, bool] = {} + self._session_id: str | None = None + self._grants: ToolGrantStore | None = None + + def evaluate(self, tool: Tool, approved: bool = False) -> PolicyDecision: + if tool.risk_level == RiskLevel.BLOCKED: + return PolicyDecision(False, False, "Tool is blocked by policy.") + if tool.read_only: + if tool.provider_id: + if self._provider_connected(tool.provider_id): + return PolicyDecision(True, False, f"Read-only action on connected {tool.provider_id}.") + elif self._granted_this_session(tool.name): + return PolicyDecision(True, False, "Read-only action already allowed this session.") + if tool.risk_level == RiskLevel.HIGH: + if approved: + return PolicyDecision(True, True, "High-risk action approved.") + return PolicyDecision(False, True, "High-risk action requires explicit approval.") + if tool.requires_approval and not approved: + return PolicyDecision(False, True, "Tool requires explicit approval.") + return PolicyDecision(True, tool.requires_approval, "Allowed by local policy.") + + def _provider_connected(self, provider_id: str) -> bool: + if self.config is None: + return False + if provider_id in self._connected_cache: + return self._connected_cache[provider_id] + connected = False + try: + if self._connected_lookup is not None: + connected = bool(self._connected_lookup(provider_id)) + else: + from stram.connectors import ConnectorRuntime + + connected = bool(ConnectorRuntime(self.config).readiness(provider_id).get("connected")) + except Exception: + connected = False + self._connected_cache[provider_id] = connected + return connected + + def _granted_this_session(self, tool_name: str) -> bool: + if self.config is None: + return False + try: + if self._session_id is None: + self._session_id = current_session_id(self.config) + if not self._session_id: + return False + if self._grants is None: + self._grants = ToolGrantStore(self.config.approvals_db_path) + return self._grants.has(tool_name, self._session_id) + except Exception: + return False +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `python -m pytest tests/test_policy.py -v` +Expected: PASS, 11 tests. + +- [ ] **Step 5: Confirm no regression in the security suite** + +Run: `python -m pytest tests/test_approvals_security.py tests/test_executor.py tests/test_approval_queue.py -v` +Expected: PASS. These still construct `PolicyEngine()` with no config indirectly, and no tool sets `read_only` yet, so behaviour is byte-identical. + +- [ ] **Step 6: Commit** + +```bash +git add stram/safety/policy.py tests/test_policy.py +git commit -m "feat: make PolicyEngine connector-aware with session grants" +``` + +--- + +## Task 4: Wire config through, mint the session, record grants + +**Files:** +- Modify: `stram/orchestrator.py:52` +- Modify: `stram/runtime.py:31` and the approval path +- Modify: `stram/tools/workflow/implementation.py:655` +- Modify: `stram/safety/permissions.py:26` +- Modify: `stram/api.py:1612` +- Test: `tests/test_approval_queue.py` + +**Interfaces:** +- Consumes: `PolicyEngine(config)` (Task 3), `start_session` / `ToolGrantStore` (Task 2). +- Produces: a `tool_grants` row after approving a `read_only` tool with no `provider_id`. + +- [ ] **Step 1: Write the failing test** + +Append to `tests/test_approval_queue.py` (match the existing config/tmpdir fixture style already used in that file): + +```python +def test_approving_privacy_read_records_session_grant(self) -> None: + from stram.safety.grants import ToolGrantStore, current_session_id, start_session + from stram.safety.approvals import ApprovalStore + from stram.schemas import ApprovalRequest, RiskLevel + + session = start_session(self.config) + store = ApprovalStore(self.config.approvals_db_path) + store.create_pending( + "run-grant", + "read the clipboard", + ApprovalRequest( + tool_name="os_clipboard_read", + tool_input={}, + risk_level=RiskLevel.HIGH, + reason="privacy read", + approval_token="token-grant", + ), + ) + + from stram.runtime import approve_pending_action + + approve_pending_action(self.config, "token-grant", "approved in test") + + self.assertEqual(current_session_id(self.config), session) + self.assertTrue(ToolGrantStore(self.config.approvals_db_path).has("os_clipboard_read", session)) + +def test_approving_provider_backed_tool_records_no_grant(self) -> None: + from stram.safety.grants import ToolGrantStore, start_session + from stram.safety.approvals import ApprovalStore + from stram.schemas import ApprovalRequest, RiskLevel + + session = start_session(self.config) + store = ApprovalStore(self.config.approvals_db_path) + store.create_pending( + "run-noshell", + "run a shell command", + ApprovalRequest( + tool_name="run_shell_command", + tool_input={"argv": ["python", "--version"]}, + risk_level=RiskLevel.HIGH, + reason="shell", + approval_token="token-noshell", + ), + ) + + from stram.runtime import approve_pending_action + + approve_pending_action(self.config, "token-noshell", "approved in test") + + self.assertFalse(ToolGrantStore(self.config.approvals_db_path).has("run_shell_command", session)) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/test_approval_queue.py -k "grant" -v` +Expected: FAIL — no `tool_grants` row is written, so the first test's final assert is False. + +- [ ] **Step 3: Pass config to every PolicyEngine construction site** + +Four one-line edits: + +`stram/orchestrator.py:52` +```python + self.executor = Executor(self.tools, PolicyEngine(self.config)) +``` + +`stram/runtime.py:31` +```python + executor = Executor(default_tools(config), PolicyEngine(config)) +``` + +`stram/tools/workflow/implementation.py:655` +```python + executor = Executor(tools, PolicyEngine(config)) +``` + +`stram/safety/permissions.py:26` +```python + policy = PolicyEngine(normalized) +``` + +- [ ] **Step 4: Record the grant on approval** + +In `stram/runtime.py`, inside `approve_pending_action`, after the tool has been executed successfully and `approvals.mark_executed(...)` has been called, add: + +```python + approved_tool = default_tools(config).get(record.tool_name) + if approved_tool is not None and approved_tool.read_only and not approved_tool.provider_id: + session_id = current_session_id(config) + if session_id: + ToolGrantStore(config.approvals_db_path).record(record.tool_name, session_id) +``` + +Add the import at the top of `stram/runtime.py`: + +```python +from stram.safety.grants import ToolGrantStore, current_session_id +``` + +- [ ] **Step 5: Mint the session on server start** + +In `stram/api.py`, inside `create_api_server` (line 1612), immediately after the host validation and before `StramAPIServer(...)` is constructed: + +```python + start_session(config.normalized()) +``` + +Add the import at the top of `stram/api.py`: + +```python +from stram.safety.grants import start_session +``` + +- [ ] **Step 6: Mark the five privacy reads as read-only** + +Set `read_only=True` on exactly these five tools. Do **not** touch their `risk_level` or `requires_approval`: + +- `stram/tools/os_control/implementation.py:97` — `os_observe_ui` +- `stram/tools/os_control/implementation.py:992` — `os_clipboard_read` +- `stram/tools/os_control/implementation.py:1150` — `screenshot_capture` +- `stram/tools/external/implementation.py:482` — `screenpipe_search` +- `stram/tools/browser/live_tools.py:1535` — `browser_live_screenshot` + +Each is a `super().__init__(...)` call; add `read_only=True` alongside the existing `requires_approval=True`. + +- [ ] **Step 7: Run the tests** + +Run: `python -m pytest tests/test_approval_queue.py tests/test_policy.py tests/test_executor.py tests/test_approvals_security.py tests/test_api.py -v` +Expected: PASS. `test_executor.py:247-260,366-376` still see `NEEDS_APPROVAL` because each test uses a fresh tmp `data_dir` with no session file and no grants. + +- [ ] **Step 8: Run the full suite** + +Run: `python -m pytest` +Expected: PASS. If `tests/test_planning.py` fails, a `risk_level` or `requires_approval` was changed in Step 6 — revert that and re-run. + +- [ ] **Step 9: Commit** + +```bash +git add stram/orchestrator.py stram/runtime.py stram/api.py stram/safety/permissions.py \ + stram/tools/workflow/implementation.py stram/tools/os_control/implementation.py \ + stram/tools/external/implementation.py stram/tools/browser/live_tools.py \ + tests/test_approval_queue.py +git commit -m "feat: wire connector-aware policy and session grants into the runtime" +``` + +--- + +## Task 5: Read-only GitHub API tools + +**Files:** +- Modify: `stram/tools/github/implementation.py` +- Modify: `stram/connectors/providers/manifests.py:185` +- Test: `tests/test_tools.py` + +**Interfaces:** +- Consumes: `Tool.read_only` / `Tool.provider_id` (Task 1); `ConnectorRuntime.execute_operation` (`stram/connectors/runtime.py:202`) and `ConnectorOperationRequest` (`stram/connectors/models.py:117`). +- Produces: tool names `github_repos_list`, `github_issues_list`, `github_pulls_list`, `github_checks_list`. + +**Note:** `ConnectorPolicy.check_scopes` (`stram/connectors/policy.py:7`) raises `ValueError` when the provider is not connected and `PermissionError` on missing scopes. `execute_operation` surfaces those, so an unconnected GitHub produces a clear failure rather than a silent skip. + +- [ ] **Step 1: Write the failing test** + +Append to `tests/test_tools.py`: + +```python +def test_github_read_tools_are_read_only_and_provider_scoped(self) -> None: + from stram.tools.github import default_github_tools + + tools = default_github_tools() + for name in ("github_repos_list", "github_issues_list", "github_pulls_list", "github_checks_list"): + tool = tools[name] + self.assertTrue(tool.read_only, name) + self.assertEqual(tool.provider_id, "github", name) + self.assertEqual(tool.risk_level, RiskLevel.LOW, name) + self.assertFalse(tool.requires_approval, name) + +def test_github_read_tool_fails_clearly_when_not_connected(self) -> None: + import tempfile + from pathlib import Path + from stram.config import AgentConfig + from stram.schemas import ActionStatus + from stram.tools.github import default_github_tools + + with tempfile.TemporaryDirectory() as tmp: + config = AgentConfig(workspace=Path(tmp), data_dir=Path("artifacts")).normalized() + result = default_github_tools()["github_repos_list"].execute({}, config) + self.assertEqual(result.status, ActionStatus.FAILED) + self.assertIn("not connected", (result.error or "").lower()) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/test_tools.py -k "github_read" -v` +Expected: FAIL — `KeyError: 'github_repos_list'` + +- [ ] **Step 3: Write the implementation** + +Add to `stram/tools/github/implementation.py`, after `GitHubWorkflowArtifactInspectTool`: + +```python +class GitHubReadTool(Tool): + """Read-only GitHub API call routed through the workspace connector.""" + + def __init__( + self, + name: str, + description: str, + *, + operation: str, + path_template: str, + required_scopes: tuple[str, ...], + properties: dict[str, dict[str, Any]], + required: list[str], + ) -> None: + super().__init__( + name=name, + description=description, + risk_level=RiskLevel.LOW, + requires_approval=False, + input_schema=object_input_schema( + { + **properties, + "per_page": { + "type": "integer", + "description": "Maximum items to return (1-100).", + }, + }, + required=required, + ), + capability_group="github", + read_only=True, + provider_id="github", + ) + self._operation = operation + self._path_template = path_template + self._required_scopes = required_scopes + + def execute(self, tool_input: dict[str, Any], config: AgentConfig) -> ToolResult: + from stram.connectors import ConnectorOperationRequest, ConnectorRuntime + + try: + path = self._path_template.format(**{key: _github_path_segment(tool_input, key) for key in _template_keys(self._path_template)}) + except ValueError as exc: + return ToolResult(self.name, ActionStatus.FAILED, self.risk_level, str(exc), error=str(exc)) + + per_page = tool_input.get("per_page") + try: + per_page_value = max(1, min(int(per_page), 100)) if per_page is not None else 30 + except (TypeError, ValueError): + per_page_value = 30 + + request = ConnectorOperationRequest( + provider_id="github", + operation=self._operation, + method="GET", + path=path, + query={"per_page": per_page_value}, + required_scopes=self._required_scopes, + reason=f"Read-only GitHub metadata for {self.name}.", + ) + try: + result = ConnectorRuntime(config).execute_operation(request) + except (ValueError, PermissionError) as exc: + return ToolResult(self.name, ActionStatus.FAILED, self.risk_level, str(exc), error=str(exc)) + except Exception as exc: + return ToolResult(self.name, ActionStatus.FAILED, self.risk_level, f"{self.name} failed.", error=str(exc)) + + response = result.get("response") + items = response if isinstance(response, list) else [response] + trimmed = items[:MAX_GITHUB_ITEMS] + return ToolResult( + self.name, + ActionStatus.SUCCEEDED, + self.risk_level, + f"Read {len(trimmed)} item(s) from GitHub via {self._operation}.", + { + "operation": self._operation, + "path": path, + "status_code": result.get("status_code"), + "count": len(trimmed), + "items": trimmed, + }, + ) + + +def _template_keys(template: str) -> tuple[str, ...]: + import re + + return tuple(re.findall(r"\{([a-zA-Z0-9_]+)\}", template)) + + +def _github_path_segment(tool_input: dict[str, Any], key: str) -> str: + value = str(tool_input.get(key) or "").strip().strip("/") + if not value: + raise ValueError(f"{key} is required.") + if "/" in value and key != "repo": + raise ValueError(f"{key} must be a single path segment.") + return value +``` + +- [ ] **Step 4: Register the four tools** + +Extend `default_github_tools()` in the same file: + +```python +def default_github_tools() -> dict[str, Tool]: + tools: list[Tool] = [ + GitHubIssueDraftCreateTool(), + GitHubIssueDraftCreateTool("github_issue_packet_create"), + GitHubPrSummaryCreateTool(), + GitHubPrSummaryCreateTool("github_pr_packet_create"), + CiFailureReportCreateTool(), + GitHubRepoStateReportCreateTool(), + GitHubWorkflowArtifactInspectTool(), + GitHubWorkflowArtifactInspectTool("github_artifact_inspect"), + GitHubReadTool( + "github_repos_list", + "List repositories the connected GitHub account can access.", + operation="github_repos_list", + path_template="/user/repos", + required_scopes=("repo",), + properties={}, + required=[], + ), + GitHubReadTool( + "github_issues_list", + "List open issues for a repository, given repo as 'owner/name'.", + operation="github_issues_list", + path_template="/repos/{repo}/issues", + required_scopes=("repo",), + properties={"repo": {"type": "string", "description": "Repository as owner/name."}}, + required=["repo"], + ), + GitHubReadTool( + "github_pulls_list", + "List pull requests for a repository, given repo as 'owner/name'.", + operation="github_pulls_list", + path_template="/repos/{repo}/pulls", + required_scopes=("repo",), + properties={"repo": {"type": "string", "description": "Repository as owner/name."}}, + required=["repo"], + ), + GitHubReadTool( + "github_checks_list", + "List CI check runs for a commit ref in a repository.", + operation="github_checks_list", + path_template="/repos/{repo}/commits/{ref}/check-runs", + required_scopes=("repo", "workflow"), + properties={ + "repo": {"type": "string", "description": "Repository as owner/name."}, + "ref": {"type": "string", "description": "Commit SHA, branch, or tag."}, + }, + required=["repo", "ref"], + ), + ] + return {tool.name: tool for tool in tools} +``` + +- [ ] **Step 5: Point the manifest at the new tools** + +In `stram/connectors/providers/manifests.py:185`, replace the GitHub `tool_hints` line: + +```python + tool_hints=( + "github_repos_list", + "github_issues_list", + "github_pulls_list", + "github_checks_list", + "github_repo_state_report_create", + "github_pr_packet_create", + "github_issue_packet_create", + "ci_failure_report_create", + ), +``` + +- [ ] **Step 6: Run the tests** + +Run: `python -m pytest tests/test_tools.py -k github -v && python -m pytest tests/test_workspace_connectors.py -v` +Expected: PASS. + +- [ ] **Step 7: Run the full suite** + +Run: `python -m pytest` +Expected: PASS. New tool names add entries to planner catalogs rather than editing existing expectations. + +- [ ] **Step 8: Commit** + +```bash +git add stram/tools/github/implementation.py stram/connectors/providers/manifests.py tests/test_tools.py +git commit -m "feat: add read-only GitHub API tools routed through the connector" +``` + +--- + +## Task 6: macOS approval notifications + +**Files:** +- Create: `apps/macos/Sources/ApprovalNotifier.swift` +- Modify: `apps/macos/Sources/AppViewModel.swift` +- Modify: `apps/macos/Sources/StramMacApp.swift` + +**Interfaces:** +- Consumes: `ApprovalItem` (`apps/macos/Sources/Models.swift:303-334`), `AppViewModel.api.approvals()` / `.approve(_:note:)` / `.reject(_:note:)` (`AgentAPIClient.swift:95,462,466`). +- Produces: `ApprovalNotifier.shared` with `configure(approve:reject:)`, `requestAuthorization()`, `notify(_ approval: ApprovalItem)`. + +**Constraint:** `UNUserNotificationCenter` requires a bundle identifier, so this cannot be verified with `swift run`. Verification is Task 8. + +- [ ] **Step 1: Create the notifier** + +Create `apps/macos/Sources/ApprovalNotifier.swift`: + +```swift +import Foundation +import UserNotifications + +@MainActor +final class ApprovalNotifier: NSObject, UNUserNotificationCenterDelegate { + static let shared = ApprovalNotifier() + + private let categoryIdentifier = "APPROVAL" + private let approveActionIdentifier = "APPROVAL_APPROVE" + private let rejectActionIdentifier = "APPROVAL_REJECT" + private let tokenKey = "approval_token" + + private var approveHandler: ((String) -> Void)? + private var rejectHandler: ((String) -> Void)? + private var authorized = false + + func configure(approve: @escaping (String) -> Void, reject: @escaping (String) -> Void) { + approveHandler = approve + rejectHandler = reject + } + + func requestAuthorization() { + let center = UNUserNotificationCenter.current() + center.delegate = self + + let approve = UNNotificationAction(title: "Approve", identifier: approveActionIdentifier, options: [.authenticationRequired]) + let reject = UNNotificationAction(title: "Reject", identifier: rejectActionIdentifier, options: [.destructive]) + let category = UNNotificationCategory( + identifier: categoryIdentifier, + actions: [approve, reject], + intentIdentifiers: [], + options: [] + ) + center.setNotificationCategories([category]) + + center.requestAuthorization(options: [.alert, .sound]) { [weak self] granted, _ in + Task { @MainActor in + self?.authorized = granted + } + } + } + + func notify(_ approval: ApprovalItem) { + guard authorized else { return } + + let content = UNMutableNotificationContent() + content.title = "Stram needs permission" + content.subtitle = approval.toolName + content.body = approval.reason + content.categoryIdentifier = categoryIdentifier + content.userInfo = [tokenKey: approval.approvalToken] + content.sound = .default + + let request = UNNotificationRequest( + identifier: approval.approvalToken, + content: content, + trigger: nil + ) + UNUserNotificationCenter.current().add(request, withCompletionHandler: nil) + } + + func withdraw(token: String) { + UNUserNotificationCenter.current().removeDeliveredNotifications(withIdentifiers: [token]) + } + + nonisolated func userNotificationCenter( + _ center: UNUserNotificationCenter, + didReceive response: UNNotificationResponse, + withCompletionHandler completionHandler: @escaping () -> Void + ) { + let userInfo = response.notification.request.content.userInfo + let actionIdentifier = response.actionIdentifier + Task { @MainActor in + defer { completionHandler() } + guard let token = userInfo[self.tokenKey] as? String else { return } + switch actionIdentifier { + case self.approveActionIdentifier: + self.approveHandler?(token) + case self.rejectActionIdentifier: + self.rejectHandler?(token) + default: + break + } + } + } + + nonisolated func userNotificationCenter( + _ center: UNUserNotificationCenter, + willPresent notification: UNNotification, + withCompletionHandler completionHandler: @escaping (UNNotificationPresentationOptions) -> Void + ) { + completionHandler([.banner, .sound]) + } +} +``` + +- [ ] **Step 2: Add the approval poll to AppViewModel** + +Add these properties to `AppViewModel` (near `@Published var notice: String?` at line 55): + +```swift + private var notifiedApprovalTokens: Set = [] + private var approvalWatchTask: Task? +``` + +Add these methods to `AppViewModel`: + +```swift + func startApprovalWatch() { + guard approvalWatchTask == nil else { return } + ApprovalNotifier.shared.configure( + approve: { [weak self] token in + Task { await self?.approve(token: token) } + }, + reject: { [weak self] token in + Task { await self?.reject(token: token) } + } + ) + ApprovalNotifier.shared.requestAuthorization() + + approvalWatchTask = Task { [weak self] in + while !Task.isCancelled { + await self?.pollApprovalsForNotification() + try? await Task.sleep(nanoseconds: 2_000_000_000) + } + } + } + + func stopApprovalWatch() { + approvalWatchTask?.cancel() + approvalWatchTask = nil + } + + private func pollApprovalsForNotification() async { + await refreshApprovals() + let pending = approvals + let pendingTokens = Set(pending.map(\.approvalToken)) + notifiedApprovalTokens.formIntersection(pendingTokens) + for approval in pending where !notifiedApprovalTokens.contains(approval.approvalToken) { + notifiedApprovalTokens.insert(approval.approvalToken) + ApprovalNotifier.shared.notify(approval) + } + } + + func approve(token: String) async { + do { + _ = try await api.approve(token, note: "Approved from a Stram notification.") + ApprovalNotifier.shared.withdraw(token: token) + notice = "Approved \(token.prefix(8))." + } catch { + notice = "Could not approve \(token.prefix(8)): \(error.localizedDescription)" + } + notifiedApprovalTokens.remove(token) + await refreshRuns() + await refreshApprovals() + } + + func reject(token: String) async { + do { + _ = try await api.reject(token, note: "Rejected from a Stram notification.") + ApprovalNotifier.shared.withdraw(token: token) + notice = "Rejected \(token.prefix(8))." + } catch { + notice = "Could not reject \(token.prefix(8)): \(error.localizedDescription)" + } + notifiedApprovalTokens.remove(token) + await refreshRuns() + await refreshApprovals() + } +``` + +If `refreshRuns()` / `refreshApprovals()` are not `async` in this file, drop the `await` on those two calls to match their real signatures at `AppViewModel.swift:211-220`. + +- [ ] **Step 3: Start the watch on launch** + +In `apps/macos/Sources/StramMacApp.swift`, extend the existing `.task` modifier on `RootView`: + +```swift + .task { + await model.bootstrap() + model.startApprovalWatch() + } +``` + +- [ ] **Step 4: Build** + +Run: `swift build --package-path apps/macos` +Expected: `Build complete!` with no errors. Fix any signature mismatches against the real `AgentAPIClient` method names before continuing. + +- [ ] **Step 5: Commit** + +```bash +git add apps/macos/Sources/ApprovalNotifier.swift apps/macos/Sources/AppViewModel.swift apps/macos/Sources/StramMacApp.swift +git commit -m "feat: raise native macOS notifications for pending approvals" +``` + +--- + +## Task 7: Windows approval notifications + +**DESCOPED BY THE USER — NOT IMPLEMENTED.** No `ApprovalNotifier.cs` exists and `grep -rn AppNotification apps/windows` is empty. The only Windows change that shipped is the daemon-lifecycle stop in `MainWindow.xaml.cs:62,771`, which is uncompiled. Nothing below this line was built. + +**Files:** +- Create: `apps/windows/Stram.App/Services/ApprovalNotifier.cs` +- Modify: `apps/windows/Stram.App/MainWindow.xaml.cs` + +**Interfaces:** +- Consumes: `ApprovalItem` (`apps/windows/Stram.App/Models/AgentModels.cs:1004-1035`), `AgentApiClient.GetApprovalsAsync` / `ApproveAsync` / `RejectAsync` (`Services/AgentApiClient.cs:351,356,361`). +- Produces: `ApprovalNotifier` with `Register(Action approve, Action reject)`, `Notify(ApprovalItem approval)`, `Withdraw(string token)`. + +**Cannot be verified on macOS.** Build and manual verification must happen on a Windows machine. Commit it as untested and say so in the commit message. + +- [ ] **Step 1: Create the notifier** + +Create `apps/windows/Stram.App/Services/ApprovalNotifier.cs`: + +```csharp +using Microsoft.Windows.AppNotifications; +using Microsoft.Windows.AppNotifications.Builder; + +namespace Stram.App.Services; + +public sealed class ApprovalNotifier +{ + private const string TokenKey = "approvalToken"; + private const string ActionKey = "action"; + + private Action? _approve; + private Action? _reject; + private bool _registered; + + public void Register(Action approve, Action reject) + { + _approve = approve; + _reject = reject; + + if (_registered) + { + return; + } + + var manager = AppNotificationManager.Default; + manager.NotificationInvoked += OnNotificationInvoked; + manager.Register(); + _registered = true; + } + + public void Unregister() + { + if (!_registered) + { + return; + } + + AppNotificationManager.Default.Unregister(); + _registered = false; + } + + public void Notify(ApprovalItem approval) + { + var notification = new AppNotificationBuilder() + .AddText("Stram needs permission") + .AddText(approval.ToolName) + .AddText(approval.Reason) + .AddButton(new AppNotificationButton("Approve") + .AddArgument(ActionKey, "approve") + .AddArgument(TokenKey, approval.ApprovalToken)) + .AddButton(new AppNotificationButton("Reject") + .AddArgument(ActionKey, "reject") + .AddArgument(TokenKey, approval.ApprovalToken)) + .BuildNotification(); + + notification.Tag = approval.ApprovalToken; + AppNotificationManager.Default.Show(notification); + } + + public void Withdraw(string token) + { + _ = AppNotificationManager.Default.RemoveByTagAsync(token); + } + + private void OnNotificationInvoked(AppNotificationManager sender, AppNotificationActivatedEventArgs args) + { + if (!args.Arguments.TryGetValue(TokenKey, out var token) || string.IsNullOrWhiteSpace(token)) + { + return; + } + + if (!args.Arguments.TryGetValue(ActionKey, out var action)) + { + return; + } + + if (action == "approve") + { + _approve?.Invoke(token); + } + else if (action == "reject") + { + _reject?.Invoke(token); + } + } +} +``` + +If `ApprovalItem` lives in a different namespace, add the matching `using` — check `Models/AgentModels.cs:1004`. + +- [ ] **Step 2: Add the poll to MainWindow** + +Add fields to `MainWindow`: + +```csharp + private readonly ApprovalNotifier _approvalNotifier = new(); + private readonly HashSet _notifiedApprovalTokens = new(); + private DispatcherTimer? _approvalTimer; +``` + +Add these methods: + +```csharp + private void StartApprovalWatch() + { + _approvalNotifier.Register( + token => DispatcherQueue.TryEnqueue(async () => await ApproveFromNotificationAsync(token)), + token => DispatcherQueue.TryEnqueue(async () => await RejectFromNotificationAsync(token))); + + _approvalTimer = new DispatcherTimer { Interval = TimeSpan.FromSeconds(2) }; + _approvalTimer.Tick += async (_, _) => await PollApprovalsForNotificationAsync(); + _approvalTimer.Start(); + } + + private async Task PollApprovalsForNotificationAsync() + { + List pending; + try + { + pending = await _api.GetApprovalsAsync(); + } + catch + { + return; + } + + var pendingTokens = pending.Select(item => item.ApprovalToken).ToHashSet(); + _notifiedApprovalTokens.IntersectWith(pendingTokens); + + foreach (var approval in pending) + { + if (_notifiedApprovalTokens.Add(approval.ApprovalToken)) + { + _approvalNotifier.Notify(approval); + } + } + } + + private async Task ApproveFromNotificationAsync(string token) + { + try + { + await _api.ApproveAsync(token, "Approved from a Stram notification."); + _approvalNotifier.Withdraw(token); + ShowNotice($"Approved {token[..8]}.", InfoBarSeverity.Success); + } + catch (Exception ex) + { + ShowNotice($"Could not approve {token[..8]}: {ex.Message}", InfoBarSeverity.Error); + } + + _notifiedApprovalTokens.Remove(token); + await RefreshRuntimeAsync(); + } + + private async Task RejectFromNotificationAsync(string token) + { + try + { + await _api.RejectAsync(token, "Rejected from a Stram notification."); + _approvalNotifier.Withdraw(token); + ShowNotice($"Rejected {token[..8]}.", InfoBarSeverity.Warning); + } + catch (Exception ex) + { + ShowNotice($"Could not reject {token[..8]}: {ex.Message}", InfoBarSeverity.Error); + } + + _notifiedApprovalTokens.Remove(token); + await RefreshRuntimeAsync(); + } +``` + +Call `StartApprovalWatch();` at the end of the `MainWindow` constructor. Match `_api`, `ShowNotice` (`MainWindow.xaml.cs:2498`), and `RefreshRuntimeAsync` (`:636`) to their real names in the file. + +- [ ] **Step 3: Commit as untested** + +```bash +git add apps/windows/Stram.App/Services/ApprovalNotifier.cs apps/windows/Stram.App/MainWindow.xaml.cs +git commit -m "feat: raise native Windows notifications for pending approvals + +Written but not compiled or run — no Windows machine available. +Needs a build and manual verification before release." +``` + +--- + +## Task 8: Build and launch locally on macOS + +**Files:** none modified. + +- [ ] **Step 1: Run the whole Python suite** + +Run: `python -m pytest` +Expected: PASS. Do not proceed past a failure. + +- [ ] **Step 2: Build and launch the bundled app** + +Run: `./script/build_and_run.sh` +Expected: `swift build` succeeds, `dist/StramMac.app` is rebuilt, and the app launches. `swift run` will **not** work for notifications — the bundle is required. + +- [ ] **Step 3: Grant notification permission** + +macOS prompts once for notification permission on first launch. Accept it. If the prompt does not appear, check System Settings → Notifications → Stram. + +- [ ] **Step 4: Verify the read-only fast path** + +In the app, connect GitHub if not already connected, then ask Stram to list your repositories. Expected: it calls `github_repos_list` and returns results **with no approval prompt**. If it still asks, check that the GitHub connector reports `connected` and that the planner picked `github_repos_list` rather than `run_shell_command`. + +- [ ] **Step 5: Verify the notification path** + +Ask Stram to do something high-risk that is not read-only — e.g. write a file outside the workspace. Expected: a native notification titled "Stram needs permission" with Approve and Reject buttons. Click Approve; the run should continue and the Permissions page should show the token as executed. + +- [ ] **Step 6: Verify the session grant** + +Ask Stram to read the clipboard twice. Expected: a notification the first time, none the second. Restart the app and ask again: the notification returns. + +- [ ] **Step 7: Report results** + +Report what passed and what did not, with the actual observed behaviour. Do not claim success for any step not actually run. + +--- + +## Self-Review Notes + +**Spec coverage:** Part 1 → Tasks 1, 3, 4. Part 1b → Tasks 2, 4 (Steps 4-6). Part 2 → Task 5. Part 3 → Tasks 6, 7. Error handling (fail closed, BLOCKED first, already-decided token) → Task 3 Step 3, Task 6 Step 2, Task 7 Step 2. Testing section → Tasks 1-5 test steps. Local launch → Task 8. + +**Deviation from spec, deliberate:** the spec said the connected check happens in `evaluate`; the plan caches it per `PolicyEngine` instance because `permissions_snapshot` evaluates every tool twice and would otherwise open the connector database hundreds of times per call. Behaviour is unchanged; cost is not. + +**Known gap:** Task 7 ships uncompiled. Flagged in its commit message and in Task 8's scope (macOS only). diff --git a/docs/superpowers/specs/2026-08-06-connector-aware-approvals-design.md b/docs/superpowers/specs/2026-08-06-connector-aware-approvals-design.md new file mode 100644 index 0000000..ee5caab --- /dev/null +++ b/docs/superpowers/specs/2026-08-06-connector-aware-approvals-design.md @@ -0,0 +1,394 @@ +# Connector-aware approvals + native approval notifications + +Date: 2026-08-06 +Status: implemented, with deltas — read the next section before trusting any +detail below it. Windows approval notifications were descoped and are NOT +implemented. + +## What actually shipped — deltas from this design + +The design text below is kept as-is so the reasoning behind each decision stays +readable. Where the two disagree, **this section is what shipped**. + +- **Grants are keyed `(tool_name, session_id, tool_input)` and matched by EXACT + argument equality.** Rule 3 below and the `PRIMARY KEY (tool_name, + session_id)` schema are both WRONG: per-tool-name grants were unsound. + Subset matching was tried and also proven unsound — every gated tool resolves + an omitted argument to a default that is *broader* than any explicit value + (`content_type="all"`, `max_chars=4000`, no time bound), so a call that drops + a key asks for MORE, not less. **Do not "restore" name-keyed or subset + matching from the old text: it reopens the hole.** See + `tests/test_tool_grants.py::test_omitting_an_approved_key_is_not_covered`. +- `evaluate` gained an optional `tool_input` keyword, and `stram/executor.py` + WAS edited to pass it. The claim below that no call site needed editing is + wrong. Constructor-injected config shipped as designed. +- A grant additionally requires: the tool actually **SUCCEEDED**; a **live + owning process** (`data_dir/session_pid` plus a liveness probe, since a killed + server cannot clean up its own session file); and an age under + `GRANT_TTL_SECONDS` (300s, `stram/safety/grants.py`). The TTL exists because + every session-granted tool reads *ambient* state — argument equality bounds + the request but not the disclosure, so `os_observe_ui` approved over a notes + app would otherwise authorize the byte-identical call over a banking app. +- The liveness probe is `stram.process.pid_alive`, never `os.kill(pid, 0)`: on + Windows that call TERMINATES the target rather than probing it. On Windows it + reports every non-self pid dead, so a non-owner never inherits a grant. +- Session teardown that the design does not mention: `clear_session` on + `server_close`, a SIGTERM handler in `run_api_server`, and + `applicationWillTerminate` stopping the daemon (commit `cc04e63`). +- `PolicyEngine` has an injectable `connected_lookup` seam for tests. +- The four `GitHubReadTool`s ship as `MEDIUM` + `requires_approval=True`, not + `LOW` / no-approval, so the connected-provider rule is load-bearing: + connected → no prompt, disconnected → a normal approval prompt instead of a + raw `PermissionError`. This is a deliberate exception to the "no + reclassification" non-goal. +- **The claim that CLI runs prompt every time is now WRONG.** With a daemon + running, `stram run` reads the live `session_id` file and inherits that + session's grants. Threat model: the local API is unauthenticated, so any local + process that can reach loopback likewise inherits the live session's grants. +- The macOS approval poll does NOT stop when no run is active — it is a + deliberate 2s/20s backoff, because an autonomous tick can park on an approval + with no UI event to restart the watch. +- **Windows approval notifications are descoped and not implemented.** + `AppNotificationManager` is presented below as shipping; it is not — + `grep -rn AppNotification apps/windows` is empty. The Windows change that DID + ship is the daemon-lifecycle stop + (`apps/windows/Stram.App/MainWindow.xaml.cs:62,771`), and it is **UNCOMPILED** + — it has never been built or run on a Windows machine. +- The macOS approval notification body shows the user's request plus a truncated + rendering of `tool_input`, not `PolicyDecision.reason` (a per-risk-class + constant that says nothing about the action being approved). +- **Fixed a real crash, found only by live-launching a properly signed build:** + `ApprovalNotifier.requestAuthorization()` (`apps/macos/Sources/ApprovalNotifier.swift`) + is a method on the `@MainActor`-isolated `ApprovalNotifier` class. The + completion closure passed to `UNUserNotificationCenter.requestAuthorization` + is written inside that method, so Swift 6 inferred `@MainActor` isolation onto + the closure literal itself — but `UNUserNotificationCenter` always invokes that + closure on an arbitrary background queue, never the main actor. The runtime's + isolation check therefore failed every single launch, crashing with + `EXC_BREAKPOINT`/`SIGTRAP` inside `swift_task_checkIsolatedSwift` a few seconds + after `requestAuthorization()` was called. `swift build` never caught this — + it is a dynamic isolation check, not a compile-time one. Fixed by marking the + closure `{ @Sendable granted, _ in ... }`, which tells the compiler the closure + is not actor-isolated. Confirmed via crash report + (`~/Library/Logs/DiagnosticReports/StramMac-*.ips`) before the fix and a clean + 20+ second live run after it. +- **Native notification delivery is unverifiable from a local dev build on this + macOS version, and this is a real environment limit, not a code defect.** + Reproduced and eliminated every alternative explanation before concluding + this: + - Ad-hoc signing (`codesign --sign -`): `UNUserNotificationCenter. + requestAuthorization` returns `didGrant: 0, hasError: 1` in ~5ms, every + time — no system prompt ever shown. + - A locally-created, keychain-trusted self-signed code-signing certificate + ("Stram Local Dev", trusted via `security add-trusted-cert -p codeSign`): + identical `didGrant: 0, hasError: 1` result. + - `tccutil reset All ai.stram.mac` (clears any stale cached decision): no + change. + - Checked and ruled out: Focus/Do Not Disturb (`~/Library/DoNotDisturb/DB/ + Assertions.json` empty), a stale `com.apple.ncprefs` entry (grepped all 93 + app entries, none present), a stale LaunchServices registration, and a + stale/rebuilt binary (rebuilt and relaunched fresh for every attempt). + - Root cause isolated with `spctl -a -vvv`: **Gatekeeper rejects the app** + (`rejected, origin=Stram Local Dev`) even though `codesign --verify --deep + --strict` confirms the binary is validly signed on disk. Gatekeeper's + policy — independent of local keychain trust — requires a Developer ID + Application certificate chained to Apple's root, or notarization. A + self-signed certificate satisfies neither, no matter how much you trust it + locally. + - The manual override, `sudo spctl --add --label ... `, is **no longer + supported on this macOS version** ("This operation is no longer + supported.") — Apple has removed the local allowlist escape hatch entirely. + - Conclusion: shipping this feature for real requires the existing signed + release pipeline (`script/package_macos.sh`, which already applies + `MACOS_CODESIGN_IDENTITY`) plus notarization. It is not something a local + dev loop can produce. Verify notification delivery only against a + Developer-ID-signed, notarized build. + +## Problem + +Two complaints, one shared root. + +**1. Reading a connected app asks for approval.** Asking Stram to read GitHub +produced an approval request even though the user had already completed the +GitHub OAuth flow. + +The stated cause ("GitHub access requires high-risk shell command execution") +was accurate but the framing was wrong. The gate did not fire on GitHub. It +fired on `run_shell_command`, which is `RiskLevel.HIGH` +(`stram/tools/files/implementation.py:982`), and `PolicyEngine` cannot see +connector state at all — `evaluate(self, tool, approved=False)` +(`stram/safety/policy.py:19`) receives no config, no provider id, no connector +runtime. It decides purely from `tool.risk_level` and `tool.requires_approval`. + +That approval was also a dead end: `ALLOWED_SHELL_COMMANDS = ("python", +"python.exe")` (`stram/tools/files/implementation.py:50`), so `git` and `gh` are +not allowlisted. Approving the token would have returned `BLOCKED`. + +The deeper cause: **no tool in the codebase calls `api.github.com`.** The GitHub +connector manifest is real and complete (`stram/connectors/providers/manifests.py:178-191` +— `api_base_url="https://api.github.com"`, scopes `("repo", "read:org", +"workflow")`, full OAuth), but its `tool_hints` point at +`github_repo_state_report_create`, `github_pr_packet_create`, +`github_issue_packet_create`, `ci_failure_report_create` — all of which write +local markdown artifacts (`stram/tools/github/implementation.py`). The shell was +the only path to GitHub, so the planner chose the shell. + +**2. Approvals surface only as chat text.** A run parked on an approval renders a +generic spinner labelled "Thinking / Waiting for permission" +(`apps/macos/Sources/Models.swift:1166`) with no token, tool name, risk, reason, +or buttons. The approval becomes actionable only by manually navigating to the +Permissions page (`apps/macos/Sources/RunsApprovalsViews.swift:80-155`). + +Worse, `refreshApprovals()` (`apps/macos/Sources/AppViewModel.swift:211-220`) has +**no timer** — it fires only on bootstrap, window activation, or a manual Refresh +button. A new approval may not appear on the Permissions page at all until the +user pokes it. Meanwhile the chat poll spins for 600 seconds because +`needs_approval` is deliberately excluded from `terminalChatStatuses` +(`AppViewModel.swift:565`). + +Neither desktop app has any modal or notification infrastructure. Greps for +`.sheet`, `.alert`, `confirmationDialog`, `NSAlert`, `ContentDialog`, +`MessageBox`, `Flyout`, `UNUserNotification` across both apps return zero +matches. The only interrupt pattern is a non-blocking `notice` banner +(`apps/macos/Sources/RootView.swift:20-38`). + +## Non-goals + +- Reclassifying existing tool risk levels. Existing `risk_level` and + `requires_approval` values stay exactly as they are. +- Replacing the Permissions pages. They remain the fallback surface. +- Wiring up the dead SSE paths (`AppViewModel.streamActivities`, + `AgentAPIClient.streamStimulus` — both zero-caller). Out of scope; polling is + sufficient and already the live pattern. +- Changing `ConnectorPolicy` (`stram/connectors/policy.py:7`), which enforces + connected-and-scoped at HTTP-call time. It stays as the second line of defence. + +## Design + +Three parts. Part 1 is the gate, Part 2 is what the gate opens onto, Part 3 is +the surface. + +### Part 1 — `read_only` metadata and a connector-aware policy + +`Tool` (`stram/tools/base.py:12-20`) gains two fields: + +```python +read_only: bool = False +provider_id: str | None = None +``` + +`PolicyEngine` takes `config` via its **constructor**, not via `evaluate`: +`PolicyEngine(config)`. This keeps the `evaluate(tool, approved)` signature +untouched, so every existing call site — `stram/executor.py:38` and the two in +`permissions_snapshot` (`stram/safety/permissions.py:31-32`) — needs no edit. + +Only the four construction sites change, and all four already have `config` in +scope (verified): `stram/orchestrator.py:52` (`self.config`), +`stram/runtime.py:31`, `stram/tools/workflow/implementation.py:655`, +`stram/safety/permissions.py:26`. + +Threading config through `evaluate` instead was considered and rejected: it is a +larger diff for no benefit, and it would have forced changes to the +`permissions_snapshot` call sites and likely to +`tests/test_api.py:255-276`. + +One wrinkle to be aware of: `Executor.execute` also receives a `config` +per call (`stram/executor.py:17`), which could in principle differ from the one +the `PolicyEngine` was built with. At all four sites today they are the same +object, so this is a latent inconsistency rather than a live bug. The +implementation should not try to reconcile them; it should use the +constructor-injected config and leave a comment noting the assumption. + +New decision order: + +1. `BLOCKED` → deny. Unchanged, and still wins first: a blocked tool marked + `read_only` is still blocked. +2. `read_only` **and** `provider_id` **and** that connector reports + `connected` → allow, no approval. +3. `read_only` **and** `provider_id is None` **and** a session grant exists for + this tool → allow, no approval. +4. `HIGH` → require approval. Unchanged. +5. `requires_approval` → require approval. Unchanged. +6. Allow. + +Connected state comes from `ConnectorTokenStatus.connected` +(`stram/connectors/models.py:66-68`), reached via +`ConnectorRuntime.readiness(provider_id)` (`stram/connectors/runtime.py:138`), +already re-exported for tool use at +`stram/integrations/workspace_connectors.py:11-65`. + +`_ToolAlias` (`stram/tools/__init__.py:44-53`) currently copies `risk_level`, +`requires_approval`, `input_schema`, and `capability_group` verbatim. It must +also copy `read_only` and `provider_id`, or every alias silently loses the +fast path. + +### Part 1b — session grants for privacy-gated reads + +Some read-only tools are gated for **privacy**, not mutation: +`os_clipboard_read` (`stram/tools/os_control/implementation.py:992`), +`screenshot_capture` (`:1150`), `os_observe_ui` (`:97`), +`screenpipe_search` (`stram/tools/external/implementation.py:482`), +`browser_live_screenshot` (`stram/tools/browser/live_tools.py:1535`). No OAuth +token speaks to these, so they have no `provider_id` and rule 2 never applies. + +They get rule 3 instead: prompt the first time each session, then run freely +until the runtime restarts. + +**Why this needs persistence.** `PolicyEngine()` is constructed fresh per +`AgentOrchestrator` (`stram/orchestrator.py:52`), and an orchestrator is built +per run. An in-memory grant set would die between runs, degrading "ask once per +session" into "ask every run" — i.e. today's behaviour. So grants persist. + +A `tool_grants` table is added to the existing approvals SQLite database +(`config.approvals_db_path`, already managed by `ApprovalStore` at +`stram/safety/approvals.py:34`) — no new file, no new connection management: + +```sql +CREATE TABLE IF NOT EXISTS tool_grants ( + tool_name TEXT NOT NULL, + session_id TEXT NOT NULL, + granted_at TEXT NOT NULL, + PRIMARY KEY (tool_name, session_id) +) +``` + +**Session identity.** The runtime writes `data_dir/session_id` containing a +fresh uuid4 at API server startup (`stram/api.py` serve entry point). Grants are +keyed to it, so restarting the app — which restarts the Python runtime the +desktop app spawns — invalidates every grant. Rows for stale session ids are +deleted on startup so the table cannot grow unbounded. + +`approve_pending_action` (`stram/runtime.py:21`) records a grant after a +successful approval when the approved tool is `read_only` with no `provider_id`. + +CLI runs (`stram run`) are each their own process with no `session_id` file +written by a server; they read the file if present and otherwise prompt every +run. That is the correct conservative default and is called out here so it is +not mistaken for a bug. + +**Test isolation.** Because grants live in the per-test tmp +`approvals_db_path` rather than a module global, tests cannot leak grants into +each other and no ordering dependency is introduced. + +### Part 2 — read-only GitHub tools + +Without these, Part 1 is inert for GitHub: there is no read-only GitHub tool to +un-gate, and the planner keeps reaching for the shell. + +Four new tools in `stram/tools/github/implementation.py`, all `RiskLevel.LOW`, +`read_only=True`, `provider_id="github"`: + +| tool | endpoint | +|---|---| +| `github_repos_list` | `GET /user/repos` | +| `github_issues_list` | `GET /repos/{owner}/{repo}/issues` | +| `github_pulls_list` | `GET /repos/{owner}/{repo}/pulls` | +| `github_checks_list` | `GET /repos/{owner}/{repo}/commits/{ref}/check-runs` | + +They call the API through the existing `ConnectorHttpClient`, so +`ConnectorPolicy.check_scopes` (`stram/connectors/policy.py:7`) still raises on +an unconnected or under-scoped token — meaning an unconnected GitHub fails with a +clear error rather than silently skipping approval. + +The manifest's `tool_hints` (`stram/connectors/providers/manifests.py:185`) is +updated to list these four, so the planner prefers them over +`run_shell_command`. + +### Part 3 — native approval notifications + +**Shared prerequisite: an approval detector.** Nothing currently notices a new +approval. Both apps add a repeating ~2s poll of the existing +`GET /approvals?status=pending` endpoint, holding a set of already-seen tokens +and raising one notification per unseen token. This reuses endpoints both apps +already call and needs no server change. The poll stops when no run is active. + +**macOS.** `UNUserNotificationCenter` with a `UNNotificationCategory("APPROVAL")` +carrying Approve and Reject actions, plus an `NSApplicationDelegate` adopting +`UNUserNotificationCenterDelegate` to handle the response. Authorization is +requested once on first launch. + +Constraint: `UNUserNotificationCenter` requires a bundle identifier, so it does +not work under `swift run`. Dev testing must go through +`script/build_and_run.sh`. The generated Info.plist already sets +`CFBundleIdentifier` (`script/package_macos.sh:103`), so no packaging change is +needed. + +**Windows.** `AppNotificationManager` from WinAppSDK, which the app already +references at 1.6.240829007 (`apps/windows/Stram.App/Stram.App.csproj`). The app +is unpackaged (`WindowsPackageType=None`), which is supported: unpackaged +notification support landed in WinAppSDK 1.2. Requires calling +`AppNotificationManager.Default.Register()` at startup and handling +`NotificationInvoked`. No MSIX migration required. + +Caveat: unpackaged activation resolves the exe by path, so moving the +installed app breaks action buttons until it is relaunched once. Acceptable for +a locally installed desktop app. + +**Both** route their action buttons to the endpoints already in use: +`POST /approvals/{token}/approve` and `POST /approvals/{token}/reject` +(`stram/api.py:1364-1379`). Windows already has an approval note field +(`MainWindow.xaml:864`); macOS keeps its hardcoded note. + +A notification is dismissible in a way a modal is not, so both Permissions +pages stay exactly as they are, as the recovery surface for a dismissed +notification. + +## Error handling + +- Connector lookup failure inside `evaluate` (missing DB, corrupt token row) + fails **closed** — treated as not connected, so the approval prompt appears. + A read-only fast path must never open because a check errored. +- `BLOCKED` is evaluated before any fast path, so no combination of + `read_only` and a connected provider can run a blocked tool. +- Notification authorization denied: the app falls back silently to the + existing Permissions page. No repeated nagging. +- Approval token already decided (approved elsewhere, or run cancelled): the + existing endpoints already raise on non-pending tokens + (`stram/safety/approvals.py:128`). The notification handler surfaces that via + the existing `notice` / `InfoBar` banner and drops the token from the seen set. + +## Testing + +Policy, in `tests/test_policy.py`: + +- read-only tool + connected provider → allowed, no approval +- read-only tool + provider not connected → approval required +- read-only tool + provider lookup raises → approval required (fails closed) +- `BLOCKED` + `read_only=True` + connected → still denied +- privacy read (no provider) → approval required, then allowed after a grant +- privacy read with a grant from a *different* session id → approval required + +Grants, in `tests/test_approval_queue.py`: + +- approving a privacy read writes a `tool_grants` row +- approving a provider-backed or non-read-only tool writes no grant +- stale-session rows are purged at startup + +GitHub tools, in `tests/test_tools.py`: + +- the four new tools report `read_only=True`, `provider_id="github"`, `LOW` +- unconnected GitHub → clean failure, not a silent skip +- aliases of a read-only tool preserve `read_only` and `provider_id` + +Regression surface to keep green: `tests/test_approvals_security.py` (all four +tests are `run_shell_command`-based and must not change), +`tests/test_api.py:255-276` (`permissions_snapshot` shape — unchanged, because +constructor injection leaves `evaluate` alone), +`tests/test_executor.py:247-260,366-376` (clipboard and live-screenshot still +`NEEDS_APPROVAL` on first call). + +`tests/test_planning.py` embeds roughly 200 `"requires_approval"` values in +planner fixtures. Because no existing tool's `risk_level` or `requires_approval` +changes, these stay stable; the four new tools add entries rather than editing +existing ones. + +## Risks + +- **The largest risk is Part 1 touching the central gate.** Mitigated by + ordering `BLOCKED` first, failing closed on lookup errors, and adding + `read_only=True` only to new tools plus the five named privacy reads. +- A dismissed notification is a missed approval. Mitigated by keeping both + Permissions pages and their pending counts. +- `permissions_snapshot` (`stram/safety/permissions.py:30`) evaluates every tool + with `approved=False`; once policy is connector-aware its output becomes + connector-dependent. Callers must not cache it across a connect/disconnect. diff --git a/stram/agent/store.py b/stram/agent/store.py index 4551e5d..c9df4c3 100644 --- a/stram/agent/store.py +++ b/stram/agent/store.py @@ -31,6 +31,7 @@ class AgentStore: def __init__(self, path: Path) -> None: self.path = path self.path.parent.mkdir(parents=True, exist_ok=True) + _migrate_legacy_agent_files(self.path) self._init_db() def _connect(self) -> sqlite3.Connection: @@ -41,6 +42,7 @@ def _connect(self) -> sqlite3.Connection: def _init_db(self) -> None: with closing(self._connect()) as connection: + _migrate_legacy_agent_tables(connection) connection.execute( """ CREATE TABLE IF NOT EXISTS active_event_routes ( @@ -2058,6 +2060,27 @@ def _json_loads_list(value: str) -> list[Any]: return parsed if isinstance(parsed, list) else [] +def _migrate_legacy_agent_files(path: Path) -> None: + """Carry a pre-rename janus.sqlite3 database (and its WAL sidecars) over to the new name.""" + legacy = path.with_name("janus.sqlite3") + if path.exists() or not legacy.exists(): + return + for suffix in ("", "-wal", "-shm"): + source = legacy.with_name(legacy.name + suffix) + if source.exists(): + source.rename(path.with_name(path.name + suffix)) + + +def _migrate_legacy_agent_tables(connection: sqlite3.Connection) -> None: + """Rename the pre-rename janus_activations table so existing activations survive the upgrade.""" + names = { + str(row[0]) + for row in connection.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall() + } + if "janus_activations" in names and "agent_activations" not in names: + connection.execute("ALTER TABLE janus_activations RENAME TO agent_activations") + + def _ensure_column(connection: sqlite3.Connection, table: str, column: str, definition: str) -> None: rows = connection.execute(f"PRAGMA table_info({table})").fetchall() if any(str(row[1]) == column for row in rows): diff --git a/stram/api.py b/stram/api.py index 3f34715..65b88b1 100644 --- a/stram/api.py +++ b/stram/api.py @@ -4,6 +4,7 @@ import binascii import json import os +import signal import threading import time import uuid @@ -115,6 +116,7 @@ from stram.memory.summary import summarize_memory from stram.orchestrator import AgentOrchestrator from stram.performance import run_benchmarks +from stram.safety.grants import clear_session, start_session from stram.runtime import ( approval_record_to_dict, approve_pending_action, @@ -1582,6 +1584,7 @@ def __init__(self, server_address: tuple[str, int], handler_class: type[BaseHTTP self._background_threads: list[threading.Thread] = [] self._background_threads_lock = threading.Lock() self._stop_event = threading.Event() + self.session_config: AgentConfig | None = None super().__init__(server_address, handler_class) def start_background_worker(self, worker: threading.Thread) -> None: @@ -1606,6 +1609,8 @@ def join_background_workers(self, timeout_seconds: float = 10.0) -> None: def server_close(self) -> None: self._stop_event.set() self.join_background_workers() + if self.session_config is not None: + clear_session(self.session_config) super().server_close() @@ -1613,6 +1618,8 @@ def create_api_server(config: AgentConfig, host: str = "127.0.0.1", port: int = if host not in {"127.0.0.1", "localhost", "::1"}: raise ValueError("Stram API binds to loopback hosts only by default.") server = StramAPIServer((host, port), make_handler(config)) + server.session_config = config.normalized() + start_session(server.session_config) server.start_background_worker( threading.Thread( target=_collector_background_worker, @@ -1708,6 +1715,17 @@ def run_api_server(config: AgentConfig, host: str = "127.0.0.1", port: int = 876 server = create_api_server(config, host=host, port=port) address, actual_port = server.server_address print(f"Stram API listening on http://{address}:{actual_port}") + + def _stop_on_sigterm(signum: int, frame: Any) -> None: + # Reuse the Ctrl-C path: raising here unblocks serve_forever in the main + # thread. Calling server.shutdown() from the handler would deadlock. + raise KeyboardInterrupt + + try: + signal.signal(signal.SIGTERM, _stop_on_sigterm) + except ValueError: + pass # not the main thread; Ctrl-C path still applies + # No handler restore: the only caller is `stram serve`, whose process exits here. try: server.serve_forever() except KeyboardInterrupt: diff --git a/stram/connectors/providers/manifests.py b/stram/connectors/providers/manifests.py index f120987..83ee834 100644 --- a/stram/connectors/providers/manifests.py +++ b/stram/connectors/providers/manifests.py @@ -182,7 +182,16 @@ api_base_url="https://api.github.com", default_scopes=("repo", "read:org", "workflow"), workspace_apps=("GitHub", "Issues", "Pull Requests", "Actions"), - tool_hints=("github_repo_state_report_create", "github_pr_packet_create", "github_issue_packet_create", "ci_failure_report_create"), + tool_hints=( + "github_repos_list", + "github_issues_list", + "github_pulls_list", + "github_checks_list", + "github_repo_state_report_create", + "github_pr_packet_create", + "github_issue_packet_create", + "ci_failure_report_create", + ), auth_url="https://github.com/login/oauth/authorize", token_url="https://github.com/login/oauth/access_token", credential_fields=("client_id", "client_secret"), diff --git a/stram/executor.py b/stram/executor.py index 0c23b04..cd33f40 100644 --- a/stram/executor.py +++ b/stram/executor.py @@ -35,7 +35,7 @@ def execute(self, step: PlannedStep, config: AgentConfig, approved: bool = False output={"input_schema": tool.input_schema}, error=str(exc), ) - decision = self.policy.evaluate(tool, approved=approved) + decision = self.policy.evaluate(tool, approved=approved, tool_input=step.tool_input) if not decision.allowed: status = ActionStatus.NEEDS_APPROVAL if decision.requires_approval else ActionStatus.BLOCKED output = {} diff --git a/stram/orchestrator.py b/stram/orchestrator.py index 9cfa30a..e9abe30 100644 --- a/stram/orchestrator.py +++ b/stram/orchestrator.py @@ -49,7 +49,7 @@ def __init__(self, config: AgentConfig) -> None: self.audit = AuditLog(self.config.audit_db_path) self.approvals = ApprovalStore(self.config.approvals_db_path) self.memory = EventStore(self.config.memory_db_path) - self.executor = Executor(self.tools, PolicyEngine()) + self.executor = Executor(self.tools, PolicyEngine(self.config)) def _build_plan_provider(self) -> PlanProvider: fallback = ExplicitFallbackPlanProvider(set(self.tools.keys())) diff --git a/stram/process.py b/stram/process.py new file mode 100644 index 0000000..b397087 --- /dev/null +++ b/stram/process.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +import os + + +def pid_alive(pid: int) -> bool: + """Best-effort liveness probe that is safe on every platform. + + `os.kill(pid, 0)` is NOT a probe on Windows: CPython opens the process with + PROCESS_ALL_ACCESS and calls TerminateProcess for any signal other than + CTRL_C_EVENT / CTRL_BREAK_EVENT, sig=0 included. So never call it there. + Anything unknown is reported dead, so callers fail closed. + """ + if not 0 < pid < 2**31: + return False + if pid == os.getpid(): + return True # e.g. the daemon evaluating its own session + if os.name == "nt": + # No safe stdlib cross-process probe on Windows, and a non-owner process + # must never inherit an owner's grant. + return False + try: + os.kill(pid, 0) + except Exception: + return False + return True diff --git a/stram/runtime.py b/stram/runtime.py index 94ddf6f..aa65aad 100644 --- a/stram/runtime.py +++ b/stram/runtime.py @@ -8,6 +8,7 @@ from stram.memory.event_store import EventStore from stram.safety.approvals import ApprovalRecord, ApprovalStore from stram.safety.audit import AuditLog +from stram.safety.grants import ToolGrantStore, current_session_id from stram.safety.policy import PolicyEngine from stram.schemas import ActionStatus, PlannedStep from stram.tools import default_tools @@ -28,7 +29,7 @@ def approve_pending_action(config: AgentConfig, approval_token: str, note: str) audit = AuditLog(config.audit_db_path) memory = EventStore(config.memory_db_path) - executor = Executor(default_tools(config), PolicyEngine()) + executor = Executor(default_tools(config), PolicyEngine(config)) run_id = record.run_id audit.log_run_event( run_id, @@ -60,6 +61,21 @@ def approve_pending_action(config: AgentConfig, approval_token: str, note: str) {"status": status.value, "approval_token": approval_token}, ) updated = approval_store.mark_executed(approval_token, tool_result, note=note) + approved_tool = executor.tools.get(record.tool_name) + try: + if ( + tool_result.status == ActionStatus.SUCCEEDED + and approved_tool is not None + and approved_tool.read_only + and not approved_tool.provider_id + ): + session_id = current_session_id(config) + if session_id: + ToolGrantStore(config.approvals_db_path).record(record.tool_name, session_id, record.tool_input) + except Exception: + # The action already succeeded. A store hiccup here only costs the user + # another prompt next time, which is the safe direction. + pass memory.append( "approval_decision", { diff --git a/stram/safety/grants.py b/stram/safety/grants.py new file mode 100644 index 0000000..31d50d7 --- /dev/null +++ b/stram/safety/grants.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import json +import os +import sqlite3 +from contextlib import closing +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any +from uuid import uuid4 + +from stram.config import AgentConfig +from stram.process import pid_alive + +SESSION_FILE_NAME = "session_id" +SESSION_PID_FILE_NAME = "session_pid" +# Session grants authorize ambient reads (whatever is on screen / in the clipboard +# now), so argument equality bounds the request but not the disclosure. Time-box it. +GRANT_TTL_SECONDS = 300 + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _session_path(config: AgentConfig) -> Path: + return config.data_dir / SESSION_FILE_NAME + + +def _session_pid_path(config: AgentConfig) -> Path: + return config.data_dir / SESSION_PID_FILE_NAME + + +def _session_owner_alive(config: AgentConfig) -> bool: + """True only when the server process that minted the session is still running. + + A killed server (Windows' Process.Kill, or any crash) cannot run cleanup, so + the session file alone is not evidence of a live session. Anything unreadable, + unparseable or unreachable is treated as dead. + """ + try: + pid = int(_session_pid_path(config).read_text(encoding="utf-8").strip()) + except Exception: + return False + return pid_alive(pid) + + +def current_session_id(config: AgentConfig) -> str: + """Return the live runtime's session id, or "" when no server owns one.""" + if not _session_owner_alive(config): + return "" + try: + return _session_path(config).read_text(encoding="utf-8").strip() + except (OSError, UnicodeDecodeError): + return "" + + +def start_session(config: AgentConfig) -> str: + """Mint a fresh session id and drop every grant from previous sessions.""" + session_id = str(uuid4()) + path = _session_path(config) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(session_id, encoding="utf-8") + _session_pid_path(config).write_text(str(os.getpid()), encoding="utf-8") + ToolGrantStore(config.approvals_db_path).purge_other_sessions(session_id) + return session_id + + +def clear_session(config: AgentConfig) -> None: + """End the session: forget the session id and every grant tied to it.""" + for path in (_session_path(config), _session_pid_path(config)): + try: + path.unlink(missing_ok=True) + except OSError: + pass + ToolGrantStore(config.approvals_db_path).purge_all() + + +class ToolGrantStore: + """Session-scoped 'ask once' grants for read-only tools with no connector.""" + + def __init__(self, db_path: Path) -> None: + self.db_path = db_path + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self._init_db() + + def _connect(self) -> sqlite3.Connection: + return sqlite3.connect(self.db_path) + + def _init_db(self) -> None: + with closing(self._connect()) as connection: + columns = {row[1] for row in connection.execute("PRAGMA table_info(tool_grants)")} + if columns and "tool_input" not in columns: + # Grants are ephemeral session data; dropping is safer than migrating. + connection.execute("DROP TABLE tool_grants") + connection.execute( + """ + CREATE TABLE IF NOT EXISTS tool_grants ( + tool_name TEXT NOT NULL, + session_id TEXT NOT NULL, + tool_input TEXT NOT NULL, + granted_at TEXT NOT NULL, + PRIMARY KEY (tool_name, session_id, tool_input) + ) + """ + ) + connection.commit() + + def record(self, tool_name: str, session_id: str, tool_input: dict[str, Any] | None) -> None: + # Symmetric with has(): an unknown input (None) must never become a grant that + # a later {} call matches. + if not tool_name or not session_id or tool_input is None: + return + with closing(self._connect()) as connection: + connection.execute( + """ + INSERT OR REPLACE INTO tool_grants (tool_name, session_id, tool_input, granted_at) + VALUES (?, ?, ?, ?) + """, + (tool_name, session_id, _canonical(tool_input), _now()), + ) + connection.commit() + + def has(self, tool_name: str, session_id: str, tool_input: dict[str, Any] | None) -> bool: + """True only when this exact argument set was already approved this session. + + Exact equality, not subset: every gated tool resolves an omitted argument to + a default that is broader than any explicit value (content_type="all", + max_chars=4000, no time bound), so a call that drops a key asks for MORE, + not less. Unknown arguments (None) never match. + + Grants also expire after GRANT_TTL_SECONDS: these tools read ambient state, + so the same arguments disclose something different an hour later. + """ + if not tool_name or not session_id or tool_input is None: + return False + with closing(self._connect()) as connection: + row = connection.execute( + "SELECT granted_at FROM tool_grants WHERE tool_name = ? AND session_id = ? AND tool_input = ?", + (tool_name, session_id, _canonical(tool_input)), + ).fetchone() + if row is None: + return False + try: + granted_at = datetime.fromisoformat(row[0]) + except (TypeError, ValueError): + return False + if granted_at.tzinfo is None: + granted_at = granted_at.replace(tzinfo=timezone.utc) + return datetime.now(timezone.utc) - granted_at < timedelta(seconds=GRANT_TTL_SECONDS) + + def purge_other_sessions(self, session_id: str) -> None: + if not session_id: + return + with closing(self._connect()) as connection: + connection.execute("DELETE FROM tool_grants WHERE session_id != ?", (session_id,)) + connection.commit() + + def purge_all(self) -> None: + with closing(self._connect()) as connection: + connection.execute("DELETE FROM tool_grants") + connection.commit() + + +def _canonical(tool_input: dict[str, Any] | None) -> str: + return json.dumps(tool_input or {}, ensure_ascii=False, sort_keys=True) diff --git a/stram/safety/permissions.py b/stram/safety/permissions.py index 87e1a18..79f07c4 100644 --- a/stram/safety/permissions.py +++ b/stram/safety/permissions.py @@ -23,7 +23,7 @@ def permissions_snapshot( index_status: dict[str, Any] | None = None, ) -> dict[str, Any]: normalized = config.normalized() - policy = PolicyEngine() + policy = PolicyEngine(normalized) tools = [] groups: dict[str, dict[str, Any]] = {} plugin_manifests = discover_plugin_manifests(normalized) diff --git a/stram/safety/policy.py b/stram/safety/policy.py index c14657c..3f07ec0 100644 --- a/stram/safety/policy.py +++ b/stram/safety/policy.py @@ -1,7 +1,11 @@ from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass +from typing import Any +from stram.config import AgentConfig +from stram.safety.grants import ToolGrantStore, current_session_id from stram.schemas import RiskLevel from stram.tools.base import Tool @@ -14,11 +18,39 @@ class PolicyDecision: class PolicyEngine: - """Central action gate. All tool calls pass through here before execution.""" + """Central action gate. All tool calls pass through here before execution. - def evaluate(self, tool: Tool, approved: bool = False) -> PolicyDecision: + Assumes the config given here is the same one Executor.execute is called + with; at every construction site today it is. + """ + + def __init__( + self, + config: AgentConfig | None = None, + *, + connected_lookup: Callable[[str], bool] | None = None, + ) -> None: + self.config = config + self._connected_lookup = connected_lookup + self._connected_cache: dict[str, bool] = {} + self._session_id: str | None = None + self._grants: ToolGrantStore | None = None + + def evaluate( + self, + tool: Tool, + approved: bool = False, + *, + tool_input: dict[str, Any] | None = None, + ) -> PolicyDecision: if tool.risk_level == RiskLevel.BLOCKED: return PolicyDecision(False, False, "Tool is blocked by policy.") + if tool.read_only: + if tool.provider_id: + if self._provider_connected(tool.provider_id): + return PolicyDecision(True, False, f"Read-only action on connected {tool.provider_id}.") + elif tool_input is not None and self._granted_this_session(tool.name, tool_input): + return PolicyDecision(True, False, "Read-only action already allowed this session.") if tool.risk_level == RiskLevel.HIGH: if approved: return PolicyDecision(True, True, "High-risk action approved.") @@ -26,3 +58,35 @@ def evaluate(self, tool: Tool, approved: bool = False) -> PolicyDecision: if tool.requires_approval and not approved: return PolicyDecision(False, True, "Tool requires explicit approval.") return PolicyDecision(True, tool.requires_approval, "Allowed by local policy.") + + def _provider_connected(self, provider_id: str) -> bool: + if self.config is None: + return False + if provider_id in self._connected_cache: + return self._connected_cache[provider_id] + connected = False + try: + if self._connected_lookup is not None: + connected = bool(self._connected_lookup(provider_id)) + else: + from stram.connectors import ConnectorRuntime + + connected = bool(ConnectorRuntime(self.config).readiness(provider_id).get("connected")) + except Exception: + connected = False + self._connected_cache[provider_id] = connected + return connected + + def _granted_this_session(self, tool_name: str, tool_input: dict[str, Any]) -> bool: + if self.config is None: + return False + try: + if self._session_id is None: + self._session_id = current_session_id(self.config) + if not self._session_id: + return False + if self._grants is None: + self._grants = ToolGrantStore(self.config.approvals_db_path) + return self._grants.has(tool_name, self._session_id, tool_input) + except Exception: + return False diff --git a/stram/tools/__init__.py b/stram/tools/__init__.py index 81eaa71..2ebeed4 100644 --- a/stram/tools/__init__.py +++ b/stram/tools/__init__.py @@ -50,6 +50,8 @@ def __init__(self, alias: str, target: Tool) -> None: requires_approval=target.requires_approval, input_schema=target.input_schema, capability_group=target.capability_group, + read_only=target.read_only, + provider_id=target.provider_id, ) self._target = target diff --git a/stram/tools/base.py b/stram/tools/base.py index 0636535..c8e9413 100644 --- a/stram/tools/base.py +++ b/stram/tools/base.py @@ -16,6 +16,8 @@ class Tool(ABC): requires_approval: bool = False input_schema: dict[str, Any] = field(default_factory=lambda: {"type": "object", "properties": {}}) capability_group: str = "core" + read_only: bool = False + provider_id: str | None = None @abstractmethod def execute(self, tool_input: dict[str, Any], config: AgentConfig) -> ToolResult: diff --git a/stram/tools/browser/live_tools.py b/stram/tools/browser/live_tools.py index 0965e7c..f4e2972 100644 --- a/stram/tools/browser/live_tools.py +++ b/stram/tools/browser/live_tools.py @@ -1534,6 +1534,7 @@ def __init__(self) -> None: description="Save a screenshot of a Playwright-backed live browser session after explicit approval.", risk_level=RiskLevel.HIGH, requires_approval=True, + read_only=True, input_schema=object_input_schema( { "live_session_id": {"type": "string"}, diff --git a/stram/tools/external/implementation.py b/stram/tools/external/implementation.py index 029a19b..2ed5b05 100644 --- a/stram/tools/external/implementation.py +++ b/stram/tools/external/implementation.py @@ -486,6 +486,7 @@ def __init__(self) -> None: ), risk_level=RiskLevel.MEDIUM, requires_approval=True, + read_only=True, input_schema=object_input_schema( { "query": {"type": "string", "description": "Natural-language or keyword search query."}, diff --git a/stram/tools/files/implementation.py b/stram/tools/files/implementation.py index 4c1d870..9c346ca 100644 --- a/stram/tools/files/implementation.py +++ b/stram/tools/files/implementation.py @@ -13,6 +13,7 @@ from typing import Any from stram.config import AgentConfig +from stram.process import pid_alive from stram.schemas import ActionStatus, RiskLevel, ToolResult from stram.tools.base import Tool, object_input_schema @@ -1245,15 +1246,11 @@ def _save_process_records(config: AgentConfig, records: dict[str, dict[str, Any] def _pid_status(pid: int) -> str: + # pid_alive, not os.kill(pid, 0): on Windows that call terminates the process + # this function is only meant to be querying. if pid <= 0: return "unknown" - try: - os.kill(pid, 0) - except ProcessLookupError: - return "exited" - except PermissionError: - return "unknown" - return "running" + return "running" if pid_alive(pid) else "exited" def _tail_text(path: Path, limit: int = 4000) -> str: diff --git a/stram/tools/github/implementation.py b/stram/tools/github/implementation.py index cea0f82..e6550e4 100644 --- a/stram/tools/github/implementation.py +++ b/stram/tools/github/implementation.py @@ -3,6 +3,7 @@ from datetime import datetime, timezone import json from pathlib import Path +import re from typing import Any from uuid import uuid4 @@ -318,6 +319,113 @@ def execute(self, tool_input: dict[str, Any], config: AgentConfig) -> ToolResult ) +class GitHubReadTool(Tool): + """Read-only GitHub API call routed through the workspace connector.""" + + def __init__( + self, + name: str, + description: str, + *, + operation: str, + path_template: str, + required_scopes: tuple[str, ...], + properties: dict[str, dict[str, Any]], + required: list[str], + ) -> None: + super().__init__( + name=name, + description=description, + # MEDIUM + requires_approval so the policy engine's connected-provider rule + # decides: connected GitHub reads run prompt-free, a disconnected one asks + # instead of failing deep in the connector with a raw PermissionError. + risk_level=RiskLevel.MEDIUM, + requires_approval=True, + input_schema=object_input_schema( + { + **properties, + "per_page": { + "type": "integer", + "description": "Maximum items to return (1-100).", + }, + }, + required=required, + ), + capability_group="github", + read_only=True, + provider_id="github", + ) + self._operation = operation + self._path_template = path_template + self._required_scopes = required_scopes + + def execute(self, tool_input: dict[str, Any], config: AgentConfig) -> ToolResult: + from stram.connectors import ConnectorOperationRequest, ConnectorRuntime + + try: + path = self._path_template.format(**{key: _github_path_segment(tool_input, key) for key in _template_keys(self._path_template)}) + except ValueError as exc: + return ToolResult(self.name, ActionStatus.FAILED, self.risk_level, str(exc), error=str(exc)) + + per_page = tool_input.get("per_page") + try: + per_page_value = max(1, min(int(per_page), 100)) if per_page is not None else 30 + except (TypeError, ValueError, OverflowError): + per_page_value = 30 + + request = ConnectorOperationRequest( + provider_id="github", + operation=self._operation, + method="GET", + path=path, + query={"per_page": per_page_value}, + required_scopes=self._required_scopes, + reason=f"Read-only GitHub metadata for {self.name}.", + ) + try: + result = ConnectorRuntime(config).execute_operation(request) + except (ValueError, PermissionError) as exc: + return ToolResult(self.name, ActionStatus.FAILED, self.risk_level, str(exc), error=str(exc)) + except Exception as exc: + return ToolResult(self.name, ActionStatus.FAILED, self.risk_level, f"{self.name} failed.", error=str(exc)) + + response = result.get("response") + items = response if isinstance(response, list) else [response] + trimmed = items[:MAX_GITHUB_ITEMS] + return ToolResult( + self.name, + ActionStatus.SUCCEEDED, + self.risk_level, + f"Read {len(trimmed)} item(s) from GitHub via {self._operation}.", + { + "operation": self._operation, + "path": path, + "status_code": result.get("status_code"), + "count": len(trimmed), + "items": trimmed, + }, + ) + + +def _template_keys(template: str) -> tuple[str, ...]: + return tuple(re.findall(r"\{([a-zA-Z0-9_]+)\}", template)) + + +_SEGMENT_RE = re.compile(r"[A-Za-z0-9_.-]+") + + +def _github_path_segment(tool_input: dict[str, Any], key: str) -> str: + value = str(tool_input.get(key) or "").strip().strip("/") + segments = value.split("/") + if key != "repo" and len(segments) > 1: + raise ValueError(f"{key} must be a single path segment.") + if not value or len(segments) > 2 or any( + segment in (".", "..") or not _SEGMENT_RE.fullmatch(segment) for segment in segments + ): + raise ValueError(f"{key} must be plain path segment(s).") + return value + + def default_github_tools() -> dict[str, Tool]: tools: list[Tool] = [ GitHubIssueDraftCreateTool(), @@ -328,6 +436,45 @@ def default_github_tools() -> dict[str, Tool]: GitHubRepoStateReportCreateTool(), GitHubWorkflowArtifactInspectTool(), GitHubWorkflowArtifactInspectTool("github_artifact_inspect"), + GitHubReadTool( + "github_repos_list", + "List repositories the connected GitHub account can access.", + operation="github_repos_list", + path_template="/user/repos", + required_scopes=("repo",), + properties={}, + required=[], + ), + GitHubReadTool( + "github_issues_list", + "List open issues for a repository, given repo as 'owner/name'.", + operation="github_issues_list", + path_template="/repos/{repo}/issues", + required_scopes=("repo",), + properties={"repo": {"type": "string", "description": "Repository as owner/name."}}, + required=["repo"], + ), + GitHubReadTool( + "github_pulls_list", + "List pull requests for a repository, given repo as 'owner/name'.", + operation="github_pulls_list", + path_template="/repos/{repo}/pulls", + required_scopes=("repo",), + properties={"repo": {"type": "string", "description": "Repository as owner/name."}}, + required=["repo"], + ), + GitHubReadTool( + "github_checks_list", + "List CI check runs for a commit ref in a repository.", + operation="github_checks_list", + path_template="/repos/{repo}/commits/{ref}/check-runs", + required_scopes=("repo",), + properties={ + "repo": {"type": "string", "description": "Repository as owner/name."}, + "ref": {"type": "string", "description": "Commit SHA, branch, or tag."}, + }, + required=["repo", "ref"], + ), ] return {tool.name: tool for tool in tools} diff --git a/stram/tools/os_control/implementation.py b/stram/tools/os_control/implementation.py index d805a8f..1ab722b 100644 --- a/stram/tools/os_control/implementation.py +++ b/stram/tools/os_control/implementation.py @@ -96,6 +96,7 @@ def __init__(self) -> None: ), risk_level=RiskLevel.HIGH, requires_approval=True, + read_only=True, input_schema=object_input_schema( { "max_elements": { @@ -991,6 +992,7 @@ def __init__(self) -> None: description="Read current Windows clipboard text after approval. Clipboard contents can be sensitive.", risk_level=RiskLevel.HIGH, requires_approval=True, + read_only=True, input_schema=object_input_schema( { "max_chars": {"type": "integer", "minimum": 1, "maximum": 20000, "description": "Maximum clipboard characters to return."}, @@ -1149,6 +1151,7 @@ def __init__(self) -> None: ), risk_level=RiskLevel.HIGH, requires_approval=True, + read_only=True, input_schema=object_input_schema( { "reason": { diff --git a/stram/tools/workflow/implementation.py b/stram/tools/workflow/implementation.py index 30ebf60..c6e26b6 100644 --- a/stram/tools/workflow/implementation.py +++ b/stram/tools/workflow/implementation.py @@ -652,7 +652,7 @@ def _run_workflow_until_blocked(config: AgentConfig, workflow: dict[str, Any], * from stram.tools import default_tools tools = default_tools(config) - executor = Executor(tools, PolicyEngine()) + executor = Executor(tools, PolicyEngine(config)) for step in workflow["steps"]: if step["status"] in {"succeeded", "skipped"}: continue diff --git a/tests/test_agent.py b/tests/test_agent.py index 8a42967..1155291 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -1,6 +1,8 @@ import json +import sqlite3 import tempfile import unittest +from contextlib import closing from pathlib import Path from unittest.mock import patch @@ -26,7 +28,7 @@ select_activity_guides, validate_activity_guides, ) -from stram.agent.models import RouteClass +from stram.agent.models import RouteClass, TaskContext from stram.cognition.knowledge import KnowledgeStore from stram.cognition import FocusStore from stram.collectors.consumers.agent import AgentConsumer, agent_consumer_name @@ -37,6 +39,45 @@ from stram.planning.model_clients import ModelClientError, StaticModelClient +class LegacyAgentStoreMigrationTests(unittest.TestCase): + def test_legacy_janus_database_and_activations_survive_the_rename(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + data_dir = Path(tmp_dir) + legacy_path = data_dir / "janus.sqlite3" + agent_path = data_dir / "agent.sqlite3" + + legacy = AgentStore(legacy_path) + legacy.upsert_task_context( + TaskContext( + task_context_id="ctx-legacy", + status="active", + source="user_declared", + user_declared_goal="Survive the rename.", + episode_id="episode-legacy", + assistant_mode="supportive", + privacy_mode="metadata_first", + summary="Task context written before the rename.", + ) + ) + with closing(sqlite3.connect(legacy_path)) as connection: + connection.execute("ALTER TABLE agent_activations RENAME TO janus_activations") + connection.commit() + del legacy + + store = AgentStore(agent_path) + + self.assertFalse(legacy_path.exists()) + self.assertTrue(agent_path.exists()) + self.assertEqual([context["task_context_id"] for context in store.task_contexts()], ["ctx-legacy"]) + with closing(sqlite3.connect(agent_path)) as connection: + tables = { + str(row[0]) + for row in connection.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall() + } + self.assertIn("agent_activations", tables) + self.assertNotIn("janus_activations", tables) + + class AgentTests(unittest.TestCase): def test_context_events_are_stored_without_model_decision(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: diff --git a/tests/test_approval_queue.py b/tests/test_approval_queue.py index 1e1698d..49346ea 100644 --- a/tests/test_approval_queue.py +++ b/tests/test_approval_queue.py @@ -1,12 +1,40 @@ import tempfile import unittest from pathlib import Path +from unittest.mock import patch from stram.config import AgentConfig from stram.orchestrator import AgentOrchestrator from stram.runtime import approve_pending_action, update_pending_approval_input from stram.safety.approvals import ApprovalStore from stram.safety.audit import AuditLog +from stram.schemas import ActionStatus, RiskLevel, ToolResult +from stram.tools.base import Tool, object_input_schema + + +class FakeReadTool(Tool): + """Read-only stand-in that succeeds, or is skipped under dry_run.""" + + def execute(self, tool_input, config): + status = ActionStatus.SKIPPED if config.dry_run else ActionStatus.SUCCEEDED + return ToolResult( + tool_name=self.name, + status=status, + risk_level=self.risk_level, + summary=f"{self.name} {status.value}", + ) + + +def fake_read_tool(name: str, provider_id: str | None = None) -> FakeReadTool: + return FakeReadTool( + name=name, + description="fake read-only tool", + risk_level=RiskLevel.HIGH, + requires_approval=True, + input_schema=object_input_schema({"reason": {"type": "string"}}, ["reason"]), + read_only=True, + provider_id=provider_id, + ) class ApprovalQueueTests(unittest.TestCase): @@ -90,6 +118,104 @@ def test_pending_approval_edit_validates_tool_input(self) -> None: pending = ApprovalStore(config.approvals_db_path).get(token) self.assertEqual(pending.tool_input, {"argv": ["python", "--version"]}) + def _approve_fake_tool(self, config: AgentConfig, tool: Tool, tool_input: dict) -> None: + from stram.schemas import ApprovalRequest + + ApprovalStore(config.approvals_db_path).create_pending( + f"run-{tool.name}", + f"use {tool.name}", + ApprovalRequest( + tool_name=tool.name, + tool_input=tool_input, + risk_level=RiskLevel.HIGH, + reason="privacy read", + approval_token=f"token-{tool.name}", + ), + ) + with patch("stram.runtime.default_tools", return_value={tool.name: tool}): + approve_pending_action(config, f"token-{tool.name}", "approved in test") + + def test_approving_privacy_read_records_session_grant(self) -> None: + from stram.safety.grants import ToolGrantStore, start_session + + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + config = AgentConfig(workspace=workspace, data_dir=workspace / "artifacts", planner_provider="explicit").normalized() + + session = start_session(config) + tool_input = {"reason": "read the copied link"} + self._approve_fake_tool(config, fake_read_tool("fake_privacy_read"), tool_input) + + grants = ToolGrantStore(config.approvals_db_path) + self.assertTrue(grants.has("fake_privacy_read", session, tool_input)) + self.assertFalse(grants.has("fake_privacy_read", session, {"reason": "something else"})) + + def test_approving_a_failed_execution_records_no_grant(self) -> None: + from stram.safety.grants import ToolGrantStore, start_session + + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + config = AgentConfig( + workspace=workspace, + data_dir=workspace / "artifacts", + planner_provider="explicit", + dry_run=True, + ).normalized() + + session = start_session(config) + tool_input = {"reason": "read the copied link"} + self._approve_fake_tool(config, fake_read_tool("fake_privacy_read"), tool_input) + + approval = ApprovalStore(config.approvals_db_path).get("token-fake_privacy_read") + self.assertEqual(approval.result["status"], ActionStatus.SKIPPED.value) + self.assertFalse(ToolGrantStore(config.approvals_db_path).has("fake_privacy_read", session, tool_input)) + + def test_approving_provider_backed_tool_records_no_grant(self) -> None: + from stram.safety.grants import ToolGrantStore, start_session + + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + config = AgentConfig(workspace=workspace, data_dir=workspace / "artifacts", planner_provider="explicit").normalized() + + session = start_session(config) + tool_input = {"reason": "list my issues"} + tool = fake_read_tool("fake_github_read", provider_id="github") + self._approve_fake_tool(config, tool, tool_input) + + approval = ApprovalStore(config.approvals_db_path).get("token-fake_github_read") + self.assertEqual(approval.result["status"], ActionStatus.SUCCEEDED.value) + self.assertFalse(ToolGrantStore(config.approvals_db_path).has("fake_github_read", session, tool_input)) + + def test_approving_non_read_only_tool_records_no_grant(self) -> None: + from stram.safety.grants import ToolGrantStore, start_session + from stram.schemas import ApprovalRequest + + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + config = AgentConfig(workspace=workspace, data_dir=workspace / "artifacts", planner_provider="explicit").normalized() + + session = start_session(config) + store = ApprovalStore(config.approvals_db_path) + store.create_pending( + "run-noshell", + "run a shell command", + ApprovalRequest( + tool_name="run_shell_command", + tool_input={"argv": ["python", "--version"]}, + risk_level=RiskLevel.HIGH, + reason="shell", + approval_token="token-noshell", + ), + ) + + approve_pending_action(config, "token-noshell", "approved in test") + + self.assertFalse( + ToolGrantStore(config.approvals_db_path).has( + "run_shell_command", session, {"argv": ["python", "--version"]} + ) + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_policy.py b/tests/test_policy.py index 3b6cf12..7492d9f 100644 --- a/tests/test_policy.py +++ b/tests/test_policy.py @@ -1,5 +1,13 @@ +from __future__ import annotations + +import subprocess +import sys +import tempfile import unittest +from pathlib import Path +from stram.config import AgentConfig +from stram.safety.grants import ToolGrantStore, start_session from stram.safety.policy import PolicyEngine from stram.schemas import RiskLevel from stram.tools.base import Tool @@ -11,6 +19,13 @@ def execute(self, tool_input, config): class PolicyTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.config = AgentConfig(workspace=Path(self._tmp.name), data_dir=Path("artifacts")).normalized() + + def tearDown(self) -> None: + self._tmp.cleanup() + def test_low_risk_tool_is_allowed(self) -> None: tool = DummyTool("dummy", "test", RiskLevel.LOW) decision = PolicyEngine().evaluate(tool) @@ -29,6 +44,118 @@ def test_blocked_tool_is_never_allowed(self) -> None: self.assertFalse(decision.allowed) self.assertFalse(decision.requires_approval) + def test_blocked_read_only_tool_is_still_blocked(self) -> None: + tool = DummyTool("dummy", "test", RiskLevel.BLOCKED, read_only=True, provider_id="github") + engine = PolicyEngine(self.config, connected_lookup=lambda provider_id: True) + decision = engine.evaluate(tool, approved=True) + self.assertFalse(decision.allowed) + + def test_read_only_tool_on_connected_provider_skips_approval(self) -> None: + tool = DummyTool("gh_read", "test", RiskLevel.HIGH, read_only=True, provider_id="github") + engine = PolicyEngine(self.config, connected_lookup=lambda provider_id: True) + decision = engine.evaluate(tool) + self.assertTrue(decision.allowed) + self.assertFalse(decision.requires_approval) + + def test_read_only_tool_on_disconnected_provider_requires_approval(self) -> None: + tool = DummyTool("gh_read", "test", RiskLevel.HIGH, read_only=True, provider_id="github") + engine = PolicyEngine(self.config, connected_lookup=lambda provider_id: False) + decision = engine.evaluate(tool) + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_connector_lookup_failure_fails_closed(self) -> None: + def explode(provider_id: str) -> bool: + raise RuntimeError("connector store unavailable") + + tool = DummyTool("gh_read", "test", RiskLevel.HIGH, read_only=True, provider_id="github") + engine = PolicyEngine(self.config, connected_lookup=explode) + decision = engine.evaluate(tool) + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_unknown_provider_fails_closed_against_real_lookup(self) -> None: + tool = DummyTool("gh_read", "test", RiskLevel.HIGH, read_only=True, provider_id="nope_not_a_provider") + decision = PolicyEngine(self.config).evaluate(tool) + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_privacy_read_requires_approval_then_honours_grant(self) -> None: + tool = DummyTool("os_clipboard_read", "test", RiskLevel.HIGH, requires_approval=True, read_only=True) + session = start_session(self.config) + tool_input = {"reason": "check the copied link"} + + first = PolicyEngine(self.config).evaluate(tool, tool_input=tool_input) + self.assertFalse(first.allowed) + self.assertTrue(first.requires_approval) + + ToolGrantStore(self.config.approvals_db_path).record("os_clipboard_read", session, tool_input) + + second = PolicyEngine(self.config).evaluate(tool, tool_input=tool_input) + self.assertTrue(second.allowed) + self.assertFalse(second.requires_approval) + + def test_grant_does_not_cover_a_broader_call(self) -> None: + tool = DummyTool("os_observe_ui", "test", RiskLevel.HIGH, requires_approval=True, read_only=True) + session = start_session(self.config) + ToolGrantStore(self.config.approvals_db_path).record("os_observe_ui", session, {"include_values": False}) + + decision = PolicyEngine(self.config).evaluate(tool, tool_input={"include_values": True}) + + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_grant_does_not_cover_a_call_that_omits_an_approved_argument(self) -> None: + tool = DummyTool("screenpipe_search", "test", RiskLevel.HIGH, requires_approval=True, read_only=True) + session = start_session(self.config) + ToolGrantStore(self.config.approvals_db_path).record( + "screenpipe_search", session, {"query": "invoice", "content_type": "ocr", "limit": 5} + ) + + # dropping content_type/limit resolves to "all" over the whole history + decision = PolicyEngine(self.config).evaluate(tool, tool_input={"query": "invoice"}) + + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_grant_is_ignored_after_the_owning_server_dies(self) -> None: + tool = DummyTool("screenshot_capture", "test", RiskLevel.HIGH, requires_approval=True, read_only=True) + session = start_session(self.config) + tool_input = {"reason": "read the error dialog"} + ToolGrantStore(self.config.approvals_db_path).record("screenshot_capture", session, tool_input) + self.assertTrue(PolicyEngine(self.config).evaluate(tool, tool_input=tool_input).allowed) + + dead = subprocess.Popen([sys.executable, "-c", "pass"]) + dead.wait() + (self.config.data_dir / "session_pid").write_text(str(dead.pid), encoding="utf-8") + + decision = PolicyEngine(self.config).evaluate(tool, tool_input=tool_input) + + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_grant_is_ignored_when_no_tool_input_is_known(self) -> None: + tool = DummyTool("os_clipboard_read", "test", RiskLevel.HIGH, requires_approval=True, read_only=True) + session = start_session(self.config) + ToolGrantStore(self.config.approvals_db_path).record("os_clipboard_read", session, {}) + + decision = PolicyEngine(self.config).evaluate(tool) + + self.assertFalse(decision.allowed) + self.assertTrue(decision.requires_approval) + + def test_grant_from_another_session_is_ignored(self) -> None: + tool = DummyTool("os_clipboard_read", "test", RiskLevel.HIGH, requires_approval=True, read_only=True) + start_session(self.config) + ToolGrantStore(self.config.approvals_db_path).record("os_clipboard_read", "stale-session", {}) + decision = PolicyEngine(self.config).evaluate(tool, tool_input={}) + self.assertFalse(decision.allowed) + + def test_read_only_without_config_requires_approval(self) -> None: + tool = DummyTool("os_clipboard_read", "test", RiskLevel.HIGH, read_only=True) + decision = PolicyEngine().evaluate(tool) + self.assertFalse(decision.allowed) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_tool_grants.py b/tests/test_tool_grants.py new file mode 100644 index 0000000..b846a63 --- /dev/null +++ b/tests/test_tool_grants.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +from stram.config import AgentConfig +from stram.safety.grants import ToolGrantStore, clear_session, current_session_id, start_session + + +class ToolGrantTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.config = AgentConfig(workspace=Path(self._tmp.name), data_dir=Path("artifacts")).normalized() + + def tearDown(self) -> None: + self._tmp.cleanup() + + def test_no_session_file_means_no_session(self) -> None: + self.assertEqual(current_session_id(self.config), "") + + def test_start_session_writes_readable_id(self) -> None: + session = start_session(self.config) + self.assertTrue(session) + self.assertEqual(current_session_id(self.config), session) + + def test_grant_is_visible_within_session_only(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_clipboard_read", "session-a", {"reason": "check"}) + self.assertTrue(store.has("os_clipboard_read", "session-a", {"reason": "check"})) + self.assertFalse(store.has("os_clipboard_read", "session-b", {"reason": "check"})) + self.assertFalse(store.has("screenshot_capture", "session-a", {"reason": "check"})) + + def test_empty_session_never_matches(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_clipboard_read", "", {}) + self.assertFalse(store.has("os_clipboard_read", "", {})) + + def test_restart_purges_previous_session_grants(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_clipboard_read", "session-a", {}) + store.purge_other_sessions("session-b") + self.assertFalse(store.has("os_clipboard_read", "session-a", {})) + + def test_grant_covers_only_the_identical_argument_set(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_observe_ui", "session-a", {"include_values": False, "reason": "find Save"}) + + # identical arguments still work without a new prompt + self.assertTrue(store.has("os_observe_ui", "session-a", {"include_values": False, "reason": "find Save"})) + # key order is irrelevant + self.assertTrue(store.has("os_observe_ui", "session-a", {"reason": "find Save", "include_values": False})) + # dropping a key is NOT "asking for less": include_values falls back to a default + self.assertFalse(store.has("os_observe_ui", "session-a", {"reason": "find Save"})) + self.assertFalse(store.has("os_observe_ui", "session-a", {})) + # a changed value is a different request + self.assertFalse(store.has("os_observe_ui", "session-a", {"include_values": True, "reason": "find Save"})) + # an extra key is a broader request + self.assertFalse( + store.has("os_observe_ui", "session-a", {"include_values": False, "reason": "find Save", "app": "Mail"}) + ) + + def test_omitting_an_approved_key_is_not_covered(self) -> None: + """Omitted arguments resolve to broader defaults, so they must re-prompt.""" + store = ToolGrantStore(self.config.approvals_db_path) + + # content_type="all" and no time bound are far broader than what was approved + store.record( + "screenpipe_search", + "session-a", + {"query": "invoice", "content_type": "ocr", "limit": 5, "start_time": "2026-08-01T00:00:00Z"}, + ) + self.assertFalse(store.has("screenpipe_search", "session-a", {"query": "invoice"})) + self.assertFalse( + store.has("screenpipe_search", "session-a", {"query": "invoice", "content_type": "ocr"}) + ) + + # max_chars defaults to 4000 + store.record("os_clipboard_read", "session-a", {"reason": "check the copied link", "max_chars": 50}) + self.assertFalse(store.has("os_clipboard_read", "session-a", {"reason": "check the copied link"})) + + # max_elements defaults to 40 + store.record("os_observe_ui", "session-a", {"reason": "find Save", "max_elements": 5}) + self.assertFalse(store.has("os_observe_ui", "session-a", {"reason": "find Save"})) + + def test_empty_arguments_do_not_match_a_non_empty_grant(self) -> None: + """os_observe_ui has required=[], so `{}` is schema-valid and must not wildcard.""" + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_observe_ui", "session-a", {"reason": "find Save", "max_elements": 5}) + self.assertFalse(store.has("os_observe_ui", "session-a", {})) + + def test_unknown_arguments_never_match(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_observe_ui", "session-a", {}) + self.assertTrue(store.has("os_observe_ui", "session-a", {})) + # None means "we do not know what is being asked for": fail closed + self.assertFalse(store.has("os_observe_ui", "session-a", None)) + + def test_distinct_argument_sets_are_stored_side_by_side(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("screenpipe_search", "session-a", {"query": "invoice"}) + store.record("screenpipe_search", "session-a", {"query": "passwords"}) + self.assertTrue(store.has("screenpipe_search", "session-a", {"query": "invoice"})) + self.assertTrue(store.has("screenpipe_search", "session-a", {"query": "passwords"})) + self.assertFalse(store.has("screenpipe_search", "session-a", {"query": "bank"})) + + def test_clear_session_ends_session_and_drops_grants(self) -> None: + session = start_session(self.config) + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_clipboard_read", session, {}) + + clear_session(self.config) + + self.assertEqual(current_session_id(self.config), "") + self.assertFalse(store.has("os_clipboard_read", session, {})) + clear_session(self.config) # tolerates a missing session file + + def test_legacy_table_without_tool_input_is_replaced(self) -> None: + import sqlite3 + + path = self.config.approvals_db_path + path.parent.mkdir(parents=True, exist_ok=True) + with sqlite3.connect(path) as connection: + connection.execute( + "CREATE TABLE tool_grants (tool_name TEXT NOT NULL, session_id TEXT NOT NULL, " + "granted_at TEXT NOT NULL, PRIMARY KEY (tool_name, session_id))" + ) + connection.execute("INSERT INTO tool_grants VALUES ('os_clipboard_read', 'session-a', 'then')") + + store = ToolGrantStore(path) + + self.assertFalse(store.has("os_clipboard_read", "session-a", {})) + store.record("os_clipboard_read", "session-a", {"reason": "ok"}) + self.assertTrue(store.has("os_clipboard_read", "session-a", {"reason": "ok"})) + + def test_invalid_utf8_in_session_file_returns_empty_string(self) -> None: + start_session(self.config) # live owner pid, so only the id file is at fault + (self.config.data_dir / "session_id").write_bytes(b"\xff\xfe\x00bad") + self.assertEqual(current_session_id(self.config), "") + + def test_session_does_not_outlive_its_owning_process(self) -> None: + session = start_session(self.config) + self.assertEqual(current_session_id(self.config), session) + + dead = subprocess.Popen([sys.executable, "-c", "pass"]) + dead.wait() + (self.config.data_dir / "session_pid").write_text(str(dead.pid), encoding="utf-8") + + # the session id file survives an uncatchable kill; the session must not + self.assertEqual((self.config.data_dir / "session_id").read_text(encoding="utf-8").strip(), session) + self.assertEqual(current_session_id(self.config), "") + + def test_missing_or_unparseable_pid_file_fails_closed(self) -> None: + start_session(self.config) + pid_path = self.config.data_dir / "session_pid" + + for bad in ("", "not-a-pid", "0", "-1", "99999999999999999999"): + pid_path.write_text(bad, encoding="utf-8") + self.assertEqual(current_session_id(self.config), "", bad) + + pid_path.unlink() + self.assertEqual(current_session_id(self.config), "") + + def test_none_input_records_nothing_and_is_not_covered_by_empty_call(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_observe_ui", "session-a", None) + self.assertFalse(store.has("os_observe_ui", "session-a", {})) + self.assertFalse(store.has("os_observe_ui", "session-a", None)) + + def test_grant_expires_after_the_ttl(self) -> None: + import sqlite3 + from datetime import datetime, timedelta, timezone + + from stram.safety.grants import GRANT_TTL_SECONDS + + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_observe_ui", "session-a", {"reason": "find Save"}) + self.assertTrue(store.has("os_observe_ui", "session-a", {"reason": "find Save"})) + + stale = (datetime.now(timezone.utc) - timedelta(seconds=GRANT_TTL_SECONDS + 1)).isoformat() + with sqlite3.connect(self.config.approvals_db_path) as connection: + connection.execute("UPDATE tool_grants SET granted_at = ?", (stale,)) + self.assertFalse(store.has("os_observe_ui", "session-a", {"reason": "find Save"})) + + def test_unparseable_granted_at_fails_closed(self) -> None: + import sqlite3 + + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_observe_ui", "session-a", {}) + with sqlite3.connect(self.config.approvals_db_path) as connection: + connection.execute("UPDATE tool_grants SET granted_at = 'whenever'") + self.assertFalse(store.has("os_observe_ui", "session-a", {})) + + def test_pid_alive_reports_self_alive_and_reaped_pid_dead(self) -> None: + import os + + from stram.process import pid_alive + + self.assertTrue(pid_alive(os.getpid())) + + dead = subprocess.Popen([sys.executable, "-c", "pass"]) + dead.wait() + self.assertFalse(pid_alive(dead.pid)) + self.assertFalse(pid_alive(0)) + self.assertFalse(pid_alive(-1)) + self.assertFalse(pid_alive(2**31)) + + def test_pid_alive_never_signals_on_windows(self) -> None: + """os.kill(pid, 0) TERMINATES the target on Windows; it must never be reached.""" + import os + from unittest.mock import patch + + from stram.process import pid_alive + + with patch("stram.process.os.name", "nt"), patch("stram.process.os.kill") as kill: + self.assertFalse(pid_alive(os.getpid() + 1)) + kill.assert_not_called() + + def test_purge_other_sessions_with_empty_id_does_not_wipe_table(self) -> None: + store = ToolGrantStore(self.config.approvals_db_path) + store.record("os_clipboard_read", "session-a", {}) + store.purge_other_sessions("") + self.assertTrue(store.has("os_clipboard_read", "session-a", {})) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tools.py b/tests/test_tools.py index 7412e28..2ba40df 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -1905,6 +1905,108 @@ def fail_client(_config): self.assertIn("no semantic fallback", synced.summary) self.assertIn("does not guess", synced.output["safety_note"]) + def test_tool_defaults_are_not_read_only(self) -> None: + from stram.tools.base import Tool + from stram.schemas import RiskLevel + + class Probe(Tool): + def execute(self, tool_input, config): + raise NotImplementedError + + tool = Probe("probe", "test", RiskLevel.LOW) + self.assertFalse(tool.read_only) + self.assertIsNone(tool.provider_id) + + def test_alias_preserves_read_only_metadata(self) -> None: + from stram.tools import _ToolAlias + from stram.tools.base import Tool + from stram.schemas import RiskLevel + + class Probe(Tool): + def execute(self, tool_input, config): + raise NotImplementedError + + target = Probe("probe", "test", RiskLevel.LOW, read_only=True, provider_id="github") + alias = _ToolAlias("probe_alias", target) + self.assertTrue(alias.read_only) + self.assertEqual(alias.provider_id, "github") + + def test_github_read_tools_are_read_only_and_provider_scoped(self) -> None: + from stram.schemas import RiskLevel + from stram.tools.github import default_github_tools + + tools = default_github_tools() + for name in ("github_repos_list", "github_issues_list", "github_pulls_list", "github_checks_list"): + tool = tools[name] + self.assertTrue(tool.read_only, name) + self.assertEqual(tool.provider_id, "github", name) + # MEDIUM + requires_approval: the policy engine's connected-provider rule is + # what waives the prompt, so a disconnected GitHub asks instead of erroring. + self.assertEqual(tool.risk_level, RiskLevel.MEDIUM, name) + self.assertTrue(tool.requires_approval, name) + + def test_connected_github_waives_the_prompt_and_disconnected_asks(self) -> None: + import tempfile + from pathlib import Path + from stram.config import AgentConfig + from stram.safety.policy import PolicyEngine + from stram.tools.github import default_github_tools + + tool = default_github_tools()["github_issues_list"] + with tempfile.TemporaryDirectory() as tmp: + config = AgentConfig(workspace=Path(tmp), data_dir=Path("artifacts")).normalized() + + connected = PolicyEngine(config, connected_lookup=lambda _provider: True).evaluate( + tool, tool_input={"repo": "o/n"} + ) + self.assertTrue(connected.allowed) + self.assertFalse(connected.requires_approval) + + disconnected = PolicyEngine(config, connected_lookup=lambda _provider: False).evaluate( + tool, tool_input={"repo": "o/n"} + ) + self.assertFalse(disconnected.allowed) + self.assertTrue(disconnected.requires_approval) + + def test_github_read_tool_fails_clearly_when_not_connected(self) -> None: + import tempfile + from pathlib import Path + from stram.config import AgentConfig + from stram.schemas import ActionStatus + from stram.tools.github import default_github_tools + + with tempfile.TemporaryDirectory() as tmp: + config = AgentConfig(workspace=Path(tmp), data_dir=Path("artifacts")).normalized() + result = default_github_tools()["github_repos_list"].execute({}, config) + self.assertEqual(result.status, ActionStatus.FAILED) + self.assertIn("not connected", (result.error or "").lower()) + # The ValueError must be handled by the specific clause, which surfaces the + # cause in the summary. The generic handler would summarise "... failed." + self.assertIn("not connected", result.summary.lower()) + self.assertNotIn("github_repos_list failed", result.summary) + + def test_github_read_tool_reports_missing_scopes(self) -> None: + import tempfile + from pathlib import Path + from unittest.mock import patch + from stram.config import AgentConfig + from stram.connectors.models import ConnectorTokenStatus + from stram.connectors.oauth import ConnectorOAuthService + from stram.schemas import ActionStatus + from stram.tools.github import default_github_tools + + connected_without_repo = ConnectorTokenStatus(provider_id="github", connected=True, scopes=("read:org",)) + with tempfile.TemporaryDirectory() as tmp: + config = AgentConfig(workspace=Path(tmp), data_dir=Path("artifacts")).normalized() + with patch.object(ConnectorOAuthService, "token_status", return_value=connected_without_repo): + result = default_github_tools()["github_checks_list"].execute({"repo": "o/n", "ref": "main"}, config) + self.assertEqual(result.status, ActionStatus.FAILED) + self.assertIn("missing scopes", (result.error or "").lower()) + self.assertIn("repo", (result.error or "")) + # PermissionError must also be handled by the specific clause. + self.assertIn("missing scopes", result.summary.lower()) + self.assertNotIn("github_checks_list failed", result.summary) + if __name__ == "__main__": unittest.main()