feat: Add message observing status API

- Add separate message_observing_tracking table (decoupled from messages)
- Implement Go API endpoint: GET /session/{id}/observing-status
- Add Python SDK method: messages_observing_status()
- Add TypeScript SDK method: messagesObservingStatus()
- Returns counts of observed, in_process, pending messages

Backend (Go):
- Model layer: MessageObservingStatus, MessageObservingTracking
- Repository layer: GORM with LEFT JOIN query
- Service layer: Business logic wrapper
- Handler layer: HTTP endpoint with unit tests
- Router: URL registration
- Dependency injection: Full wiring

Python SDK:
- Type: MessageObservingStatus (Pydantic)
- Method: sessions.messages_observing_status()

TypeScript SDK:
- Schema: MessageObservingStatusSchema (Zod)
- Type: MessageObservingStatus
- Method: sessions.messagesObservingStatus()

Database:
- Migration: 002_create_message_observing_tracking.sql
- Table: message_observing_tracking with FK to messages

Tests:
- Handler tests with mocks (all passing)
- End-to-end testing pending migration

Author: slyt3

src/client/acontext-py/src/acontext/resources/sessions.py

@@ -15,6 +15,7 @@
     LearningStatus,
     ListSessionsOutput,
     Message,
+    MessageObservingStatus,
     Session,
     TokenCounts,
 )
