diff --git a/GraphcodeKit/Sources/CLI/GraphcodeCommand.swift b/GraphcodeKit/Sources/CLI/GraphcodeCommand.swift index 62fc0065..9b611189 100644 --- a/GraphcodeKit/Sources/CLI/GraphcodeCommand.swift +++ b/GraphcodeKit/Sources/CLI/GraphcodeCommand.swift @@ -86,6 +86,11 @@ public enum GraphcodeCommand: Equatable, Sendable { the app's pinned "Graph" row — which every other verb accepts wherever appears. + A loop created with --into lives inside its composite's sub-graph, but its id is + still unique across the whole tree: node stop/delete/send/update/memo/refine and + edge create accept it as-is, paired with the project path of the graph the + composite belongs to. + RECOVERY AND SAFETY Use `graphcode projects` to discover project paths and `status` to inspect state before retrying a command. `GRAPHCODE_SUPPORT_DIR` selects the workspace for diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index f3870785..c13cb5a4 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -66,6 +66,11 @@ public actor GraphStore { private let onRefinePlaybook: (@Sendable (UUID, String) -> Bool)? /// Restores the previous playbook, consuming a snapshot (`NodeMemory.rollbackPlaybook`). private let onRollbackPlaybook: (@Sendable (UUID) -> Bool)? + /// Receives an error raised in a sub-graph store — `runInSubGraph` hands the child + /// a sink it drains and re-announces on the parent, whose connections are the ones + /// clients actually listen on. A child owns none of its own, so without this every + /// refusal inside a composite was said to nobody. + private let onAnnounceError: (@Sendable (String) -> Void)? /// Whether the daemon-heartbeat experiment is on, read fresh at every gate — creation, /// and every tick — so flipping the Settings toggle applies immediately. `nil` (tests /// that don't care, and any client that never wires it) means off, which is the @@ -110,6 +115,12 @@ public actor GraphStore { /// (b) nesting beyond `maxSubGraphDepth` is refused outright, so a runaway agent /// can't stack composites forever. private let subGraphDepth: Int + /// Where this store hands poller/heartbeat arm-and-cancel requests when it is too + /// ephemeral to own them — every sub-graph store, which is built per command and + /// whose timers would die with it. `nil` at the project root, which owns recurrence + /// for its own loops directly and for sub-graph loops via the descent in + /// `evaluateGoalDescending`/`deliverHeartbeatDescending`. + private let recurrence: RecurrenceSink? static let maxSubGraphDepth = 6 static let maxNodesPerGraph = 50 private var goalPollers: [UUID: Task] = [:] @@ -119,19 +130,14 @@ public actor GraphStore { /// the experiment on mid-run starts existing heartbeat loops beating without anyone /// re-arming anything. private var heartbeatTimers: [UUID: Task] = [:] - /// Workspace fingerprint at the last *failing* predicate run, per node — what - /// `GoalSpec.skipsUnchangedWorkspace` compares against. In-memory on purpose: a - /// daemon restart forgetting these costs one extra predicate run, and persisting a - /// cache whose whole point is skipping work would be work. - private var failedPredicateFingerprints: [UUID: String] = [:] - /// The failure tail last relayed to each node's session, so an unchanged failure is - /// never repeated at it poll after poll. - private var lastPredicateFeedback: [UUID: String] = [:] - /// The fingerprint whose unchanged tree has already bought an idle loop its one - /// re-awake (see `evaluateGoal`'s skip path). Cleared with the poller like the other - /// two caches; a new failing run at a new fingerprint naturally leaves it stale, and - /// stale here means one more notice the next freeze is allowed to spend. - private var reawakenedFingerprints: [UUID: String] = [:] + /// Workspace fingerprints at the last *failing* predicate run, the failure tail + /// last relayed to each node's session, and the fingerprint whose unchanged tree + /// has already bought an idle loop its one re-awake. In-memory on purpose: a daemon + /// restart forgetting these costs one extra predicate run, and persisting a cache + /// whose whole point is skipping work would be work. Shared with sub-graph stores + /// (which are built per command and would otherwise forget all three between + /// one-shot evaluations) via `goalCache`. + private let goalCache: GoalEvaluationCache /// `node send --follow-up` messages waiting for their target to finish its current /// turn — drained whenever the store settles (`drainAndBroadcast`) and on each /// presence poll. The content is in the target's memory log from the moment it was @@ -140,10 +146,10 @@ public actor GraphStore { private var pendingFollowUps: [(nodeID: UUID, text: String)] = [] /// A poller holds `self` weakly, so a store going away already stops it *doing* - /// anything — but the task itself keeps sleeping in its loop forever. That was - /// harmless while every store outlived the process; `runInSubGraph` builds one per - /// command and throws it away, so without this a goal loop inside a composite leaks a - /// sleeping task every time anything addresses that sub-graph. + /// anything — but the task itself keeps sleeping in its loop forever. Harmless for + /// the long-lived project store; sub-graph stores are built per command and hold no + /// timers at all (recurrence for their loops is forwarded up), so this deinit is a + /// backstop rather than a leak fix. deinit { for poller in goalPollers.values { poller.cancel() } for timer in heartbeatTimers.values { timer.cancel() } @@ -197,11 +203,14 @@ public actor GraphStore { onRemoveMemory: (@Sendable (UUID) -> Void)? = nil, onRefinePlaybook: (@Sendable (UUID, String) -> Bool)? = nil, onRollbackPlaybook: (@Sendable (UUID) -> Bool)? = nil, + onAnnounceError: (@Sendable (String) -> Void)? = nil, onHeartbeatEnabled: (@Sendable () -> Bool)? = nil, onComposeBoard: ( @Sendable (LoopNode, LoopSummary, String?, String?) async -> SummaryBoard? )? = nil, onBoardsEnabled: (@Sendable () -> Bool)? = nil, + goalCache: GoalEvaluationCache? = nil, + recurrence: RecurrenceSink? = nil, subGraphDepth: Int = 0 ) { self.graph = graph @@ -222,9 +231,12 @@ public actor GraphStore { self.onRemoveMemory = onRemoveMemory self.onRefinePlaybook = onRefinePlaybook self.onRollbackPlaybook = onRollbackPlaybook + self.onAnnounceError = onAnnounceError self.onHeartbeatEnabled = onHeartbeatEnabled self.onComposeBoard = onComposeBoard self.onBoardsEnabled = onBoardsEnabled + self.goalCache = goalCache ?? GoalEvaluationCache() + self.recurrence = recurrence } private func recordMemory(_ nodeID: UUID, _ entry: String) { @@ -271,6 +283,13 @@ public actor GraphStore { // MARK: - Commands public func handle(_ command: GraphCommand) async { + // A loop inside a composite addresses itself by its own id — its briefing tells it + // to `node memo `, and ids are unique across the whole tree, + // so a caller has no reason to know how deep its target sits (the same rule + // `runInSubGraph` already honours for already-wrapped commands). A command whose + // target names no top-level loop but lives inside a sub-graph is wrapped for the + // composite that holds it rather than refused by a lookup that never looked down. + let command = routeIntoSubGraph(command) ?? command switch command { case .createNode(var draft): // A child inherits its creator's backend unless one was named: a Copilot loop @@ -342,12 +361,27 @@ public actor GraphStore { armHeartbeat(for: node) case .createEdge(let from, let to, let spec): + guard from != to else { return } + // Refused out loud rather than dropped: routing has already sent pairs that + // share a sub-graph down into it, so an endpoint missing from this graph's own + // nodes is either a loop inside a composite — and no edge may span two graphs, + // not even a sub-graph and its parent — or a loop that exists nowhere. Either + // way the caller is waiting for an answer, and silence reads as a timeout, not + // a refusal. (A duplicate of the same kind still collapses quietly, as before.) + guard graph.nodes[id: from] != nil, graph.nodes[id: to] != nil else { + let missing = graph.nodes[id: from] == nil ? from : to + announceError( + graph.containsAtAnyDepth(missing) + ? "edge refused: an edge may not span two graphs — \(missing) lives inside " + + "a composite, so both of its endpoints must share that sub-graph" + : "edge refused: no loop \(missing) in this graph") + return + } // Duplicates are scoped per kind, not per pair: a `.handoff` and a `.message` // between the same two loops are different relationships (one sequences them, // one lets them talk mid-flight), so both are allowed to exist at once. Two // edges of the *same* kind between the same pair still collapse to one. - guard from != to, graph.nodes[id: from] != nil, graph.nodes[id: to] != nil, - !graph.edges.contains(where: { $0.from == from && $0.to == to && $0.kind == spec.kind }) + guard !graph.edges.contains(where: { $0.from == from && $0.to == to && $0.kind == spec.kind }) else { return } // A guard that bounds nothing would turn a cycle into an unattended infinite loop // spending tokens forever. Refused outright rather than silently dropped, so the @@ -433,6 +467,45 @@ public actor GraphStore { // MARK: - Composites + /// Wraps a command whose target loop lives inside a composite's sub-graph, for + /// dispatch through `runInSubGraph` — `nil` when the command needs no routing. + /// + /// Node commands used to resolve their target against this graph's own nodes only, + /// which locked a composite's children out of the CLI: `node memo`, `node refine`, + /// `node send`, `node delete`, `edge create` all answered "no loop in this + /// graph" for a child that plainly existed, and a piloted loop told to memo or + /// refine itself could never succeed. The owner searched for here is the *top-level* + /// composite holding the target; `runInSubGraph` and the child store's own routing + /// descend the rest of the way, one hop each, so nesting costs nothing extra here. + /// + /// A command naming a loop that exists nowhere still returns `nil`: the command's + /// own guard then refuses it with the message a caller expects. + private func routeIntoSubGraph(_ command: GraphCommand) -> GraphCommand? { + func subGraphOwner(of target: UUID) -> UUID? { + guard graph.nodes[id: target] == nil, + let owner = graph.nodes.first(where: { $0.subGraph?.containsAtAnyDepth(target) == true }) + else { return nil } + return owner.id + } + switch command { + case .createEdge(let from, let to, _): + // An edge lives in the graph holding both of its endpoints, so only a pair that + // shares one sub-graph can be routed there; anything else is refused below, as + // it always was. + guard from != to, let ownerID = subGraphOwner(of: from), subGraphOwner(of: to) == ownerID + else { return nil } + return .subGraphCommand(nodeID: ownerID, command: command) + case .nodeCheckApproved(let id), .nodeCheckRejected(let id), .renameNode(let id, _), + .updateNode(let id, _), .promoteNode(let id, _, _), .memoNode(let id, _, _), + .refineNode(let id, _, _), .rollbackRefinement(let id, _), .messageNode(let id, _, _, _), + .deleteNode(let id), .stopNode(let id): + guard let ownerID = subGraphOwner(of: id) else { return nil } + return .subGraphCommand(nodeID: ownerID, command: command) + default: + return nil + } + } + /// Runs a command against a composite node's sub-graph, then rolls the result up. /// /// The nested graph is orchestrated by a real `GraphStore` — the same type, the same @@ -465,6 +538,7 @@ public actor GraphStore { // Built fresh per command rather than cached: the sub-graph lives on the parent // node, which is the persisted source of truth, so a long-lived child store would // just be a copy that can drift from it. + let effects = SubGraphEffects() let child = GraphStore( graph: subGraph, // Deliberately *not* forwarded. A loop inside a composite is a template with no @@ -484,8 +558,18 @@ public actor GraphStore { onRemoveMemory: onRemoveMemory, onRefinePlaybook: onRefinePlaybook, onRollbackPlaybook: onRollbackPlaybook, + onAnnounceError: effects.errors.append, + goalCache: goalCache, + recurrence: effects.recurrence, subGraphDepth: subGraphDepth + 1) await child.handle(command) + // Settled before the write-back and roll-up below, so a client sees the refusal + // ahead of the broadcast it would otherwise time out against, and an update's + // re-armed poller is in place before anyone sees the graph it belongs to. + for message in effects.errors.drained { + announceError(message) + } + processRecurrence(effects.recurrence) graph.nodes[id: nodeID]?.subGraph = await child.graph rollUpComposite(nodeID) } @@ -529,6 +613,11 @@ public actor GraphStore { for child in subGraph.nodes where child.runsUnattended { ensureSession(child) } + // The pilot is also the moment the composite's loops become real, so it is the + // moment their recurrence becomes real: a goal child's stop condition and a time + // child's cadence are armed here on this store, keyed by the child's id, ticking + // into the sub-graph by descent (a per-command child store cannot hold a timer). + armRecurrence(for: subGraph.nodes) } graph.nodes[id: nodeID]?.pilotState = .piloted await refreshUsage() @@ -1301,10 +1390,15 @@ public actor GraphStore { // `graph.nodes` — the same blind spot `requestStop` covers when stopping, and the // sessions `pilotComposite` and `spawnInstance` started for them are just as real. // Killed rather than asked, unlike a stop: the nodes cease to exist with their - // parent, so there is nothing left for a polite stop request to resolve. + // parent, so there is nothing left for a polite stop request to resolve. Their + // recurrence is cancelled here too — the pollers and heartbeats live on the + // project store keyed by the workers' own ids, and a deleted loop must not keep + // being polled. for worker in node.subGraph?.nodesAtAnyDepth ?? [] { terminateSession(worker) onRemoveMemory?(worker.id) + cancelGoalPoller(worker.id) + cancelHeartbeat(worker.id) } } @@ -1547,11 +1641,14 @@ public actor GraphStore { if instance.runsUnattended { ensureSession(instance) } if instance.loopType == .goalBased { armGoalPoller(for: instance) } // A composite's work is its sub-graph's, so instantiating one has to start what's - // inside it — otherwise the spawn produces a node that merely looks busy. + // inside it — otherwise the spawn produces a node that merely looks busy. The + // instance is armed rather than awaiting a pilot, so its loops' recurrence starts + // with them. if let subGraph = instance.subGraph { for child in subGraph.nodes where child.runsUnattended { ensureSession(child) } + armRecurrence(for: subGraph.nodes) } } @@ -1966,6 +2063,7 @@ public actor GraphStore { for id in connections.keys { send(.errorOccurred(message), to: id) } + onAnnounceError?(message) } private func unblockIfStillIdle(_ nodeID: UUID) { @@ -1989,6 +2087,14 @@ public actor GraphStore { /// headlessly, leaving nothing to attach to. This one only asks an outside question /// about work that is running in a perfectly ordinary session the whole time. private func armGoalPoller(for node: LoopNode) { + // A sub-graph store is built per command; a timer armed here dies with it, so the + // request is handed up to the store that owns recurrence for this loop. The parent + // applies the pilot gate — an unpiloted composite's loops are templates, and a + // poller that resolved a template's goal would mark work done that never ran. + if subGraphDepth > 0 { + recurrence?.append(.armGoalPoller(node)) + return + } guard let goal = node.goal else { return } // Three independent reasons to poll: a predicate to evaluate, a stall bound to // enforce, or a token budget to hold the line on. A goal stated only in prose still @@ -2004,20 +2110,151 @@ public actor GraphStore { while !Task.isCancelled { try? await Task.sleep(for: .seconds(interval)) guard !Task.isCancelled else { return } - await self?.evaluateGoal(nodeID) + await self?.evaluateGoalDescending(nodeID) } } } private func cancelGoalPoller(_ nodeID: UUID) { + if subGraphDepth > 0 { + recurrence?.append(.cancelGoalPoller(nodeID)) + return + } goalPollers.removeValue(forKey: nodeID)?.cancel() // The caches ride the poller's lifecycle: a resolved node needs none of them, and // an update that changed the predicate must not skip the new command on the old // tree's fingerprint, suppress its first failure as "already relayed", or count // the old tree's re-awake as spent. - failedPredicateFingerprints.removeValue(forKey: nodeID) - lastPredicateFeedback.removeValue(forKey: nodeID) - reawakenedFingerprints.removeValue(forKey: nodeID) + goalCache.clear(for: nodeID) + } + + // MARK: - Recurrence for sub-graph loops + + /// One poller tick, wherever the loop lives. Pollers are armed here on the project + /// store — including for loops inside composites, which per-command sub-graph stores + /// cannot hold — so the tick descends into the owning sub-graph when the id names no + /// loop of this graph's own. + private func evaluateGoalDescending(_ nodeID: UUID) async { + if graph.nodes[id: nodeID] != nil { + await evaluateGoal(nodeID) + return + } + guard let owner = graph.nodes.first(where: { $0.subGraph?.containsAtAnyDepth(nodeID) == true }), + let subGraph = owner.subGraph + else { + // The loop is gone from the tree; nothing left to tick at. + cancelGoalPoller(nodeID) + return + } + let effects = SubGraphEffects() + let child = subGraphStore(for: subGraph, effects: effects) + await child.evaluateGoalDescending(nodeID) + await settle(child: child, ownerID: owner.id, effects: effects) + } + + /// The heartbeat timer's descent — same shape, same reasoning, see + /// `evaluateGoalDescending`. + private func deliverHeartbeatDescending(_ nodeID: UUID) async { + if graph.nodes[id: nodeID] != nil { + await deliverHeartbeat(nodeID) + return + } + guard let owner = graph.nodes.first(where: { $0.subGraph?.containsAtAnyDepth(nodeID) == true }), + let subGraph = owner.subGraph + else { + cancelHeartbeat(nodeID) + return + } + let effects = SubGraphEffects() + let child = subGraphStore(for: subGraph, effects: effects) + await child.deliverHeartbeatDescending(nodeID) + await settle(child: child, ownerID: owner.id, effects: effects) + } + + /// A child store built for one tick of recurrence — the same construction + /// `runInSubGraph` uses, sharing the goal cache so a one-shot evaluation inherits the + /// fingerprints and failure tails of every evaluation before it. Without the shared + /// cache, a failing predicate would be relayed to the session afresh on every poll. + private func subGraphStore(for subGraph: LoopGraph, effects: SubGraphEffects) -> GraphStore { + GraphStore( + graph: subGraph, + onTerminateSession: onTerminateSession, + onEvaluatePredicate: onEvaluatePredicate, + onCheckPredicate: onCheckPredicate, + onDeliverMessage: onDeliverMessage, + onCaptureScript: onCaptureScript, + onReadUsage: onReadUsage, + onReadPresence: onReadPresence, + onAppendMemory: onAppendMemory, + onRemoveMemory: onRemoveMemory, + onRefinePlaybook: onRefinePlaybook, + onRollbackPlaybook: onRollbackPlaybook, + onAnnounceError: effects.errors.append, + goalCache: goalCache, + recurrence: effects.recurrence, + subGraphDepth: subGraphDepth + 1) + } + + /// Writes a tick's mutations back into the persisted tree, rolls the composite up, + /// and settles what the child handed up — errors re-announced, recurrence applied. + private func settle(child: GraphStore, ownerID: UUID, effects: SubGraphEffects) async { + for message in effects.errors.drained { + announceError(message) + } + processRecurrence(effects.recurrence) + graph.nodes[id: ownerID]?.subGraph = await child.graph + rollUpComposite(ownerID) + await drainAndBroadcast() + } + + /// Applies the recurrence requests a child store handed up, in order — an update's + /// cancel-then-rearm must land as a pair or a `--poll` change kills its own poller. + /// At depth this store is itself a per-command child, so requests keep travelling up. + private func processRecurrence(_ sink: RecurrenceSink) { + for request in sink.drained { + if subGraphDepth > 0 { + recurrence?.append(request) + continue + } + switch request { + case .armGoalPoller(let node): + guard pilotedCompositeDirectlyContains(node.id) else { continue } + armGoalPoller(for: node) + case .armHeartbeat(let node): + guard pilotedCompositeDirectlyContains(node.id) else { continue } + armHeartbeat(for: node) + case .cancelGoalPoller(let nodeID): + cancelGoalPoller(nodeID) + case .cancelHeartbeat(let nodeID): + cancelHeartbeat(nodeID) + } + } + } + + /// Whether the composite whose sub-graph *directly* holds `nodeID` has been piloted + /// or armed — the gate on recurrence handed up from a child store. A piloted outer + /// composite does not make an unpiloted inner one live: its loops have no sessions. + private func pilotedCompositeDirectlyContains(_ nodeID: UUID) -> Bool { + func search(_ nodes: some Collection) -> Bool { + for node in nodes { + guard let sub = node.subGraph else { continue } + if sub.nodes.contains(where: { $0.id == nodeID }) { + return node.pilotState == .piloted || node.pilotState == .armed + } + if search(sub.nodes) { return true } + } + return false + } + return search(graph.nodes) + } + + /// Arms recurrence for the loops a piloted or armed composite brought live — its + /// direct children only, since piloting starts sessions one level at a time. + private func armRecurrence(for children: some Collection) { + for child in children where child.runsUnattended && !child.isResolved { + if child.loopType == .goalBased { armGoalPoller(for: child) } + if child.loopType == .timeBased { armHeartbeat(for: child) } + } } /// One poll. Called on the timer in production and directly from tests, so the @@ -2078,7 +2315,7 @@ public actor GraphStore { // before its turn left the tree unmoved. After that the skip holds again until // the tree moves: re-delivering every poll would be a full agent turn a minute, // the unbounded spend the failure-tail dedup exists to prevent. - if let fingerprint, failedPredicateFingerprints[nodeID] == fingerprint { + if let fingerprint, goalCache.fingerprint(for: nodeID) == fingerprint { let presence: Presence? if let onReadPresence { presence = await onReadPresence(node, graph.project.path).presence @@ -2089,9 +2326,9 @@ public actor GraphStore { // idle, so falling through would pay the predicate's price for a wake that can // never land. Such a loop's exits are its stall bound and its human. guard presence == .idle else { return } - guard reawakenedFingerprints[nodeID] != fingerprint else { return } - reawakenedFingerprints[nodeID] = fingerprint - lastPredicateFeedback.removeValue(forKey: nodeID) + guard goalCache.reawakened(for: nodeID) != fingerprint else { return } + goalCache.setReawakened(fingerprint, for: nodeID) + goalCache.clearFeedback(for: nodeID) } } @@ -2112,7 +2349,7 @@ public actor GraphStore { await drainAndBroadcast() return } - if let fingerprint { failedPredicateFingerprints[nodeID] = fingerprint } + if let fingerprint { goalCache.setFingerprint(fingerprint, for: nodeID) } await relayPredicateFailure(to: current, predicate: predicate, outcome: outcome) } @@ -2171,7 +2408,7 @@ public actor GraphStore { to node: LoopNode, predicate: String, outcome: PredicateOutcome ) async { let tail = outcome.outputTail.trimmingCharacters(in: .whitespacesAndNewlines) - guard !tail.isEmpty, lastPredicateFeedback[node.id] != tail else { return } + guard !tail.isEmpty, goalCache.feedback(for: node.id) != tail else { return } let presence: Presence? if let onReadPresence { presence = await onReadPresence(node, graph.project.path).presence @@ -2183,7 +2420,7 @@ public actor GraphStore { "[graphcode] Goal not met yet: `\(predicate)` still exits non-zero. " + "Its output ends with: \(tail)" guard await deliverToSession(node, message) else { return } - lastPredicateFeedback[node.id] = tail + goalCache.setFeedback(tail, for: node.id) recordMemory(node.id, "predicate feedback: \(tail)") } @@ -2229,6 +2466,12 @@ public actor GraphStore { /// the session on every beat, so the timer itself holds no authority anything else /// would need revoking. private func armHeartbeat(for node: LoopNode) { + // Forwarded up for the same reason the goal poller is: a per-command store cannot + // own a timer. The parent applies the same pilot gate on receipt. + if subGraphDepth > 0 { + recurrence?.append(.armHeartbeat(node)) + return + } guard node.loopType == .timeBased, let interval = node.effectiveHeartbeatInterval, interval > 0, onDeliverMessage != nil else { return } @@ -2239,12 +2482,16 @@ public actor GraphStore { while !Task.isCancelled { try? await Task.sleep(for: .seconds(beat)) guard !Task.isCancelled else { return } - await self?.deliverHeartbeat(nodeID) + await self?.deliverHeartbeatDescending(nodeID) } } } private func cancelHeartbeat(_ nodeID: UUID) { + if subGraphDepth > 0 { + recurrence?.append(.cancelHeartbeat(nodeID)) + return + } heartbeatTimers.removeValue(forKey: nodeID)?.cancel() } @@ -2303,6 +2550,23 @@ public actor GraphStore { if !node.isResolved { armHeartbeat(for: node) } ensureSession(node) } + armPilotedSubGraphRecurrence(graph.nodes) + } + + /// The boot-time half of the pilot's arming. Pollers and heartbeats are in-memory, + /// so a daemon restart drops every piloted composite's recurrence along with the + /// top-level loops'; this re-arms it for the loops whose composite is still piloted + /// or armed. Sessions are not re-ensued here beyond what the loop above already did + /// — child sessions reattach to their `zmx` names, and the liveness sweep is the + /// place that restarts the ones it cannot reach. + private func armPilotedSubGraphRecurrence(_ nodes: some Collection) { + for node in nodes { + guard let sub = node.subGraph else { continue } + if node.pilotState == .piloted || node.pilotState == .armed { + armRecurrence(for: sub.nodes) + } + armPilotedSubGraphRecurrence(sub.nodes) + } } /// The session half of `ensureUnattendedSessions`, for the repeating remote liveness @@ -2356,4 +2620,144 @@ public actor GraphStore { return } } + + /// Predicate-evaluation state shared between a project store and every sub-graph + /// store it builds: which workspace fingerprint each node's predicate last failed + /// against, which failure tail each session was last told, and which frozen tree + /// has already spent its one idle re-awake. The project store owns the box for the + /// life of the graph; per-command sub-graph stores borrow it so a one-shot + /// evaluation inherits what every evaluation before it learned — without that, a + /// failing predicate would be relayed to the session afresh on every poll. Public + /// only because `GraphStore.init` takes it; there is nothing to call. + public final class GoalEvaluationCache: @unchecked Sendable { + private let lock = NSLock() + private var fingerprints: [UUID: String] = [:] + private var feedback: [UUID: String] = [:] + private var reawakened: [UUID: String] = [:] + + func fingerprint(for nodeID: UUID) -> String? { + lock.lock() + defer { lock.unlock() } + return fingerprints[nodeID] + } + + func setFingerprint(_ value: String, for nodeID: UUID) { + lock.lock() + defer { lock.unlock() } + fingerprints[nodeID] = value + } + + func feedback(for nodeID: UUID) -> String? { + lock.lock() + defer { lock.unlock() } + return feedback[nodeID] + } + + func setFeedback(_ value: String, for nodeID: UUID) { + lock.lock() + defer { lock.unlock() } + feedback[nodeID] = value + } + + func reawakened(for nodeID: UUID) -> String? { + lock.lock() + defer { lock.unlock() } + return reawakened[nodeID] + } + + func setReawakened(_ value: String, for nodeID: UUID) { + lock.lock() + defer { lock.unlock() } + reawakened[nodeID] = value + } + + /// Forgets only the last-relayed tail — the idle re-awake uses it to let a + /// failure that reads the same be told once more. The fingerprint and the + /// re-awake marker stay: the skip must keep holding around this one delivery. + func clearFeedback(for nodeID: UUID) { + lock.lock() + defer { lock.unlock() } + feedback[nodeID] = nil + } + + /// A node's poller ended — resolved, updated, stopped, or deleted. Its next + /// predicate run starts the caches fresh, and its next wake may hear the failure + /// again even if it was told before. + func clear(for nodeID: UUID) { + lock.lock() + defer { lock.unlock() } + fingerprints[nodeID] = nil + feedback[nodeID] = nil + reawakened[nodeID] = nil + } + } + + /// What one pass through a sub-graph store hands back to the store that ran it. + /// Both channels are buffered rather than forwarded inline: the child writes from its + /// own isolation, and the parent settles both — errors first, then recurrence — + /// before its `graphChanged` broadcast, which is the order a one-shot CLI client + /// (waiting for whichever event arrives first) needs to see. + private final class SubGraphEffects: @unchecked Sendable { + let errors = SubGraphErrorSink() + let recurrence = RecurrenceSink() + } + + /// Errors a sub-graph store raises while handling one command, held until the parent + /// can re-announce them on its own connections. Written from the child's isolation, + /// read from the parent's — hence the lock. A child owns no connections of its own, + /// so without this hop its refusals were said to nobody. + private final class SubGraphErrorSink: @unchecked Sendable { + private let lock = NSLock() + private var messages: [String] = [] + + func append(_ message: String) { + lock.lock() + defer { lock.unlock() } + messages.append(message) + } + + var drained: [String] { + lock.lock() + defer { lock.unlock() } + let taken = messages + messages = [] + return taken + } + } + + /// A poller or heartbeat a sub-graph store was asked to arm or cancel. Sub-graph + /// stores are built per command and hold no timers — a timer armed there would die + /// with the store, leaving a `--poll` change or a new goal loop silently inert — so + /// the request travels up to the project store, which owns recurrence for the whole + /// tree and ticks into sub-graphs by descent. Public only because `GraphStore.init` + /// takes the sink. + public enum RecurrenceRequest: Sendable { + case armGoalPoller(LoopNode) + case armHeartbeat(LoopNode) + case cancelGoalPoller(UUID) + case cancelHeartbeat(UUID) + } + + /// Where those requests queue while the child handles its command. Written from the + /// child's isolation, drained in order by the parent — the order matters, because an + /// update re-arms by cancelling and then arming. Public only because + /// `GraphStore.init` takes it; there is nothing to call from outside. + public final class RecurrenceSink: @unchecked Sendable { + private let lock = NSLock() + private var requests: [RecurrenceRequest] = [] + + func append(_ request: RecurrenceRequest) { + lock.lock() + defer { lock.unlock() } + requests.append(request) + } + + var drained: [RecurrenceRequest] { + lock.lock() + defer { lock.unlock() } + let taken = requests + requests = [] + return taken + } + } } diff --git a/graphcode/Tests/SubGraphAddressingTests.swift b/graphcode/Tests/SubGraphAddressingTests.swift new file mode 100644 index 00000000..cb1e0ec6 --- /dev/null +++ b/graphcode/Tests/SubGraphAddressingTests.swift @@ -0,0 +1,317 @@ +import ComposableArchitecture +import Foundation +import IdentifiedCollections +import Testing + +@testable import GraphcodeKit + +/// Addressing a sub-graph child by its own id — the fix for issue #217 item 15. +/// +/// A piloted child is briefed to `node memo ` — ids are unique +/// across the whole tree, so a caller has no reason to know how deep its target sits. +/// Resolving that id against the top-level nodes only answered "no loop in this +/// graph" for a loop that plainly existed, which locked composite children out of +/// memo, refine, send, delete, and edges from the CLI entirely. +@Suite +struct SubGraphAddressingTests { + private func storeWithComposite( + subNodes: [LoopNode] = [], + onCheckPredicate: (@Sendable (ShellPredicate) async -> PredicateOutcome?)? = nil, + onAppendMemory: (@Sendable (UUID, String) -> Void)? = nil, + onRefinePlaybook: (@Sendable (UUID, String) -> Bool)? = nil, + onDeliverMessage: (@Sendable (LoopNode, String, String?) async -> Bool)? = nil, + onAnnounceError: (@Sendable (String) -> Void)? = nil + ) -> (store: GraphStore, compositeID: UUID) { + let composite = LoopNode( + title: "Triage inbox", loopType: .composite, + subGraph: LoopGraph( + project: ProjectRef(path: "sub", name: "sub"), + nodes: IdentifiedArray(uniqueElements: subNodes))) + let store = GraphStore( + graph: LoopGraph( + project: ProjectRef(path: "/tmp/p", name: "p"), nodes: [composite]), + onCheckPredicate: onCheckPredicate, + onDeliverMessage: onDeliverMessage, + onAppendMemory: onAppendMemory, + onRefinePlaybook: onRefinePlaybook, + onAnnounceError: onAnnounceError) + return (store, composite.id) + } + + @Test + func aChildLoopCanMemoItselfByItsOwnId() async { + let memories = LockIsolated<[UUID: [String]]>([:]) + let worker = LoopNode(title: "Worker", loopType: .turnBased, checkDescription: "?") + let (store, _) = storeWithComposite( + subNodes: [worker], + onAppendMemory: { id, text in + memories.withValue { $0[id, default: []].append(text) } + }) + + await store.handle(.memoNode(worker.id, text: "classified 12 items", from: nil)) + + #expect(memories.value[worker.id]?.contains("note: classified 12 items") == true) + } + + @Test + func aChildLoopCanRefineItselfByItsOwnId() async { + let refined = LockIsolated<[UUID: String]>([:]) + let worker = LoopNode(title: "Worker", loopType: .turnBased, checkDescription: "?") + let (store, _) = storeWithComposite( + subNodes: [worker], + onRefinePlaybook: { id, text in + refined.withValue { $0[id] = text } + return true + }) + + await store.handle(.refineNode(worker.id, text: "check the queue first", from: nil)) + + #expect(refined.value[worker.id] == "check the queue first") + } + + @Test + func aMessageAddressedToAChildLoopReachesItsTransport() async { + let delivered = LockIsolated<[UUID]>([]) + let worker = LoopNode(title: "Worker", loopType: .turnBased, checkDescription: "?") + let (store, _) = storeWithComposite( + subNodes: [worker], + onDeliverMessage: { node, _, _ in + delivered.withValue { $0.append(node.id) } + return true + }) + + await store.handle( + .messageNode(worker.id, text: "prioritize the inbox", from: nil, followUp: nil)) + + #expect(delivered.value == [worker.id]) + } + + @Test + func aChildLoopCanBeDeletedByItsOwnId() async { + let worker = LoopNode(title: "Worker", loopType: .turnBased, checkDescription: "?") + let (store, compositeID) = storeWithComposite(subNodes: [worker]) + + await store.handle(.deleteNode(worker.id)) + + #expect(await store.graph.nodes[id: compositeID]?.subGraph?.nodes.isEmpty == true) + } + + @Test + func anEdgeBetweenChildLoopsCanBeCreatedFromOutsideTheComposite() async { + let classify = LoopNode(title: "Classify", loopType: .turnBased, checkDescription: "?") + let draft = LoopNode(title: "Draft reply", loopType: .turnBased, checkDescription: "?") + let (store, compositeID) = storeWithComposite(subNodes: [classify, draft]) + + await store.handle(.createEdge(from: classify.id, to: draft.id, spec: EdgeSpec())) + + let sub = await store.graph.nodes[id: compositeID]?.subGraph + #expect(sub?.edges.count == 1) + // Edge semantics apply inside a composite exactly as outside it. + #expect(sub?.nodes[id: draft.id]?.state == .blocked) + } + + @Test + func aNestedChildIsReachedThroughBothLevelsByItsOwnId() async throws { + // A composite inside a composite: the deep loop's id still needs no path form — + // each level's routing descends one hop, the way `runInSubGraph` already did for + // wrapped commands. + let memories = LockIsolated<[UUID: [String]]>([:]) + let (store, outerID) = storeWithComposite( + onAppendMemory: { id, text in + memories.withValue { $0[id, default: []].append(text) } + }) + await store.handle( + .subGraphCommand( + nodeID: outerID, command: .createNode(NodeDraft(title: "Inner", loopType: .composite)))) + let innerID = try #require(await store.graph.nodes[id: outerID]?.subGraph?.nodes.first?.id) + await store.handle( + .subGraphCommand( + nodeID: innerID, + command: .createNode( + NodeDraft( + title: "Deep", loopType: .turnBased, checkDescription: "?", + firstInstruction: "Work")))) + let deepID = try #require( + await store.graph.nodes[id: outerID]?.subGraph?.nodes[id: innerID]?.subGraph?.nodes.first?.id) + + await store.handle(.memoNode(deepID, text: "made it down", from: nil)) + + #expect(memories.value[deepID]?.contains("note: made it down") == true) + } + + @Test + func aMemoForAnIdThatExistsNowhereIsStillRefusedByName() async { + // Routing must not swallow the plain failure: an id naming no loop anywhere gets + // the message the caller expects. + let errors = LockIsolated<[String]>([]) + let (store, _) = storeWithComposite( + onAnnounceError: { message in errors.withValue { $0.append(message) } }) + let missing = UUID() + + await store.handle(.memoNode(missing, text: "gone", from: nil)) + + #expect(errors.value == ["memo not recorded: no loop \(missing) in this graph"]) + } + + @Test + func aRefusalInsideASubGraphIsAnnouncedRatherThanSwallowed() async { + // The child store owns no connections, so before errors were forwarded up, a + // routed command that was refused — empty note, over-long playbook, staged + // message — was said to nobody and the CLI timed out on it. + let errors = LockIsolated<[String]>([]) + let worker = LoopNode(title: "Worker", loopType: .turnBased, checkDescription: "?") + let (store, _) = storeWithComposite( + subNodes: [worker], + onAnnounceError: { message in errors.withValue { $0.append(message) } }) + + await store.handle(.memoNode(worker.id, text: " ", from: nil)) + + #expect(errors.value == ["memo not recorded: empty note"]) + } + + @Test + func anUpdateRoutesIntoAChildLoop() async { + let worker = LoopNode(title: "Worker", loopType: .turnBased, checkDescription: "?") + let (store, compositeID) = storeWithComposite(subNodes: [worker]) + + await store.handle( + .updateNode(worker.id, update: NodeUpdate(checkDescription: "reviewed?"))) + + #expect( + await store.graph.nodes[id: compositeID]?.subGraph?.nodes[id: worker.id]? + .checkDescription == "reviewed?") + } + + @Test + func aChildLoopCanBeStoppedByItsOwnId() async { + let worker = LoopNode(title: "Worker", loopType: .turnBased, checkDescription: "?") + let (store, compositeID) = storeWithComposite(subNodes: [worker]) + + await store.handle(.stopNode(worker.id)) + + #expect( + await store.graph.nodes[id: compositeID]?.subGraph?.nodes[id: worker.id]?.state + == .stopped) + } + + @Test + func aChildLoopCanBePromotedByItsOwnId() async { + let seedling = LoopNode(title: "Seedling", loopType: .sketch) + let (store, compositeID) = storeWithComposite(subNodes: [seedling]) + + await store.handle( + .promoteNode( + seedling.id, + promotion: .goal(GoalSpec(summary: "done means the changelog is written")), + promotedBy: nil)) + + let promoted = await store.graph.nodes[id: compositeID]?.subGraph?.nodes[id: seedling.id] + #expect(promoted?.loopType == .goalBased) + #expect(promoted?.state == .running) + } + + @Test + func anEdgeFromATopLevelLoopToAChildLoopIsRefusedAsASpan() async throws { + // No edge may span two graphs. Refused out loud — the caller is waiting for an + // answer, and silence reads as a timeout, not a refusal. + let worker = LoopNode(title: "Worker", loopType: .turnBased, checkDescription: "?") + let errors = LockIsolated<[String]>([]) + let (store, _) = storeWithComposite( + subNodes: [worker], + onAnnounceError: { message in errors.withValue { $0.append(message) } }) + await store.handle(.createNode(NodeDraft(title: "Outside", loopType: .turnBased))) + let outsideID = try #require(await store.graph.nodes.last?.id) + + await store.handle(.createEdge(from: outsideID, to: worker.id, spec: EdgeSpec())) + + #expect(errors.value.count == 1) + #expect(errors.value.first?.hasPrefix("edge refused: an edge may not span two graphs") == true) + } + + @Test + func anEdgeToAnUnknownLoopIsRefusedByName() async throws { + let outside = LoopNode(title: "Outside", loopType: .turnBased, checkDescription: "?") + let errors = LockIsolated<[String]>([]) + let (store, _) = storeWithComposite( + onAnnounceError: { message in errors.withValue { $0.append(message) } }) + await store.handle(.createNode(NodeDraft(title: "Outside", loopType: .turnBased))) + let outsideID = try #require(await store.graph.nodes.last?.id) + let missing = UUID() + + await store.handle(.createEdge(from: outsideID, to: missing, spec: EdgeSpec())) + + #expect(errors.value == ["edge refused: no loop \(missing) in this graph"]) + } + + // MARK: - Recurrence for sub-graph loops + + // A per-command child store cannot hold a timer — one armed there would die with the + // store, which is how a `--poll` change on a child was silently inert. Recurrence is + // now armed on the project store and ticks into the sub-graph by descent. + + @Test + func aPilotedChildsGoalIsPolledFromTheProjectStore() async throws { + let polls = LockIsolated(0) + let worker = LoopNode( + title: "Worker", loopType: .goalBased, + goal: GoalSpec(summary: "ship it", predicate: "true", pollIntervalSeconds: 1)) + let (store, compositeID) = storeWithComposite( + subNodes: [worker], + onCheckPredicate: { _ in + polls.withValue { $0 += 1 } + return PredicateOutcome(passed: false) + }) + + await store.handle(.pilotComposite(compositeID)) + #expect(await store.graph.nodes[id: compositeID]?.pilotState == .piloted) + try await Task.sleep(for: .seconds(1.6)) + + #expect(polls.value >= 1) + } + + @Test + func anUpdateToAChildsPollIntervalReachesTheProjectStoresPoller() async throws { + // The regression: the re-arm used to land in the ephemeral child store and die + // with it, so a `--poll` change was silently ignored. The poller here was armed by + // the pilot at the default 60s; the routed update must replace it with a 1s one. + let polls = LockIsolated(0) + let worker = LoopNode( + title: "Worker", loopType: .goalBased, + goal: GoalSpec(summary: "ship it", predicate: "true", pollIntervalSeconds: 60)) + let (store, compositeID) = storeWithComposite( + subNodes: [worker], + onCheckPredicate: { _ in + polls.withValue { $0 += 1 } + return PredicateOutcome(passed: false) + }) + await store.handle(.pilotComposite(compositeID)) + + await store.handle( + .updateNode(worker.id, update: NodeUpdate(pollIntervalSeconds: 1))) + try await Task.sleep(for: .seconds(1.6)) + + #expect(polls.value >= 1) + } + + @Test + func anUnpilotedChildsGoalIsNeverPolled() async throws { + // A template's goal resolving would mark work done that never ran, so recurrence + // handed up from a child store is gated on its composite having been piloted. + let polls = LockIsolated(0) + let worker = LoopNode( + title: "Worker", loopType: .goalBased, + goal: GoalSpec(summary: "ship it", predicate: "true", pollIntervalSeconds: 1)) + let (store, _) = storeWithComposite( + subNodes: [worker], + onCheckPredicate: { _ in + polls.withValue { $0 += 1 } + return PredicateOutcome(passed: false) + }) + + await store.handle( + .updateNode(worker.id, update: NodeUpdate(pollIntervalSeconds: 1))) + try await Task.sleep(for: .seconds(1.6)) + + #expect(polls.value == 0) + } +}