refactor: remove legacy multi-agent-mux skills and infrastructure, and add Mattermost notification script and collaboration documentation

2026-07-12 23:20:48 +09:00
parent d885aa52ae
commit 9e4f286b43
40 changed files with 634 additions and 7221 deletions
@@ -776,6 +776,26 @@ Envoy Circuit Breaker 설정:

 T2 게이트웨이 자체 Circuit Breaker는 T3 디바이스 클래스별로 독립 인스턴스를 운용하여, 특정 디바이스 클래스의 대량 실패가 다른 클래스에 영향주지 않도록 격리한다.

+### 7.5 제어 멱등성 및 중복 제어 방지 (Control Idempotency)
+
+비동기 에이전트의 재시도 루프 시 네트워크 단절 또는 지연으로 인해 동일한 제어 명령이 중복 전달되는 안전 위험을 예방하기 위해, 제어 멱등성 보장 메커니즘을 내장한다.
+
+#### 1) 명령 성격 분류 (Command Classification)
+- **멱등성(Idempotent) 명령**: 동일한 인자값으로 여러 번 실행해도 물리 및 논리 상태가 일치하는 안전 명령. (예: `TelemetryService.GetLatestData()`, `DeviceService.SetState(status=SLEEP)`)
+- **비멱등성(Non-idempotent) 명령**: 실행 시마다 기기의 동작이나 화학적/물리적 상태가 누적 가산되어 중복 실행 시 위험을 초래하는 위험 명령. (예: `SprinklerService.SprayPesticide(volume=500ml)`, `PowerService.ToggleSwitch()`)
+
+#### 2) Control Intent Key (CIK) 스키마
+- 비멱등 명령 호출 시 클라이언트(T1 Orchestrator 등)는 헤더(`grpc-metadata-control-intent-key`) 및 페이로드 메타데이터 내에 UUIDv4 기반의 `Control Intent Key (CIK)`를 필수로 동반해야 한다.
+- 네트워크 문제로 RPC가 실패하여 재시도할 때, 호출자는 매번 새로운 `Job ID`를 발급하더라도 최초 생성한 `CIK`를 고정하여 전송한다.
+
+#### 3) T2 Gateway-side Pre-flight Check & Deduplication Cache
+- T2 게이트웨이의 gRPC 인터셉터(`IdempotencyInterceptor`)는 비멱등 명령에 대해 CIK를 캐시(`LRU Cache` 및 Redis/SQLite 등의 로컬 영속 DB 조합)와 대조한다.
+- **캐시 미스 (최초 요청)**: CIK를 'RUNNING' 상태로 캐시에 등록하고, 하부 T3 디바이스로 명령을 라우팅한 뒤 그 결과를 수신하여 'COMPLETED' 상태와 실행 결과 페이로드(ResultPayload)를 캐시에 저장하고 클라이언트에 응답한다.
+- **캐시 히트 (중복 요청)**:
+  - 상태가 'RUNNING'일 경우: 진행 중인 작업으로 인지하여 클라이언트에 `codes.Aborted` ("Operation in progress")를 반환하거나 대기 스트리밍 상태를 유지한다.
+  - 상태가 'COMPLETED'일 경우: 하부 T3 디바이스에 명령을 재전송하지 않고, 캐시된 `ResultPayload` 및 상태를 즉시 반환하여 중복 기기 작동을 방지한다.
+- **캐시 만료**: CIK의 유효 기간(TTL)은 각 사용 사례의 안전 마진에 따라 동적으로 지정된다(예: 스마트 팜 자동 방제 시나리오의 경우 최소 1시간 보존).
+
 ---

 ## 8. 멀티테넌시 설계
@@ -195,6 +195,7 @@ message AgentTask {
  bytes  payload        = 5;  // Protobuf Any 직렬화
  map<string, string> metadata = 6;  // X-Tenant-ID, X-Device-Class 등
  google.protobuf.Timestamp deadline = 7;
+  string control_intent_key = 8;     // 비멱등 제어 명령의 중복 필터링을 위한 CIK
 }

 // 에이전트 태스크 응답
@@ -422,14 +423,16 @@ func New(ctx context.Context, cfg GatewayConfig) (*Gateway, error) {
        return nil, fmt.Errorf("SPIFFE credentials: %w", err)
    }

-    // 2. Resume Token 관리자 초기화
+    // 2. Resume Token 및 Idempotency 관리자 초기화
    tokenMgr := NewResumeTokenManager()
