fix(server,tui): restore output panel + full event history on session reopen
Reopening a session after relaunch sent only a thin summary snapshot, so the output panel was blank and the event list showed at most 7 entries (of a 100+ event run). - SessionSnapshot now carries lastOutput/lastResponse, resolved from the last InferenceCompleted artifact in CAS; tui-go onSnapshot restores the output panel - recentEvents cap raised 7 -> 200 so a normal session's event list isn't truncated - resolveLastOutput extracted as a helper Note: artifact content is unaffected — verified intact and CAS-retrievable for role_pipeline sessions (artread); content-less artifacts occur only in workflows that record no ArtifactContentStored (e.g. healthcheck/process-result stages).
This commit is contained in:
@@ -42,6 +42,11 @@ import com.correx.core.events.types.SessionId
|
|||||||
import com.correx.core.kernel.orchestration.OrchestrationRepository
|
import com.correx.core.kernel.orchestration.OrchestrationRepository
|
||||||
import com.correx.core.tools.registry.ToolRegistry
|
import com.correx.core.tools.registry.ToolRegistry
|
||||||
|
|
||||||
|
// How many of a session's events the reopen snapshot carries. The full log lives in the event
|
||||||
|
// store; this bounds the snapshot frame while being generous enough that a normal session's event
|
||||||
|
// list isn't truncated on reopen (was 7, which dropped most of a 100+ event run).
|
||||||
|
private const val SNAPSHOT_EVENT_LIMIT = 200
|
||||||
|
|
||||||
class SessionEventBridge(
|
class SessionEventBridge(
|
||||||
private val eventStore: EventStore,
|
private val eventStore: EventStore,
|
||||||
private val artifactStore: ArtifactStore,
|
private val artifactStore: ArtifactStore,
|
||||||
@@ -53,6 +58,16 @@ class SessionEventBridge(
|
|||||||
) {
|
) {
|
||||||
private val approvalProjector = ApprovalProjector(DefaultApprovalReducer())
|
private val approvalProjector = ApprovalProjector(DefaultApprovalReducer())
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Resolves the last stage's inference output from CAS so a reopened session restores its output
|
||||||
|
* panel. Returns empty if there's no completed inference or the artifact is unavailable.
|
||||||
|
*/
|
||||||
|
private suspend fun resolveLastOutput(events: List<StoredEvent>): String =
|
||||||
|
events.lastOrNull { it.payload is InferenceCompletedEvent }
|
||||||
|
?.let { (it.payload as InferenceCompletedEvent).responseArtifactId }
|
||||||
|
?.let { artifactStore.get(it)?.toString(Charsets.UTF_8) }
|
||||||
|
.orEmpty()
|
||||||
|
|
||||||
suspend fun replaySnapshot() {
|
suspend fun replaySnapshot() {
|
||||||
val lastGlobal = eventStore.lastGlobalSequence()
|
val lastGlobal = eventStore.lastGlobalSequence()
|
||||||
eventStore.allSessionIds().forEach { sessionId ->
|
eventStore.allSessionIds().forEach { sessionId ->
|
||||||
@@ -122,7 +137,10 @@ class SessionEventBridge(
|
|||||||
|
|
||||||
val recentEvents = events
|
val recentEvents = events
|
||||||
.mapNotNull { eventToEntry(it) }
|
.mapNotNull { eventToEntry(it) }
|
||||||
.takeLast(7)
|
.takeLast(SNAPSHOT_EVENT_LIMIT)
|
||||||
|
|
||||||
|
// Restore the output panel on reopen (otherwise the snapshot carries no output).
|
||||||
|
val lastOutput = resolveLastOutput(events)
|
||||||
|
|
||||||
// Re-register pending approvals so the ApprovalCoordinator can route responses
|
// Re-register pending approvals so the ApprovalCoordinator can route responses
|
||||||
// from clients that connected after the ApprovalRequestedEvent was emitted.
|
// from clients that connected after the ApprovalRequestedEvent was emitted.
|
||||||
@@ -143,6 +161,8 @@ class SessionEventBridge(
|
|||||||
pendingApprovals = pendingApprovals,
|
pendingApprovals = pendingApprovals,
|
||||||
tools = toolRecords,
|
tools = toolRecords,
|
||||||
recentEvents = recentEvents,
|
recentEvents = recentEvents,
|
||||||
|
lastOutput = lastOutput,
|
||||||
|
lastResponse = lastOutput,
|
||||||
lastSequence = lastGlobal,
|
lastSequence = lastGlobal,
|
||||||
lastSessionSequence = lastSession,
|
lastSessionSequence = lastSession,
|
||||||
))
|
))
|
||||||
|
|||||||
@@ -119,6 +119,10 @@ sealed interface ServerMessage {
|
|||||||
val pendingApprovals: List<ApprovalDto>,
|
val pendingApprovals: List<ApprovalDto>,
|
||||||
val tools: List<ToolRecordDto> = emptyList(),
|
val tools: List<ToolRecordDto> = emptyList(),
|
||||||
val recentEvents: List<EventEntryDto> = emptyList(),
|
val recentEvents: List<EventEntryDto> = emptyList(),
|
||||||
|
// Last stage's inference output, resolved from CAS, so a reopened session restores its
|
||||||
|
// output panel instead of showing blank until new activity arrives.
|
||||||
|
val lastOutput: String = "",
|
||||||
|
val lastResponse: String = "",
|
||||||
val lastSequence: Long,
|
val lastSequence: Long,
|
||||||
val lastSessionSequence: Long,
|
val lastSessionSequence: Long,
|
||||||
override val sequence: Long? = null,
|
override val sequence: Long? = null,
|
||||||
|
|||||||
@@ -394,6 +394,8 @@ func (m *Model) onSnapshot(msg protocol.ServerMessage) {
|
|||||||
if msg.State != nil && msg.State.CurrentStageID != nil {
|
if msg.State != nil && msg.State.CurrentStageID != nil {
|
||||||
sess.CurrentStage = *msg.State.CurrentStageID
|
sess.CurrentStage = *msg.State.CurrentStageID
|
||||||
}
|
}
|
||||||
|
sess.LastOutput = msg.LastOutput
|
||||||
|
sess.LastResponse = msg.LastResponse
|
||||||
for _, e := range msg.RecentEvents {
|
for _, e := range msg.RecentEvents {
|
||||||
sess.Events = append(sess.Events, EventEntry{formatTime(e.Timestamp), e.Type, e.Detail})
|
sess.Events = append(sess.Events, EventEntry{formatTime(e.Timestamp), e.Type, e.Detail})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -67,6 +67,8 @@ type ServerMessage struct {
|
|||||||
Reason string `json:"reason"`
|
Reason string `json:"reason"`
|
||||||
Summary string `json:"outputSummary"`
|
Summary string `json:"outputSummary"`
|
||||||
Response string `json:"responseText"`
|
Response string `json:"responseText"`
|
||||||
|
LastOutput string `json:"lastOutput"` // session_snapshot: last stage output (reopen)
|
||||||
|
LastResponse string `json:"lastResponse"` // session_snapshot: last stage response (reopen)
|
||||||
Content string `json:"content"`
|
Content string `json:"content"`
|
||||||
Diff *string `json:"diff"`
|
Diff *string `json:"diff"`
|
||||||
Tier string `json:"tier"`
|
Tier string `json:"tier"`
|
||||||
|
|||||||
Reference in New Issue
Block a user