Add trajectory reviewer

.Rbuildignore

@@ -14,4 +14,5 @@ CLAUDE.md
 ^pkgdown$
 ^\.github$
 inst/hex/
+^commons-review\.jsonl$
 inst/manifest.json

.gitignore

@@ -14,3 +14,4 @@ CLAUDE.local.md
 AGENTS.override.md
 inst/hex/output
 .shinychat/
+commons-review.jsonl

DESCRIPTION

@@ -4,6 +4,7 @@ Version: 0.0.0.9000
 Authors@R: c(
     person("Simon", "Couch", , "simon.couch@posit.co", role = c("aut", "cre"),
            comment = c(ORCID = "0000-0001-5676-5107")),
+    person("Sara", "Altman", , "sara.altman@posit.co", role = "aut"),
     person("Posit Software, PBC", role = c("cph", "fnd"),
            comment = c(ROR = "03wc8by49"))
   )
@@ -38,16 +39,18 @@ Imports:
     utils
 Suggests:
     bsicons,
+    bslib (>= 0.11.0),
     dbplyr,
     dplyr,
     htmltools,
     otel (>= 0.2.0),
     otelsdk (>= 0.2.0),
     pins,
+    plotly,
     ragg,
     readr,
     rmarkdown,
-    shiny,
+    shiny (>= 1.11.1),
     shinychat (> 0.4.0),
     testthat (>= 3.0.0),
     vitals,

NAMESPACE

@@ -7,8 +7,9 @@ export(context_layer)
 export(data_source)
 export(list_tables)
 export(measure)
-export(read_trajectories)
 export(semantic_layer)
+export(trajectory_read)
+export(trajectory_review)
 importFrom(R6,R6Class)
 importFrom(coro,async_generator)
 importFrom(coro,await_each)

R/commons.R

@@ -65,7 +65,7 @@
 #'   setting itself when needed, though capture only starts once the content
 #'   restarts. Locally, commons configures \pkg{otelsdk}'s file exporter
 #'   automatically when no exporter is set up. Read trajectories back with
-#'   [read_trajectories()].
+#'   [trajectory_read()].
 #' @param share_with An optional character vector of Connect usernames granted
 #'   access to this content's trajectories when running on Posit Connect.
 #'   Reading traces requires editor-level access, so named users are added as

R/tagging.R