@@ -348,3 +349,20 @@ def get_token_counts(self, session_id: str) -> TokenCounts:
         """
         data = self._requester.request("GET", f"/session/{session_id}/token_counts")
         return TokenCounts.model_validate(data)
+    def messages_observing_status(self, session_id: str) -> MessageObservingStatus:
+        """Get message observing status counts for a session.
+        
+        Returns the count of messages by their observing status:
+        observed, in_process, and pending.
+        
+        Args:
+            session_id: The UUID of the session.
+        
+        Returns:
+            MessageObservingStatus object containing observed, in_process, 
+            pending counts and updated_at timestamp.
+        """
+        data = self._requester.request(
+            "GET", f"/session/{session_id}/observing-status"
+        )
+        return MessageObservingStatus.model_validate(data)

src/client/acontext-py/src/acontext/types/session.py

@@ -249,3 +249,19 @@ class TokenCounts(BaseModel):
         ...,
         description="Total token count for all text and tool-call parts in a session",
     )
+
+class MessageObservingStatus(BaseModel):
+    """Response model for message observing status."""
+    
+    observed: int = Field(
+        ..., description="Number of messages with observed status"
+    )
+    in_process: int = Field(
+        ..., description="Number of messages with in_process status"
+    )
+    pending: int = Field(
+        ..., description="Number of messages with pending status"
+    )
+    updated_at: str = Field(
+        ..., description="Timestamp when the status was retrieved"
+    )

src/client/acontext-ts/package-lock.json

@@ -17,6 +17,8 @@ import {
   ListSessionsOutput,
   ListSessionsOutputSchema,
   Message,
+  MessageObservingStatus,
+  MessageObservingStatusSchema, 
   MessageSchema,
   Session,
   SessionSchema,
@@ -264,5 +266,20 @@ export class SessionsAPI {
     const data = await this.requester.request('GET', `/session/${sessionId}/token_counts`);
     return TokenCountsSchema.parse(data);
   }
+
+  /**
+   * Get message observing status counts for a session.
+   *
+   * Returns the count of messages by their observing status:
+   * observed, in_process, and pending.
+   *
+   * @param sessionId - The UUID of the session.
+   * @returns MessageObservingStatus object containing observed, in_process, 
+   *          pending counts and updated_at timestamp.
+   */
+  async messagesObservingStatus(sessionId: string): Promise<MessageObservingStatus> {
+    const data = await this.requester.request('GET', `/session/${sessionId}/observing-status`);
+    return MessageObservingStatusSchema.parse(data);
+  }
 }
 

src/client/acontext-ts/src/resources/sessions.ts

@@ -126,6 +126,15 @@ export const TokenCountsSchema = z.object({
 
 export type TokenCounts = z.infer<typeof TokenCountsSchema>;
 
+export const MessageObservingStatusSchema = z.object({
+  observed: z.number(),
+  in_process: z.number(),
+  pending: z.number(),
+  updated_at: z.string(),
+});
+
+export type MessageObservingStatus = z.infer<typeof MessageObservingStatusSchema>;
+
 /**
  * Parameters for the remove_tool_result edit strategy.
  */

src/client/acontext-ts/src/types/session.ts

@@ -90,18 +90,20 @@ func main() {
 	artifactHandler := do.MustInvoke[*handler.ArtifactHandler](inj)
 	taskHandler := do.MustInvoke[*handler.TaskHandler](inj)
 	toolHandler := do.MustInvoke[*handler.ToolHandler](inj)
+	messageObservingHandler := do.MustInvoke[*handler.MessageObservingHandler](inj)
 
 	engine := router.NewRouter(router.RouterDeps{
-		Config:          cfg,
-		DB:              db,
-		Log:             log,
-		SpaceHandler:    spaceHandler,
-		BlockHandler:    blockHandler,
-		SessionHandler:  sessionHandler,
-		DiskHandler:     diskHandler,
-		ArtifactHandler: artifactHandler,
-		TaskHandler:     taskHandler,
-		ToolHandler:     toolHandler,
+		Config:                  cfg,
+		DB:                      db,
+		Log:                     log,
+		SpaceHandler:            spaceHandler,
+		BlockHandler:            blockHandler,
+		SessionHandler:          sessionHandler,
+		DiskHandler:             diskHandler,
+		ArtifactHandler:         artifactHandler,
+		TaskHandler:             taskHandler,
+		ToolHandler:             toolHandler,
+		MessageObservingHandler: messageObservingHandler,
 	})
 
 	addr := fmt.Sprintf("%s:%d", cfg.App.Host, cfg.App.Port)

src/server/api/go/cmd/server/main.go

@@ -170,6 +170,10 @@ func BuildContainer() *do.Injector {
 		return repo.NewTaskRepo(do.MustInvoke[*gorm.DB](i)), nil
 	})
 
+	do.Provide(inj, func(i *do.Injector) (repo.MessageObservingRepo, error) {
+		return repo.NewMessageObservingRepo(do.MustInvoke[*gorm.DB](i)), nil
+	})
+
 	// Service
 	do.Provide(inj, func(i *do.Injector) (service.SpaceService, error) {
 		return service.NewSpaceService(
@@ -208,6 +212,11 @@ func BuildContainer() *do.Injector {
 			do.MustInvoke[*zap.Logger](i),
 		), nil
 	})
+	do.Provide(inj, func(i *do.Injector) (service.MessageObservingService, error) {
+		return service.NewMessageObservingService(
+			do.MustInvoke[repo.MessageObservingRepo](i),
+		), nil
+	})
 
 	// Handler
 	do.Provide(inj, func(i *do.Injector) (*handler.SpaceHandler, error) {
@@ -240,6 +249,11 @@ func BuildContainer() *do.Injector {
 	do.Provide(inj, func(i *do.Injector) (*handler.ToolHandler, error) {
 		return handler.NewToolHandler(do.MustInvoke[*httpclient.CoreClient](i)), nil
 	})
+	do.Provide(inj, func(i *do.Injector) (*handler.MessageObservingHandler, error) {
+		return handler.NewMessageObservingHandler(
+			do.MustInvoke[service.MessageObservingService](i),
+		), nil
+	})
 
 	return inj
 }

src/server/api/go/internal/bootstrap/container.go

@@ -0,0 +1,60 @@
+package handler
+
+import (
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/memodb-io/Acontext/internal/modules/service"
+)
+
+// MessageObservingHandler handles HTTP requests for message observing status
+type MessageObservingHandler struct {
+	svc service.MessageObservingService
+}
+
+// NewMessageObservingHandler creates a new message observing handler
+func NewMessageObservingHandler(svc service.MessageObservingService) *MessageObservingHandler {
+	if svc == nil {
+		panic("message observing service cannot be nil")
+	}
+	return &MessageObservingHandler{
+		svc: svc,
+	}
+}
+
+// GetSessionObservingStatus handles GET /api/v1/sessions/:session_id/observing-status
+//
+// @Summary Get message observing status for a session
+// @Description Returns the count of observed, in_process, and pending messages
+// @Tags sessions
+// @Accept json
+// @Produce json
+// @Param session_id path string true "Session ID" format(uuid)
+// @Success 200 {object} model.MessageObservingStatus
+// @Failure 400 {object} map[string]string
+// @Failure 500 {object} map[string]string
+// @Router /api/v1/sessions/{session_id}/observing-status [get]
+func (h *MessageObservingHandler) GetSessionObservingStatus(c *gin.Context) {
+	// Extract session_id from URL path
+	sessionID := c.Param("session_id")
+
+	// Validate session ID
+	if sessionID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{
+			"error": "session_id is required",
+		})
+		return
+	}
+
+	// Call service
+	status, err := h.svc.GetSessionObservingStatus(c.Request.Context(), sessionID)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{
+			"error": err.Error(),
+		})
+		return
+	}
+
+	// Return success response
+	c.JSON(http.StatusOK, status)
+}

src/server/api/go/internal/modules/handler/message_observing_handler.go

@@ -0,0 +1,131 @@
+package handler
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/memodb-io/Acontext/internal/modules/model"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+// MockMessageObservingService is a mock implementation of MessageObservingService
+type MockMessageObservingService struct {
+	mock.Mock
+}
+
+func (m *MockMessageObservingService) GetSessionObservingStatus(ctx context.Context, sessionID string) (*model.MessageObservingStatus, error) {
+	args := m.Called(ctx, sessionID)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*model.MessageObservingStatus), args.Error(1)
+}
+
+func TestMessageObservingHandler_GetSessionObservingStatus_Success(t *testing.T) {
+	// Setup
+	gin.SetMode(gin.TestMode)
+
+	mockService := new(MockMessageObservingService)
+	handler := NewMessageObservingHandler(mockService)
+
+	sessionID := "550e8400-e29b-41d4-a716-446655440000"
+	expectedStatus := &model.MessageObservingStatus{
+		Observed:  10,
+		InProcess: 5,
+		Pending:   3,
+		UpdatedAt: time.Now(),
+	}
+
+	mockService.On("GetSessionObservingStatus", mock.Anything, sessionID).
+		Return(expectedStatus, nil)
+
+	// Create test request
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Params = gin.Params{
+		{Key: "session_id", Value: sessionID},
+	}
+	req, _ := http.NewRequest("GET", "/session/"+sessionID+"/observing-status", nil)
+	c.Request = req
+
+	// Execute
+	handler.GetSessionObservingStatus(c)
+
+	// Assert
+	assert.Equal(t, http.StatusOK, w.Code)
+	mockService.AssertExpectations(t)
+
+	// Check response body contains expected values
+	assert.Contains(t, w.Body.String(), `"observed":10`)
+	assert.Contains(t, w.Body.String(), `"in_process":5`)
+	assert.Contains(t, w.Body.String(), `"pending":3`)
+}
+
+func TestMessageObservingHandler_GetSessionObservingStatus_EmptySessionID(t *testing.T) {
+	// Setup
+	gin.SetMode(gin.TestMode)
+
+	mockService := new(MockMessageObservingService)
+	handler := NewMessageObservingHandler(mockService)
+
+	// Create test request with empty session_id
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Params = gin.Params{
+		{Key: "session_id", Value: ""},
+	}
+	req, _ := http.NewRequest("GET", "/session//observing-status", nil)
+	c.Request = req
+
+	// Execute
+	handler.GetSessionObservingStatus(c)
+
+	// Assert
+	assert.Equal(t, http.StatusBadRequest, w.Code)
+	assert.Contains(t, w.Body.String(), "session_id is required")
+	mockService.AssertNotCalled(t, "GetSessionObservingStatus")
+}
+
+func TestMessageObservingHandler_GetSessionObservingStatus_ServiceError(t *testing.T) {
+	// Setup
+	gin.SetMode(gin.TestMode)
+
+	mockService := new(MockMessageObservingService)
+	handler := NewMessageObservingHandler(mockService)
+
+	sessionID := "550e8400-e29b-41d4-a716-446655440000"
+	expectedError := errors.New("database connection failed")
+
+	mockService.On("GetSessionObservingStatus", mock.Anything, sessionID).
+		Return(nil, expectedError)
+
+	// Create test request
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Params = gin.Params{
+		{Key: "session_id", Value: sessionID},
+	}
+	req, _ := http.NewRequest("GET", "/session/"+sessionID+"/observing-status", nil)
+	c.Request = req
+
+	// Execute
+	handler.GetSessionObservingStatus(c)
+
+	// Assert
+	assert.Equal(t, http.StatusInternalServerError, w.Code)
+	assert.Contains(t, w.Body.String(), "database connection failed")
+	mockService.AssertExpectations(t)
+}
+
+func TestNewMessageObservingHandler_NilService(t *testing.T) {
+	// Should panic when service is nil
+	assert.Panics(t, func() {
+		NewMessageObservingHandler(nil)
+	})
+}