coze-dev
diff --git a/‎backend/api/api.go‎
Lines changed: 6 additions & 0 deletions b/‎backend/api/api.go‎
Lines changed: 6 additions & 0 deletions
diff --git a/‎backend/api/handler/coze/loop/apis/handler.go‎
Lines changed: 8 additions & 0 deletions b/‎backend/api/handler/coze/loop/apis/handler.go‎
Lines changed: 8 additions & 0 deletions
diff --git a/‎backend/api/handler/coze/loop/apis/observability_task_service.go‎
Lines changed: 45 additions & 0 deletions b/‎backend/api/handler/coze/loop/apis/observability_task_service.go‎
Lines changed: 45 additions & 0 deletions
diff --git a/‎backend/api/handler/coze/loop/apis/observability_trace_service.go‎
Lines changed: 18 additions & 0 deletions b/‎backend/api/handler/coze/loop/apis/observability_trace_service.go‎
Lines changed: 18 additions & 0 deletions
diff --git a/‎backend/api/handler/coze/loop/apis/wire.go‎
Lines changed: 7 additions & 0 deletions b/‎backend/api/handler/coze/loop/apis/wire.go‎
Lines changed: 7 additions & 0 deletions
diff --git a/‎backend/api/handler/coze/loop/apis/wire_gen.go‎
Lines changed: 11 additions & 6 deletions b/‎backend/api/handler/coze/loop/apis/wire_gen.go‎
Lines changed: 11 additions & 6 deletions
diff --git a/‎backend/api/router/coze/loop/apis/coze.loop.apis.go‎
Lines changed: 18 additions & 0 deletions b/‎backend/api/router/coze/loop/apis/coze.loop.apis.go‎
Lines changed: 18 additions & 0 deletions
diff --git a/‎backend/api/router/coze/loop/apis/middleware.go‎
Lines changed: 60 additions & 0 deletions b/‎backend/api/router/coze/loop/apis/middleware.go‎
Lines changed: 60 additions & 0 deletions
diff --git a/‎backend/cmd/consumer.go‎
Lines changed: 2 additions & 1 deletion b/‎backend/cmd/consumer.go‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎backend/cmd/main.go‎
Lines changed: 2 additions & 4 deletions b/‎backend/cmd/main.go‎
Lines changed: 2 additions & 4 deletions
@@ -13,10 +13,13 @@ import (
 	"github.com/cloudwego/kitex/client/callopt"
 	"github.com/cloudwego/kitex/pkg/endpoint"
 	"github.com/cloudwego/kitex/pkg/kerrors"
+	"github.com/coze-dev/coze-loop/backend/kitex_gen/coze/loop/observability/task"
+	"github.com/coze-dev/coze-loop/backend/loop_gen/coze/loop/observability/lotask"
 
 	"github.com/coze-dev/coze-loop/backend/infra/i18n"
 	cachemw "github.com/coze-dev/coze-loop/backend/infra/middleware/ctxcache"
 	logmw "github.com/coze-dev/coze-loop/backend/infra/middleware/logs"
+	"github.com/coze-dev/coze-loop/backend/infra/middleware/session"
 	"github.com/coze-dev/coze-loop/backend/infra/middleware/validator"
 	"github.com/coze-dev/coze-loop/backend/kitex_gen/coze/loop/data/dataset"
 	"github.com/coze-dev/coze-loop/backend/kitex_gen/coze/loop/data/tag"
@@ -194,20 +197,24 @@ type ObservabilityHandler struct {
 	obapp.ITraceApplication
 	obapp.ITraceIngestionApplication
 	obapp.IObservabilityOpenAPIApplication
+	obapp.ITaskApplication
 }
 
 func NewObservabilityHandler(
 	traceApp obapp.ITraceApplication,
 	ingestApp obapp.ITraceIngestionApplication,
 	openAPIApp obapp.IObservabilityOpenAPIApplication,
+	taskApp obapp.ITaskApplication,
 ) *ObservabilityHandler {
 	h := &ObservabilityHandler{
 		ITraceApplication:                traceApp,
 		ITraceIngestionApplication:       ingestApp,
 		IObservabilityOpenAPIApplication: openAPIApp,
+		ITaskApplication:                 taskApp,
 	}
 	bindLocalCallClient(trace.TraceService(h), &observabilityClient, lotrace.NewLocalTraceService)
 	bindLocalCallClient(traceopenapi.OpenAPIService(h), &observabilityOpenAPIClient, looptraceopenapi.NewLocalOpenAPIService)
+	bindLocalCallClient(task.TaskService(h), &observabilityTaskClient, lotask.NewLocalTaskService)
 	return h
 }
 
@@ -224,6 +231,7 @@ func defaultKiteXMiddlewares() []endpoint.Middleware {
 	return []endpoint.Middleware{
 		logmw.LogTrafficMW,
 		validator.KiteXValidatorMW,
+		session.NewRequestSessionMW(),
 		cachemw.CtxCacheMW,
 	}
 }
 
@@ -10,6 +10,8 @@ import (
 	"context"
 
 	"github.com/cloudwego/kitex/pkg/endpoint"
+	"github.com/coze-dev/coze-loop/backend/kitex_gen/coze/loop/evaluation/experimentservice"
+	task_processor "github.com/coze-dev/coze-loop/backend/modules/observability/domain/task/service/taskexe/processor"
 	"github.com/google/wire"
 
 	"github.com/coze-dev/coze-loop/backend/infra/ck"
@@ -92,6 +94,7 @@ var (
 		obapp.InitTraceApplication,
 		obapp.InitTraceIngestionApplication,
 		obapp.InitOpenAPIApplication,
+		obapp.InitTaskApplication,
 	)
 )
 
@@ -207,6 +210,10 @@ func InitObservabilityHandler(
 	tagClient tagservice.Client,
 	limiterFactory limiter.IRateLimiterFactory,
 	datasetClient datasetservice.Client,
+	redis redis.Cmdable,
+	experimentClient experimentservice.Client,
+	taskProcessor task_processor.TaskProcessor,
+	aid int32,
 ) (*ObservabilityHandler, error) {
 	wire.Build(
 		observabilitySet,
 
@@ -19,6 +19,7 @@ func MustInitConsumerWorkers(
 	experimentApplication exptapp.IExperimentApplication,
 	datasetApplication dataapp.IJobRunMsgHandler,
 	obApplication obapp.IObservabilityOpenAPIApplication,
+	taskApplication obapp.ITaskApplication,
 ) []mq.IConsumerWorker {
 	var res []mq.IConsumerWorker
 
@@ -38,7 +39,7 @@ func MustInitConsumerWorkers(
 	if err != nil {
 		panic(err)
 	}
-	workers, err = obconsumer.NewConsumerWorkers(loader, obApplication)
+	workers, err = obconsumer.NewConsumerWorkers(loader, obApplication, taskApplication)
 	if err != nil {
 		panic(err)
 	}
 
@@ -60,10 +60,8 @@ func main() {
 	if err := initTracer(handler); err != nil {
 		panic(err)
 	}
-
-	consumerWorkers := MustInitConsumerWorkers(c.cfgFactory, handler, handler, handler)
-	consumerRegistry := registry.NewConsumerRegistry(c.mqFactory).Register(consumerWorkers)
-	if err := consumerRegistry.StartAll(ctx); err != nil {
+	consumerWorkers := MustInitConsumerWorkers(c.cfgFactory, handler, handler, handler, handler)
+	if err := registry.NewConsumerRegistry(c.mqFactory).Register(consumerWorkers).StartAll(ctx); err != nil {
 		panic(err)
 	}
Original file line number	Diff line number	Diff line change
`@@ -60,10 +60,8 @@ func main() {`
`60`	`60`	`if err := initTracer(handler); err != nil {`
`61`	`61`	`panic(err)`
`62`	`62`	`}`
`63`		`-`
`64`		`- consumerWorkers := MustInitConsumerWorkers(c.cfgFactory, handler, handler, handler)`
`65`		`- consumerRegistry := registry.NewConsumerRegistry(c.mqFactory).Register(consumerWorkers)`
`66`		`- if err := consumerRegistry.StartAll(ctx); err != nil {`
	`63`	`+ consumerWorkers := MustInitConsumerWorkers(c.cfgFactory, handler, handler, handler, handler)`
	`64`	`+ if err := registry.NewConsumerRegistry(c.mqFactory).Register(consumerWorkers).StartAll(ctx); err != nil {`
`67`	`65`	`panic(err)`
`68`	`66`	`}`
`69`	`67`