ProtocolUpdater.cpp

// SPDX-License-Identifier: LGPL-3.0-or-later OR MIT

#include "codex/nodegraph/ProtocolUpdater.h"

#include <algorithm>
#include <array>
#include <cstddef>
#include <iterator>
#include <limits>
#include <stdexcept>
#include <string_view>
#include <unordered_map>
#include <unordered_set>
#include <utility>

namespace codexui::nodegraph {

void removeThreadGroup(NodeGraph::WriteAccess &write, const NodeRef &group) {
  const bool project = group->id().kind == NodeKind::Project;
  const auto relation = project ? RelationKind::ProjectMembership
                                : RelationKind::SectionMembership;
  for (const auto &node : write.orderedNodes()) {
    if (node->id().kind != NodeKind::Thread)
      continue;
    const auto membership = write.related(node, relation);
    if (std::ranges::find(membership, group) != membership.end())
      write.setField(node, project ? "projectId" : "section", Value(nullptr));
  }
  write.remove(group);
}

namespace {

constexpr std::size_t MaximumRetainedStreamBytes = 256 * 1024;
constexpr std::size_t RetainedStreamTailBytes = 192 * 1024;
constexpr std::size_t MaximumIndexedTextParts = 4096;

const Value::Object *objectMember(const Value::Object &object,
                                  std::string_view name) {
  const Value *value = valueMember(object, name);
  return value ? value->asObject() : nullptr;
}

const Value::Array *arrayMember(const Value::Object &object,
                                std::string_view name) {
  const Value *value = valueMember(object, name);
  return value ? value->asArray() : nullptr;
}

bool isLocalPrompt(NodeGraph::WriteAccess &write, const NodeRef &node) {
  if (!node || node->id().kind != NodeKind::Item)
    return false;
  const std::shared_ptr<const NodeState> state = write.state(node);
  const Value *type = valueMember(state->fields, "type");
  return type && type->asString() && *type->asString() == "localPrompt";
}

std::string firstId(const Value::Object &object,
                    std::span<const std::string_view> names) {
  for (const std::string_view name : names) {
    std::string value = scalarTextFromValue(valueMember(object, name));
    if (!value.empty())
      return value;
  }
  return {};
}

std::string scopedCanonical(std::string_view owner,
                            std::string_view protocolId) {
  std::string result = "scope:";
  result += std::to_string(owner.size());
  result += ':';
  result += owner;
  result += ':';
  result += std::to_string(protocolId.size());
  result += ':';
  result += protocolId;
  return result;
}

struct ThreadSettingDescriptor final {
  std::string_view key;
  std::string_view authority;
  std::string_view alias;
  bool acceptedFromResult = true;
  bool supersededByAlias = false;
};

constexpr std::array ThreadSettings{
    ThreadSettingDescriptor{"activePermissionProfile", "permissionProfile"},
    ThreadSettingDescriptor{"approvalPolicy", "approval"},
    ThreadSettingDescriptor{"approvalsReviewer", "reviewer"},
    ThreadSettingDescriptor{"collaborationMode", "collaboration"},
    ThreadSettingDescriptor{"cwd", "cwd"},
    ThreadSettingDescriptor{"effort", "effort", "reasoningEffort", true, true},
    ThreadSettingDescriptor{"instructionSources", "instructionSources"},
    ThreadSettingDescriptor{"model", "model"},
    ThreadSettingDescriptor{"modelProvider", "modelProvider"},
    ThreadSettingDescriptor{"permissions", "permissionProfile", {}, false},
    ThreadSettingDescriptor{"personality", "personality"},
    ThreadSettingDescriptor{"reasoningEffort", "effort", "effort"},
    ThreadSettingDescriptor{"sandbox", "sandbox", "sandboxPolicy", true, true},
    ThreadSettingDescriptor{"sandboxPolicy", "sandbox", "sandbox"},
    ThreadSettingDescriptor{"serviceTier", "serviceTier"},
    ThreadSettingDescriptor{"summary", "summary"}};

const ThreadSettingDescriptor *threadSetting(std::string_view key) noexcept {
  const auto found =
      std::ranges::find(ThreadSettings, key, &ThreadSettingDescriptor::key);
  return found == ThreadSettings.end() ? nullptr : &*found;
}

void mergeObject(NodeGraph::WriteAccess &write, const NodeRef &node,
                 const Value::Object &object,
                 std::string_view omittedChildField = {},
                 std::optional<std::uint64_t> preserveChangesAfter = {},
                 bool omitResultThreadSettings = false) {
  NodeState next = *write.state(node);
  for (const auto &[key, value] : object) {
    // Sparse retained history must not erase already observed lifecycle facts.
    if (value.isNull() &&
        (node->id().kind == NodeKind::Turn ||
         node->id().kind == NodeKind::Item) &&
        (key == "startedAt" || key == "completedAt" || key == "startedAtMs" ||
         key == "completedAtMs" || key == "durationMs") &&
        signedIntegerFromValue(valueMember(next, key)))
      continue;
    if ((!omittedChildField.empty() && key == omittedChildField) ||
        (omitResultThreadSettings && threadSetting(key) &&
         threadSetting(key)->acceptedFromResult))
      continue;
    if (preserveChangesAfter &&
        write.fieldChangedRevision(node, key) > *preserveChangesAfter)
      continue;
    next.fields.insert_or_assign(key, value);
  }
  if (const Value *status = valueMember(object, "status");
      status && (!preserveChangesAfter ||
                 write.statusChangedRevision(node) <= *preserveChangesAfter))
    next.status = nodeStatusFromValue(status);
  write.replaceState(node, std::move(next));
}

std::size_t utf8TailStart(std::string_view value,
                          std::size_t retainedBytes) noexcept {
  if (value.size() <= retainedBytes)
    return 0;
  std::size_t start = value.size() - retainedBytes;
  while (start < value.size() &&
         (static_cast<unsigned char>(value[start]) & 0xc0U) == 0x80U)
    ++start;
  return start;
}

Value::Object *textRetention(NodeState &state) {
  Value &retention = state.fields["textRetention"];
  if (!retention.isObject())
    retention = Value::Object{};
  return retention.asObject();
}

const Value::Object *textRetention(const NodeState &state) {
  const Value *retention = valueMember(state.fields, "textRetention");
  return retention ? retention->asObject() : nullptr;
}

bool hasTextRetention(const NodeState &state, std::string_view field) {
  const Value::Object *retention = textRetention(state);
  return retention && valueMember(*retention, field);
}

void clearTextRetention(NodeState &state, std::string_view field) {
  Value *retentionValue = nullptr;
  if (auto found = state.fields.find("textRetention");
      found != state.fields.end())
    retentionValue = &found->second;
  Value::Object *retention =
      retentionValue ? retentionValue->asObject() : nullptr;
  if (!retention)
    return;
  retention->erase(std::string(field));
  if (retention->empty())
    state.fields.erase("textRetention");
}

void updateTextRetention(NodeState &state, std::string_view field,
                         std::size_t retainedBytes,
                         std::size_t newlyDiscardedBytes) {
  Value::Object *retention = textRetention(state);
  Value &entryValue = (*retention)[std::string(field)];
  if (!entryValue.isObject())
    entryValue = Value::Object{};
  Value::Object &entry = *entryValue.asObject();
  const std::uint64_t previous =
      unsignedIntegerFromValue(valueMember(entry, "discardedBytes"))
          .value_or(0);
  const std::uint64_t increment =
      static_cast<std::uint64_t>(newlyDiscardedBytes);
  const std::uint64_t discarded =
      increment > std::numeric_limits<std::uint64_t>::max() - previous
          ? std::numeric_limits<std::uint64_t>::max()
          : previous + increment;
  entry.insert_or_assign("discardedBytes", Value(discarded));
  entry.insert_or_assign("retainedBytes",
                         Value(static_cast<std::uint64_t>(retainedBytes)));
}

void boundScalarText(NodeState &state, std::string_view field) {
  const auto found = state.fields.find(field);
  std::string *value =
      found == state.fields.end() ? nullptr : found->second.asString();
  if (!value)
    return;
  std::size_t discarded = 0;
  if (value->size() > MaximumRetainedStreamBytes) {
    discarded = utf8TailStart(*value, RetainedStreamTailBytes);
    value->erase(0, discarded);
  }
  if (discarded != 0 || hasTextRetention(state, field))
    updateTextRetention(state, field, value->size(), discarded);
}

void boundIndexedText(NodeState &state, std::string_view field) {
  const auto found = state.fields.find(field);
  Value::Array *parts =
      found == state.fields.end() ? nullptr : found->second.asArray();
  if (!parts)
    return;

  std::uint64_t total = 0;
  for (const Value &part : *parts) {
    const std::string *text = part.asString();
    if (!text)
      continue;
    const std::uint64_t bytes = static_cast<std::uint64_t>(text->size());
    total = bytes > std::numeric_limits<std::uint64_t>::max() - total
                ? std::numeric_limits<std::uint64_t>::max()
                : total + bytes;
  }

  std::size_t discarded = 0;
  if (total > MaximumRetainedStreamBytes) {
    std::uint64_t toDiscard = total - RetainedStreamTailBytes;
    for (Value &part : *parts) {
      std::string *text = part.asString();
      if (!text || text->empty() || toDiscard == 0)
        continue;
      std::size_t count = 0;
      if (toDiscard >= text->size()) {
        count = text->size();
      } else {
        count = utf8TailStart(*text, text->size() -
                                         static_cast<std::size_t>(toDiscard));
      }
      text->erase(0, count);
      toDiscard -= std::min<std::uint64_t>(toDiscard, count);
      discarded += count;
    }
  }

  std::size_t retained = 0;
  for (const Value &part : *parts)
    if (const std::string *text = part.asString())
      retained += text->size();
  if (discarded != 0 || hasTextRetention(state, field))
    updateTextRetention(state, field, retained, discarded);
}

void boundRetainedItemText(
    NodeGraph::WriteAccess &write, const NodeRef &item,
    const Value::Object &incoming,
    std::optional<std::uint64_t> preserveChangesAfter = {}) {
  NodeState next = *write.state(item);
  for (const std::string_view field :
       {std::string_view("text"), std::string_view("output"),
        std::string_view("aggregatedOutput"), std::string_view("summary"),
        std::string_view("content")}) {
    if (incoming.contains(field) &&
        (!preserveChangesAfter ||
         write.fieldChangedRevision(item, field) <= *preserveChangesAfter))
      clearTextRetention(next, field);
  }

  const std::string type =
      scalarTextFromValue(valueMember(next.fields, "type"));
  if (type == "commandExecution") {
    boundScalarText(next, "aggregatedOutput");
    boundScalarText(next, "output");
  } else if (type == "agentMessage" || type == "plan") {
    boundScalarText(next, "text");
  } else if (type == "reasoning") {
    boundIndexedText(next, "summary");
    boundIndexedText(next, "content");
  } else if (type == "fileChange") {
    boundScalarText(next, "output");
  } else if (type != "userMessage") {
    boundScalarText(next, "text");
    boundScalarText(next, "output");
    boundScalarText(next, "aggregatedOutput");
    boundIndexedText(next, "summary");
    boundIndexedText(next, "content");
  }
  write.replaceState(item, std::move(next));
}

void appendBoundedStringField(NodeGraph::WriteAccess &write,
                              const NodeRef &node, std::string field,
                              std::string_view suffix) {
  NodeState next = *write.state(node);
  Value &stored = next.fields[field];
  if (!stored.isString())
    stored = std::string{};
  std::string &existing = *stored.asString();

  std::size_t discarded = 0;
  if (suffix.size() > MaximumRetainedStreamBytes) {
    const std::size_t start = utf8TailStart(suffix, RetainedStreamTailBytes);
    discarded = existing.size() + start;
    existing.assign(suffix.substr(start));
  } else {
    existing.append(suffix);
    if (existing.size() > MaximumRetainedStreamBytes) {
      discarded = utf8TailStart(existing, RetainedStreamTailBytes);
      existing.erase(0, discarded);
    }
  }
  const std::size_t retained = existing.size();
  if (discarded != 0 || hasTextRetention(next, field))
    updateTextRetention(next, field, retained, discarded);
  write.replaceState(node, std::move(next));
}

void mergeSparseRateLimitFields(Value::Object &current,
                                const Value::Object &patch) {
  for (const auto &[key, value] : patch) {
    const auto found = current.find(key);
    const Value::Object *patchObject = value.asObject();
    if (found != current.end() && patchObject) {
      if (Value::Object *currentObject = found->second.asObject()) {
        mergeSparseRateLimitFields(*currentObject, *patchObject);
        continue;
      }
    }
    current.insert_or_assign(key, value);
  }
}

std::string retainedProtocolId(NodeGraph::WriteAccess &write,
                               const NodeRef &node) {
  if (!node)
    return {};
  const std::shared_ptr<const NodeState> state = write.state(node);
  const Value *retained = valueMember(*state, "protocolId");
  return retained ? scalarTextFromValue(retained) : node->id().canonical;
}

NodeRef ensureTurn(NodeGraph::WriteAccess &write, const NodeRef &thread,
                   std::string_view rawTurnId) {
  if (!thread || thread->id().kind != NodeKind::Thread || rawTurnId.empty())
    return {};
  NodeId id = scopedTurnNodeId(thread->id().canonical, rawTurnId);
  NodeRef turn = write.find(id);
  if (turn) {
    const std::shared_ptr<const NodeState> current = write.state(turn);
    if (scalarTextFromValue(valueMember(current->fields, "protocolId")) !=
            rawTurnId ||
        scalarTextFromValue(valueMember(current->fields, "protocolThreadId")) !=
            thread->id().canonical) {
      NodeState next = *current;
      next.fields.insert_or_assign("protocolId", Value(rawTurnId));
      next.fields.insert_or_assign("protocolThreadId",
                                   Value(thread->id().canonical));
      write.replaceState(turn, std::move(next));
    }
  } else {
    NodeState next;
    next.fields = {{"protocolId", Value(rawTurnId)},
                   {"protocolThreadId", Value(thread->id().canonical)}};
    turn = write.upsert(std::move(id), std::move(next));
  }
  write.setParent(thread, turn);
  return turn;
}

bool isActiveTurn(const NodeState &state) {
  return state.status == NodeStatus::Running;
}

void updateActiveTurn(NodeGraph::WriteAccess &write, const NodeRef &thread,
                      const NodeRef &turn) {
  if (!thread || !turn)
    return;
  if (isActiveTurn(*write.state(turn))) {
    const std::array<NodeRef, 1> active{turn};
    write.replaceRelated(thread, RelationKind::ActiveTurn, active);
    return;
  }
  write.unrelate(thread, RelationKind::ActiveTurn, turn);
}

void refreshActiveTurn(NodeGraph::WriteAccess &write, const NodeRef &thread) {
  NodeRef active;
  const std::vector<NodeRef> turns = write.children(thread);
  for (auto iterator = turns.rbegin(); iterator != turns.rend(); ++iterator) {
    if (*iterator && (*iterator)->id().kind == NodeKind::Turn &&
        isActiveTurn(*write.state(*iterator))) {
      active = *iterator;
      break;
    }
  }
  if (active) {
    const std::array<NodeRef, 1> current{std::move(active)};
    write.replaceRelated(thread, RelationKind::ActiveTurn, current);
  } else {
    write.replaceRelated(thread, RelationKind::ActiveTurn,
                         std::span<const NodeRef>{});
  }
}

bool mergeThreadSettings(
    NodeGraph::WriteAccess &write, const NodeRef &thread,
    const Value::Object &settings, std::uint64_t acknowledgedAt = 0,
    std::optional<std::uint64_t> preserveChangesAfter = {}) {
  if (!thread)
    return false;
  const auto current = write.state(thread);
  NodeState next = *current;
  Value::Object acknowledgements;
  if (const Value *value =
          valueMember(current->fields, "settingsAcknowledgements");
      value && value->asObject())
    acknowledgements = *value->asObject();
  bool effectiveChange = false;
  for (const auto &[key, value] : settings) {
    const ThreadSettingDescriptor *descriptor = threadSetting(key);
    if (acknowledgedAt == 0 &&
        (!descriptor || !descriptor->acceptedFromResult ||
         (descriptor->supersededByAlias &&
          settings.contains(descriptor->alias))))
      continue;
    if (preserveChangesAfter &&
        (write.fieldChangedRevision(thread, key) > *preserveChangesAfter ||
         (descriptor && !descriptor->alias.empty() &&
          write.fieldChangedRevision(thread, descriptor->alias) >
              *preserveChangesAfter) ||
         (descriptor &&
          unsignedIntegerFromValue(
              valueMember(acknowledgements, descriptor->authority))
                  .value_or(0) > *preserveChangesAfter)))
      continue;
    const Value *previous = valueMember(next.fields, key);
    if (!previous && descriptor && !descriptor->alias.empty())
      previous = valueMember(next.fields, descriptor->alias);
    effectiveChange |=
        value.isNull() ? previous != nullptr : !previous || *previous != value;
    if (descriptor && !descriptor->alias.empty()) {
      next.fields.erase(std::string(descriptor->alias));
    }
    if (value.isNull()) {
      next.fields.erase(key);
    } else {
      next.fields.insert_or_assign(key, value);
    }
    if (acknowledgedAt != 0 && descriptor)
      acknowledgements.insert_or_assign(std::string(descriptor->authority),
                                        Value(acknowledgedAt));
  }
  if (acknowledgedAt != 0 && !acknowledgements.empty())
    next.fields.insert_or_assign("settingsAcknowledgements",
                                 Value(std::move(acknowledgements)));
  if (next.fields == current->fields)
    return effectiveChange;
  next.fields.insert_or_assign("settingsRevision", Value(write.revision() + 1));
  write.replaceState(thread, std::move(next));
  return effectiveChange;
}

NodeRef ensureItem(NodeGraph::WriteAccess &write, const NodeRef &turn,
                   std::string_view rawItemId) {
  if (!turn || turn->id().kind != NodeKind::Turn || rawItemId.empty())
    return {};
  NodeId id = scopedItemNodeId(turn->id(), rawItemId);
  NodeRef item = write.find(id);
  const std::string turnId = retainedProtocolId(write, turn);
  const NodeRef thread = write.parent(turn);
  const std::string threadId = thread ? thread->id().canonical : std::string{};
  if (item) {
    const std::shared_ptr<const NodeState> current = write.state(item);
    const bool protocolChanged =
        scalarTextFromValue(valueMember(current->fields, "protocolId")) !=
            rawItemId ||
        scalarTextFromValue(valueMember(current->fields, "protocolTurnId")) !=
            turnId ||
        (!threadId.empty() &&
         scalarTextFromValue(
             valueMember(current->fields, "protocolThreadId")) != threadId);
    if (protocolChanged) {
      NodeState next = *current;
      next.fields.insert_or_assign("protocolId", Value(rawItemId));
      next.fields.insert_or_assign("protocolTurnId", Value(turnId));
      if (!threadId.empty())
        next.fields.insert_or_assign("protocolThreadId", Value(threadId));
      write.replaceState(item, std::move(next));
    }
  } else {
    NodeState next;
    next.fields = {{"protocolId", Value(rawItemId)},
                   {"protocolTurnId", Value(turnId)}};
    if (!threadId.empty())
      next.fields.emplace("protocolThreadId", Value(threadId));
    item = write.upsert(std::move(id), std::move(next));
  }
  if (write.parent(item) != turn) {
    NodeRef before;
    // Unconsumed steering stays after server history, including after request
    // acceptance. Only the correlated user item establishes its final slot.
    for (const NodeRef &prompt :
         write.related(thread, RelationKind::PendingPrompt)) {
      if (write.parent(prompt) != turn || !isLocalPrompt(write, prompt))
        continue;
      const auto state = write.state(prompt);
      const std::string dispatch =
          scalarTextFromValue(valueMember(*state, "dispatchState"));
      if (!boolFromValue(valueMember(*state, "startsTurn")) &&
          dispatch != "failed" && dispatch != "uncertain") {
        before = prompt;
        break;
      }
    }
    write.setParent(turn, item, before);
  }
  return item;
}

NodeRef applyHookNotification(NodeGraph::WriteAccess &write,
                              const DecodedMessage &message) {
  const Value::Object *run = objectMember(message.payload, "run");
  const std::string runId =
      run ? scalarTextFromValue(valueMember(*run, "id")) : std::string{};
  if (runId.empty())
    return {};

  NodeRef hook = write.upsert({NodeKind::Hook, runId});
  mergeObject(write, hook, *run);
  write.setField(hook, "protocolId", Value(runId));
  write.setField(hook, "lastMethod", Value(message.method));

  const Value *threadValue = valueMember(message.payload, "threadId");
  const Value *turnValue = valueMember(message.payload, "turnId");
  const std::string threadId = scalarTextFromValue(threadValue);
  const std::string turnId = scalarTextFromValue(turnValue);
  if (threadId.empty()) {
    write.eraseField(hook, "threadId");
    write.eraseField(hook, "protocolThreadId");
  } else {
    write.setField(hook, "threadId", Value(threadId));
    write.setField(hook, "protocolThreadId", Value(threadId));
  }
  if (turnId.empty()) {
    write.eraseField(hook, "turnId");
    write.eraseField(hook, "protocolTurnId");
  } else {
    write.setField(hook, "turnId", Value(turnId));
    write.setField(hook, "protocolTurnId", Value(turnId));
  }

  if (!threadId.empty()) {
    NodeRef thread = write.upsert({NodeKind::Thread, threadId});
    NodeRef owner = thread;
    if (!turnId.empty())
      if (NodeRef turn = ensureTurn(write, thread, turnId))
        owner = std::move(turn);
    write.setParent(owner, hook);
  }

  const Value *rawStatus = valueMember(*run, "status");
  NodeStatus status = nodeStatusFromValue(rawStatus);
  if (statusTextFromValue(rawStatus).empty()) {
    const bool started = message.method == "hook/started";
    status = started ? NodeStatus::Running : NodeStatus::Completed;
    write.setField(hook, "status", Value(started ? "running" : "completed"));
  }
  write.setStatus(hook, status);
  return hook;
}

std::optional<std::size_t> indexValue(const Value *value) {
  const auto index = unsignedIntegerFromValue(value);
  if (!index || *index > std::numeric_limits<std::size_t>::max())
    return std::nullopt;
  return static_cast<std::size_t>(*index);
}

void appendIndexedField(NodeGraph::WriteAccess &write, const NodeRef &node,
                        std::string field, std::size_t index,
                        std::string_view suffix) {
  if (index >= MaximumIndexedTextParts)
    return;
  NodeState next = *write.state(node);
  Value &stored = next.fields[field];
  if (!stored.isArray())
    stored = Value::Array{};
  Value::Array &parts = *stored.asArray();
  if (parts.size() <= index)
    parts.resize(index + 1);
  if (!parts[index].isString())
    parts[index] = std::string{};
  parts[index].asString()->append(suffix);
  boundIndexedText(next, field);
  write.replaceState(node, std::move(next));
}

void appendSemanticDelta(NodeGraph::WriteAccess &write, const NodeRef &item,
                         std::string_view method,
                         const Value::Object &payload) {
  if (method == "item/reasoning/summaryPartAdded") {
    appendIndexedField(
        write, item, "summary",
        indexValue(valueMember(payload, "summaryIndex")).value_or(0), {});
    return;
  }
  const std::string delta = scalarTextFromValue(valueMember(payload, "delta"));
  if (delta.empty())
    return;
  if (method == "item/reasoning/summaryTextDelta") {
    appendIndexedField(
        write, item, "summary",
        indexValue(valueMember(payload, "summaryIndex")).value_or(0), delta);
    return;
  }
  if (method == "item/reasoning/textDelta") {
    appendIndexedField(
        write, item, "content",
        indexValue(valueMember(payload, "contentIndex")).value_or(0), delta);
    return;
  }
  if (method == "item/commandExecution/outputDelta") {
    appendBoundedStringField(write, item, "aggregatedOutput", delta);
    return;
  }
  if (method == "item/fileChange/outputDelta") {
    appendBoundedStringField(write, item, "output", delta);
    return;
  }
  appendBoundedStringField(write, item, "text", delta);
}

std::string addressedId(const Value::Object &payload, NodeKind kind) {
  switch (kind) {
  case NodeKind::Thread: {
    constexpr std::array names{std::string_view("threadId"),
                               std::string_view("conversationId")};
    return firstId(payload, names);
  }
  case NodeKind::Turn: {
    constexpr std::array names{std::string_view("turnId")};
    return firstId(payload, names);
  }
  case NodeKind::Item: {
    constexpr std::array names{std::string_view("itemId"),
                               std::string_view("callId")};
    return firstId(payload, names);
  }
  case NodeKind::Project: {
    constexpr std::array names{std::string_view("projectId")};
    return firstId(payload, names);
  }
  case NodeKind::ThreadSection: {
    constexpr std::array names{std::string_view("sectionId")};
    return firstId(payload, names);
  }
  case NodeKind::Process: {
    constexpr std::array names{std::string_view("processId"),
                               std::string_view("processHandle"),
                               std::string_view("commandId")};
    return firstId(payload, names);
  }
  case NodeKind::RealtimeSession: {
    constexpr std::array names{std::string_view("sessionId")};
    return firstId(payload, names);
  }
  case NodeKind::FilesystemWatch: {
    constexpr std::array names{std::string_view("watchId")};
    return firstId(payload, names);
  }
  default:
    return {};
  }
}

std::string nestedId(const Value::Object &object,
                     std::string_view fallback = {}) {
  constexpr std::array names{
      std::string_view("id"), std::string_view("threadId"),
      std::string_view("turnId"), std::string_view("itemId")};
  std::string result = firstId(object, names);
  return result.empty() ? std::string(fallback) : result;
}

bool beginsWith(std::string_view value, std::string_view prefix) {
  return value.starts_with(prefix);
}

bool isItemDeltaMethod(std::string_view method) {
  return method == "item/agentMessage/delta" || method == "item/plan/delta" ||
         method == "item/reasoning/summaryTextDelta" ||
         method == "item/reasoning/summaryPartAdded" ||
         method == "item/reasoning/textDelta" ||
         method == "item/commandExecution/outputDelta" ||
         method == "item/fileChange/outputDelta";
}

bool isProviderNoticeMethod(std::string_view method) {
  return method == "error" || method == "warning" ||
         method == "guardianWarning" || method == "deprecationNotice" ||
         method == "configWarning" || method == "windows/worldWritableWarning";
}

std::string mcpServerNodeKey(std::string_view method,
                             const Value::Object &payload) {
  if (method == "mcpServer/event/stream/notification") {
    const std::string subscription =
        scalarTextFromValue(valueMember(payload, "subscriptionId"));
    return subscription.empty() ? std::string(method)
                                : scopedCanonical("subscription", subscription);
  }

  const std::string name = scalarTextFromValue(valueMember(payload, "name"));
  if (name.empty())
    return std::string(method);
  const std::string threadId =
      scalarTextFromValue(valueMember(payload, "threadId"));
  return scopedCanonical(threadId.empty() ? std::string_view("global")
                                          : std::string_view(threadId),
                         name);
}

NodeKind kindForMethod(std::string_view method) {
  if (beginsWith(method, "item/"))
    return NodeKind::Item;
  if (beginsWith(method, "turn/"))
    return NodeKind::Turn;
  if (beginsWith(method, "thread/realtime/"))
    return NodeKind::RealtimeSession;
  if (beginsWith(method, "thread/"))
    return NodeKind::Thread;
  if (beginsWith(method, "project/"))
    return NodeKind::Project;
  if (beginsWith(method, "threadSection/"))
    return NodeKind::ThreadSection;
  if (beginsWith(method, "command/") || beginsWith(method, "process/"))
    return NodeKind::Process;
  if (beginsWith(method, "account/"))
    return NodeKind::Account;
  if (beginsWith(method, "config") || beginsWith(method, "skills/config"))
    return NodeKind::Configuration;
  if (beginsWith(method, "mcpServer/"))
    return NodeKind::McpServer;
  if (beginsWith(method, "fs/"))
    return NodeKind::FilesystemWatch;
  if (method == "error" || method == "warning" || method == "guardianWarning" ||
      method == "deprecationNotice" || method == "configWarning" ||
      method == "windows/worldWritableWarning")
    return NodeKind::Notice;
  return NodeKind::Catalog;
}

std::string catalogKey(std::string_view method) {
  const std::size_t slash = method.find('/');
  return std::string(method.substr(0, slash));
}

struct CatalogEntitySeed final {
  NodeKind kind = NodeKind::CatalogEntry;
  const Value::Object *fields = nullptr;
  std::string scope;
};

std::string catalogEntityId(const Value::Object &fields) {
  constexpr std::array names{
      std::string_view("model"),      std::string_view("id"),
      std::string_view("key"),        std::string_view("name"),
      std::string_view("path"),       std::string_view("connectorId"),
      std::string_view("runtimeName")};
  return firstId(fields, names);
}

void appendDirectCatalogEntries(std::vector<CatalogEntitySeed> &entries,
                                const Value::Array *values, NodeKind kind,
                                std::string_view scope = {}) {
  if (!values)
    return;
  entries.reserve(entries.size() + values->size());
  for (const Value &value : *values) {
    if (const Value::Object *object = value.asObject())
      entries.push_back({kind, object, std::string(scope)});
  }
}

// Catalog envelopes retain cursors/errors on their Catalog node. Concrete
// rows live as ordered child nodes so declared entity kinds are not opaque
// blobs and stable NodeRefs survive ordinary refreshes.
bool applyNaturalCatalogSnapshot(NodeGraph::WriteAccess &write,
                                 const DecodedMessage &message) {
  const std::string_view method = message.method;
  const bool appNotification =
      message.kind == DecodedMessageKind::ServerNotification &&
      method == "app/list/updated";
  if (message.kind != DecodedMessageKind::ClientResult && !appNotification)
    return false;

  std::string_view catalogId;
  NodeKind directKind = NodeKind::CatalogEntry;
  const Value::Array *directEntries = nullptr;
  std::vector<CatalogEntitySeed> entries;

  if (method == "model/list") {
    catalogId = "model";
    directEntries = arrayMember(message.payload, "data");
  } else if (method == "permissionProfile/list") {
    catalogId = "permissionProfile";
    directKind = NodeKind::PermissionProfile;
    directEntries = arrayMember(message.payload, "data");
  } else if (method == "experimentalFeature/list") {
    catalogId = "experimentalFeature";
    directEntries = arrayMember(message.payload, "data");
  } else if (method == "collaborationMode/list") {
    catalogId = "collaborationMode";
    directEntries = arrayMember(message.payload, "data");
  } else if (method == "mcpServerStatus/list") {
    catalogId = "mcpServer";
    directKind = NodeKind::McpServer;
    directEntries = arrayMember(message.payload, "data");
  } else if (method == "app/list" || appNotification) {
    catalogId = "app";
    directKind = NodeKind::App;
    directEntries = arrayMember(message.payload, "data");
    if (appNotification && !directEntries)
      return false;
  } else if (method == "skills/list") {
    catalogId = "skills";
    if (const Value::Array *groups = arrayMember(message.payload, "data")) {
      for (const Value &value : *groups) {
        const Value::Object *group = value.asObject();
        if (!group)
          continue;
        appendDirectCatalogEntries(
            entries, arrayMember(*group, "skills"), NodeKind::Skill,
            scalarTextFromValue(valueMember(*group, "cwd")));
      }
    }
  } else if (method == "hooks/list") {
    catalogId = "hooks";
    if (const Value::Array *groups = arrayMember(message.payload, "data")) {
      for (const Value &value : *groups) {
        const Value::Object *group = value.asObject();
        if (!group)
          continue;
        appendDirectCatalogEntries(
            entries, arrayMember(*group, "hooks"), NodeKind::Hook,
            scalarTextFromValue(valueMember(*group, "cwd")));
      }
    }
  } else if (method == "plugin/list") {
    catalogId = "plugin";
    if (const Value::Array *marketplaces =
            arrayMember(message.payload, "marketplaces")) {
      for (const Value &value : *marketplaces) {
        const Value::Object *marketplace = value.asObject();
        if (!marketplace)
          continue;
        std::string scope =
            scalarTextFromValue(valueMember(*marketplace, "name"));
        if (scope.empty())
          scope = scalarTextFromValue(valueMember(*marketplace, "path"));
        appendDirectCatalogEntries(entries,
                                   arrayMember(*marketplace, "plugins"),
                                   NodeKind::Plugin, scope);
      }
    }
  } else {
    return false;
  }

  appendDirectCatalogEntries(entries, directEntries, directKind);
  NodeRef catalog = write.upsert({NodeKind::Catalog, std::string(catalogId)});
  const std::vector<NodeRef> previous = write.children(catalog);
  mergeObject(write, catalog, message.payload);
  write.setField(catalog, "lastMethod", Value(std::string(method)));
  write.eraseField(catalog, "stale");
  write.eraseField(catalog, "invalidatedBy");

  const bool continuation =
      !scalarTextFromValue(valueMember(message.payload, "cursor")).empty();
  std::vector<NodeRef> ordered =
      continuation ? previous : std::vector<NodeRef>{};
  ordered.reserve(ordered.size() + entries.size());
  std::unordered_set<const Node *> retained;
  retained.reserve(ordered.size() + entries.size());
  for (const NodeRef &entry : ordered)
    retained.insert(entry.get());

  std::size_t anonymous = 0;
  for (const CatalogEntitySeed &seed : entries) {
    if (!seed.fields)
      continue;
    std::string protocolId = catalogEntityId(*seed.fields);
    if (protocolId.empty())
      protocolId = "anonymous:" + std::to_string(anonymous++);
    std::string owner(catalogId);
    if (!seed.scope.empty()) {
      owner += ':';
      owner += seed.scope;
    }
    NodeState state;
    state.status = nodeStatusFromValue(valueMember(*seed.fields, "status"));
    state.fields = *seed.fields;
    state.fields.insert_or_assign("protocolId", Value(protocolId));
    state.fields.insert_or_assign("catalog", Value(std::string(catalogId)));
    if (!seed.scope.empty())
      state.fields.insert_or_assign("catalogScope", Value(seed.scope));
    NodeId id{seed.kind, scopedCanonical(owner, protocolId)};
    NodeRef entry = write.find(id);
    if (entry)
      write.replaceState(entry, std::move(state));
    else
      entry = write.upsert(std::move(id), std::move(state));
    if (retained.insert(entry.get()).second)
      ordered.emplace_back(std::move(entry));
  }

  write.replaceChildren(catalog, ordered);
  if (!continuation) {
    std::vector<NodeRef> omitted;
    omitted.reserve(previous.size());
    for (const NodeRef &entry : previous)
      if (entry && !retained.contains(entry.get()))
        omitted.emplace_back(entry);
    write.removeMany(omitted);
  }
  return true;
}

bool applyAutoApprovalReview(NodeGraph::WriteAccess &write,
                             const DecodedMessage &message) {
  const bool started = message.method == "item/autoApprovalReview/started";
  const bool completed = message.method == "item/autoApprovalReview/completed";
  if (!started && !completed)
    return false;

  const std::string threadId = addressedId(message.payload, NodeKind::Thread);
  const std::string turnId = addressedId(message.payload, NodeKind::Turn);
  if (threadId.empty() || turnId.empty())
    return true;
  NodeRef thread = write.upsert({NodeKind::Thread, threadId});
  NodeRef turn = ensureTurn(write, thread, turnId);

  const std::string targetId =
      scalarTextFromValue(valueMember(message.payload, "targetItemId"));
  NodeRef target = ensureItem(write, turn, targetId);
  std::string reviewId =
      scalarTextFromValue(valueMember(message.payload, "reviewId"));
  if (reviewId.empty() && !targetId.empty())
    reviewId = "auto-review:" + targetId;
  if (reviewId.empty()) {
    write.setField(turn, "lastAutoApprovalReview", Value(message.payload));
    return true;
  }

  NodeRef review = ensureItem(write, turn, reviewId);
  mergeObject(write, review, message.payload);
  write.setField(review, "type", Value("autoApprovalReview"));
  write.setField(review, "phase", Value(started ? "started" : "completed"));
  write.setField(review, "protocolId", Value(reviewId));
  write.setStatus(review,
                  started ? NodeStatus::Running : NodeStatus::Completed);
  if (target) {
    const std::array<NodeRef, 1> targets{target};
    write.replaceRelated(review, RelationKind::ReviewTarget, targets);
  } else {
    write.replaceRelated(review, RelationKind::ReviewTarget,
                         std::span<const NodeRef>{});
  }
  return true;
}

bool currentInteraction(NodeGraph::WriteAccess &write,
                        const NodeRef &interaction) {
  return interaction && interaction->id().kind == NodeKind::Interaction &&
         write.find(interaction->id()) == interaction;
}

bool interactionGenerationDiffers(const NodeState &state,
                                  const DecodedMessage &message) {
  if (message.connectionGeneration) {
    const Value *stored = valueMember(state.fields, "connectionGeneration");
    if (!stored || unsignedIntegerFromValue(stored).value_or(0) !=
                       *message.connectionGeneration)
      return true;
  }