+    idempotencyMgr := NewIdempotencyManager()

    // 3. gRPC 서버 인터셉터 체인 구성
    srv := grpc.NewServer(
        grpc.Creds(creds),
        grpc.ChainUnaryInterceptor(
            interceptors.DeadlineEnforcer(defaultDeadlines),
+            interceptors.IdempotencyUnary(idempotencyMgr),
            interceptors.ResumeTokenUnary(tokenMgr),
            interceptors.OTelUnary(),
        ),
@@ -1086,6 +1089,154 @@ type resumeServerStream struct {
 func (s *resumeServerStream) Context() context.Context { return s.ctx }
 ```

+#### Idempotency gRPC Interceptor (`internal/gateway/interceptors/idempotency.go`)
+
+```go
+package interceptors
+
+import (
+    "context"
+    "sync"
+    "time"
+
+    "google.golang.org/grpc"
+    "google.golang.org/grpc/codes"
+    "google.golang.org/grpc/metadata"
+    "google.golang.org/grpc/status"
+
+    agentv1 "github.com/your-org/edge-aiot-mas/gen/go/agent/v1"
+)
+
+const controlIntentKeyHeader = "grpc-metadata-control-intent-key"
+
+// CacheEntry는 CIK 캐시의 레코드를 정의한다.
+type CacheEntry struct {
+    State         string       // "RUNNING", "COMPLETED"
+    ResultPayload interface{}  // 캐시된 결과 페이로드 (AgentResult)
+    CreatedAt     time.Time
+}
+
+// IdempotencyManager는 CIK 기반 중복 제거 필터를 총괄한다.
+type IdempotencyManager struct {
+    mu    sync.RWMutex
+    cache map[string]*CacheEntry
+    ttl   time.Duration
+}
+
+// NewIdempotencyManager는 IdempotencyManager를 초기화한다.
+func NewIdempotencyManager() *IdempotencyManager {
+    mgr := &IdempotencyManager{
+        cache: make(map[string]*CacheEntry),
+        ttl:   1 * time.Hour, // 기본 TTL 1시간
+    }
+    // 백그라운드에서 캐시 만료 정리 고루틴 기동
+    go mgr.cleanupLoop()
+    return mgr
+}
+
+func (m *IdempotencyManager) cleanupLoop() {
+    ticker := time.NewTicker(10 * time.Minute)
+    for range ticker.C {
+        m.mu.Lock()
+        now := time.Now()
+        for k, v := range m.cache {
+            if now.Sub(v.CreatedAt) > m.ttl {
+                delete(m.cache, k)
+            }
+        }
+        m.mu.Unlock()
+    }
+}
+
+// Get은 CIK에 해당하는 캐시 데이터를 조회한다.
+func (m *IdempotencyManager) Get(cik string) (*CacheEntry, bool) {
+    m.mu.RLock()
+    defer m.mu.RUnlock()
+    entry, ok := m.cache[cik]
+    return entry, ok
+}
+
+// Set은 CIK에 대한 캐시 레코드를 등록하거나 갱신한다.
+func (m *IdempotencyManager) Set(cik string, entry *CacheEntry) {
+    m.mu.Lock()
+    defer m.mu.Unlock()
+    entry.CreatedAt = time.Now()
+    m.cache[cik] = entry
+}
+
+// IdempotencyUnary는 비멱등 Unary 명령의 중복 실행을 방지하는 인터셉터다.
+func IdempotencyUnary(mgr *IdempotencyManager) grpc.UnaryServerInterceptor {
+    return func(
+        ctx context.Context,
+        req interface{},
+        info *grpc.UnaryServerInfo,
+        handler grpc.UnaryHandler,
+    ) (interface{}, error) {
+        // 1. 요청 메시지가 AgentTask인지 타입 단언
+        task, ok := req.(*agentv1.AgentTask)
+        if !ok {
+            return handler(ctx, req)
+        }
+
+        // 2. 비멱등(Non-idempotent) 제어 명령 유형인지 검증 (infer, control, ota 등)
+        if task.TaskType != "control" && task.TaskType != "ota" {
+            return handler(ctx, req)
+        }
+
+        // 3. Control Intent Key (CIK) 추출
+        cik := task.ControlIntentKey
+        if cik == "" {
+            // 들어오는 메타데이터 헤더에서 추출 시도
+            if md, ok := metadata.FromIncomingContext(ctx); ok {
+                keys := md.Get(controlIntentKeyHeader)
+                if len(keys) > 0 {
+                    cik = keys[0]
+                }
+            }
+        }
+
+        // CIK가 비어 있으면 사전 검증을 통과시킬 수 없으므로 거부하거나 바이패스
+        if cik == "" {
+            return nil, status.Error(codes.InvalidArgument, "Control Intent Key (CIK) is required for non-idempotent tasks")
+        }
+
+        // 4. 캐시 조회 및 사전 검증(Pre-flight Check)
+        if entry, hit := mgr.Get(cik); hit {
+            switch entry.State {
+            case "RUNNING":
+                return nil, status.Error(codes.Aborted, "Operation is already in progress under this Control Intent Key")
+            case "COMPLETED":
+                // 중복 동작 방지: 캐시된 결과 즉시 반환
+                return entry.ResultPayload, nil
+            }
+        }
+
+        // 5. 캐시에 'RUNNING' 상태로 임시 선점
+        mgr.Set(cik, &CacheEntry{State: "RUNNING"})
+
+        // 6. 핸들러 실행 (하부 물리 계층 명령 전달)
+        resp, err := handler(ctx, req)
+        if err != nil {
+            // 실패 시 캐시 레코드 삭제하여 재시도 허용
+            m := mgr
+            m.mu.Lock()
+            delete(m.cache, cik)
+            m.mu.Unlock()
+            return nil, err
+        }
+
+        // 7. 성공 결과 캐싱 완료 처리
+        mgr.Set(cik, &CacheEntry{
+            State:         "COMPLETED",
+            ResultPayload: resp,
+        })
+
+        return resp, nil
+    }
+}
+```
+
+
 ### 4.4 A2A HTTP/3 엔드포인트

 #### `internal/gateway/a2a_handler.go`
@@ -0,0 +1,97 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "title": "AgentCard",
+  "description": "A2A 표준 기반 이종 에이전트 및 물리 장치 발견용 Agent Card 스키마 템플릿",
+  "type": "object",
+  "properties": {
+    "agent_id": {
+      "type": "string",
+      "description": "에이전트 또는 물리 노드의 고유 UUID 또는 식별자"
+    },
+    "name": {
+      "type": "string",
+      "description": "에이전트/장치 명칭 (예: smart-farm-sprinkler-01)"
+    },
+    "role": {
+      "type": "string",
+      "description": "시스템 내에서의 역할 (예: sensing-layer, control-layer, planner, developer, reviewer)"
+    },
+    "version": {
+      "type": "string",
+      "description": "에이전트 모듈 소프트웨어 버전"
+    },
+    "endpoint": {
+      "type": "object",
+      "description": "연동 엔드포인트 정보",
+      "properties": {
+        "uri": {
+          "type": "string",
+          "description": "접속 Endpoint URI (예: dns:///edge-gateway.local:50051 또는 mqtt://broker.local:1883)"
+        },
+        "protocol": {
+          "type": "string",
+          "enum": ["grpc", "grpc-over-quic", "mqtt-v5", "coap", "http2"],
+          "description": "통신 전송 프로토콜 규격"
+        },
+        "transport_fallback": {
+          "type": "string",
+          "enum": ["grpc-over-http2", "mqtt-v3.1.1", "none"],
+          "description": "네트워크 제한 시 fallback할 전송 계층"
+        }
+      },
+      "required": ["uri", "protocol"]
+    },
+    "security": {
+      "type": "object",
+      "description": "보안 및 신원 정보",
+      "properties": {
+        "auth_type": {
+          "type": "string",
+          "enum": ["spiffe-svid", "hmac-sha256", "mtls", "none"],
+          "description": "인증/인가 보안 프로토콜"
+        },
+        "spiffe_id": {
+          "type": "string",
+          "description": "SPIFFE 신원 식별자 (예: spiffe://example.org/ns/smartfarm/sa/sprinkler)"
+        },
+        "hmac_token_ref": {
+          "type": "string",
+          "description": "HMAC 토큰 서명을 위한 환경변수명"
+        }
+      },
+      "required": ["auth_type"]
+    },
+    "capabilities": {
+      "type": "array",
+      "description": "에이전트가 제공하는 과업/제어 기능 및 RPC 메소드 매핑",
+      "items": {
+        "type": "object",
+        "properties": {
+          "name": {
+            "type": "string",
+            "description": "기능/명령 이름 (예: spray_pesticide)"
+          },
+          "grpc_method": {
+            "type": "string",
+            "description": "gRPC 패키지/메서드 매핑 경로 (예: /smartfarm.ControlService/SprayPesticide)"
+          },
+          "mqtt_topic": {
+            "type": "string",
+            "description": "하부 센싱/제어 매핑 MQTT 토픽 (예: farm/device/sprinkler/control)"
+          },
+          "idempotency": {
+            "type": "string",
+            "enum": ["idempotent", "non-idempotent"],
+            "description": "제어 명령의 멱등성 여부 분류 (비멱등 시 CIK 적용 필수)"
+          },
+          "input_schema": {
+            "type": "object",
+            "description": "입력 파라미터 Protobuf/JSON 스키마 정의"
+          }
+        },
+        "required": ["name", "idempotency"]
+      }
+    }
+  },
+  "required": ["agent_id", "name", "role", "endpoint", "security", "capabilities"]
+}