@@ -28,25 +28,31 @@ commons_last_provenance <- function(client) {
 # reinstate provenance pills when an existing conversation seeds a new
 # session, since pills are otherwise only injected as live turns complete.
 commons_exchange_provenance <- function(turns, corpus = list()) {
+  lapply(split_exchanges(turns), function(exchange) {
+    derive_provenance(
+      unlist(lapply(exchange, turn_tags)) %||% character(),
+      unlist(lapply(exchange, turn_text)) %||% character(),
+      corpus
+    )
+  })
+}
+
+# Tool-result UserTurns stay with the exchange that initiated them.
+split_exchanges <- function(turns) {
   out <- list()
-  tags <- character()
-  text <- character()
-  started <- FALSE
+  current <- NULL
   for (turn in turns) {
     if (identical(turn@role, "user") && !turn_has_tool_result(turn)) {
-      if (started) {
-        out[[length(out) + 1]] <- derive_provenance(tags, text, corpus)
+      if (!is.null(current)) {
+        out[[length(out) + 1]] <- current
       }
-      tags <- character()
-      text <- character()
-      started <- TRUE
-    } else if (started) {
-      tags <- c(tags, turn_tags(turn))
-      text <- c(text, turn_text(turn))
+      current <- list(turn)
+    } else if (!is.null(current)) {
+      current[[length(current) + 1]] <- turn
     }
   }
-  if (started) {
-    out[[length(out) + 1]] <- derive_provenance(tags, text, corpus)
+  if (!is.null(current)) {
+    out[[length(out) + 1]] <- current
   }
   out
 }

R/trajectories.R → R/trajectory-read.R

@@ -1,7 +1,7 @@
 #' Read commons trajectories
 #'
 #' @description
-#' `read_trajectories()` reads conversation trajectories captured by
+#' `trajectory_read()` reads conversation trajectories captured by
 #' [commons()] when `log = TRUE`. Trajectories are recorded as OpenTelemetry
 #' spans—see the `log` argument of [commons()] for how capture is enabled—and
 #' read back from Posit Connect's content observability store or from local
@@ -51,9 +51,13 @@
 #' ```
 #'
 #' @return A list of conversations, named by conversation id and ordered
-#'   oldest-first. Each conversation is a list of [ellmer::Turn]s.
+#'   oldest-first. Each conversation is a list of [ellmer::Turn]s and carries
+#'   a `last_active` attribute: a `POSIXct` giving the time of the
+#'   conversation's most recent chat activity. The list carries a `source`
+#'   attribute identifying the local trace directory or Connect content from
+#'   which it was read.
 #' @export
-read_trajectories <- function(
+trajectory_read <- function(
   source = NULL,
   ...,
   n = NULL,
@@ -75,9 +79,24 @@ read_trajectories <- function(
   if (!is.null(n)) {
     trajectories <- utils::tail(trajectories, n)
   }
+  attr(trajectories, "source") <- trajectory_source_record(resolved)
   trajectories
 }
 
+trajectory_source_record <- function(resolved) {
+  if (identical(resolved$kind, "connect")) {
+    return(list(
+      kind = "connect",
+      server = resolved$client$server,
+      content_guid = resolved$guid
+    ))
+  }
+  list(
+    kind = "local",
+    path = normalizePath(resolved$path, mustWork = FALSE)
+  )
+}
+
 # Dates and date strings both resolve to local midnight; as.POSIXct() alone
 # would silently read a Date as UTC midnight.
 check_window_bound <- function(
@@ -528,7 +547,17 @@ posixct_nanos <- function(time) {
 # span in a conversation carries the whole trajectory: group chat spans by
 # conversation, keep the last one, and parse its GenAI-semconv messages.
 build_trajectories <- function(spans) {
-  lapply(latest_chat_spans(spans), trajectory_turns)
+  lapply(latest_chat_spans(spans), function(span) {
+    turns <- trajectory_turns(span)
+    # Keep conversations directly usable with ellmer's chat$set_turns().
+    attr(turns, "last_active") <- nano_posixct(span_time(span))
+    turns
+  })
+}
+
+# Second precision is sufficient; the origin supports R < 4.3.
+nano_posixct <- function(time) {
+  as.POSIXct(as.numeric(time) / 1e9, origin = "1970-01-01")
 }
 
 # The latest chat span per conversation, named by conversation id and

R/trajectory-review-log.R

@@ -0,0 +1,166 @@
+new_review_event <- function(
+  trajectories,
+  key,
+  action,
+  user,
+  source = trajectory_source(trajectories),
+  note = NULL
+) {
+  exchange <- key$exchange
+  record <- list(
+    schema_version = 1L,
+    event_id = new_review_event_id(),
+    time = review_timestamp(),
+    user = user,
+    source = source,
+    conversation = names(trajectories)[[key$conversation]],
+    exchange = exchange,
+    action = action,
+    note = note
+  )
+
+  if (!is.null(exchange)) {
+    turns <- split_exchanges(trajectories[[key$conversation]])[[exchange]]
+    provenance <- exchange_provenance(turns)
+    record$question <- turns[[1]]@text
+    record$tag <- if (is.na(provenance$tag)) "none" else provenance$tag
+  }
+
+  record
+}
+
+new_review_event_id <- function() {
+  suffix <- paste(sample(c(letters, 0:9), 12, replace = TRUE), collapse = "")
+  paste0(format(Sys.time(), "%Y%m%dt%H%M%OS6", tz = "UTC"), "-", suffix)
+}
+
+review_timestamp <- function() {
+  format(Sys.time(), "%Y-%m-%dT%H:%M:%SZ", tz = "UTC")
+}
+
+review_user <- function(session) {
+  user <- session$user
+  if (!rlang::is_string(user) || !nzchar(user)) {
+    return("unknown")
+  }
+  user
+}
+
+trajectory_source <- function(trajectories) {
+  attr(trajectories, "source") %||% list(kind = "unknown")
+}
+
+append_review_record <- function(file, record) {
+  line <- jsonlite::toJSON(drop_nulls(record), auto_unbox = TRUE)
+  cat(line, "\n", file = file, sep = "", append = TRUE)
+}
+
+read_review_records <- function(file) {
+  if (!file.exists(file)) {
+    return(list())
+  }
+
+  records <- list()
+  invalid <- integer()
+  lines <- readLines(file, warn = FALSE)
+  for (i in seq_along(lines)) {
+    record <- tryCatch(
+      jsonlite::fromJSON(lines[[i]], simplifyVector = FALSE),
+      error = function(err) NULL
+    )
+    if (!is_review_record(record)) {
+      invalid <- c(invalid, i)
+      next
+    }
+    records[[length(records) + 1]] <- record
+  }
+
+  if (length(invalid) > 0) {
+    cli::cli_warn(
+      "Ignoring {cli::qty(invalid)}invalid review record{?s} on line{?s}
+       {invalid} of {.file {file}}."
+    )
+  }
+  records
+}
+
+is_review_record <- function(record) {
+  if (
+    !is.list(record) ||
+      !rlang::is_string(record$conversation) ||
+      !rlang::is_string(record$action) ||
+      !record$action %in% c("flag", "unflag", "note")
+  ) {
+    return(FALSE)
+  }
+  exchange <- record$exchange
+  if (
+    !is.null(exchange) &&
+      (!is.numeric(exchange) ||
+        length(exchange) != 1 ||
+        !is.finite(exchange) ||
+        exchange < 1 ||
+        exchange != trunc(exchange))
+  ) {
+    return(FALSE)
+  }
+  !identical(record$action, "note") || rlang::is_string(record$note)
+}
+
+actionable_review_records <- function(records) {
+  note_indices <- integer()
+  latest_flag_indices <- integer()
+
+  for (i in seq_along(records)) {
+    record <- records[[i]]
+    if (identical(record$action, "note")) {
+      note_indices <- c(note_indices, i)
+    } else {
+      key <- review_key(record$conversation, record$exchange)
+      latest_flag_indices[[key]] <- i
+    }
+  }
+
+  active_flag_indices <- unname(latest_flag_indices)
+  active_flag_indices <- active_flag_indices[vapply(
+    records[active_flag_indices],
+    function(record) identical(record$action, "flag"),
+    logical(1)
+  )]
+  records[sort(c(note_indices, active_flag_indices))]
+}
+
+review_flags <- function(records) {
+  active <- Filter(
+    function(record) identical(record$action, "flag"),
+    actionable_review_records(records)
+  )
+  vapply(
+    active,
+    function(record) review_key(record$conversation, record$exchange),
+    character(1)
+  )
+}
+
+review_notes <- function(records) {
+  Filter(
+    function(record) identical(record$action, "note"),
+    records
+  )
+}
+
+review_key <- function(id, exchange = NULL) {
+  paste(c(id, exchange), collapse = "#")
+}
+
+parse_review_time <- function(x) {
+  time <- as.POSIXct(
+    x,
+    format = "%Y-%m-%dT%H:%M:%SZ",
+    tz = "UTC"
+  )
+  if (!is.na(time)) {
+    return(time)
+  }
+  as.POSIXct(x, format = "%Y-%m-%dT%H:%M:%S%z")
+}